big-data/hadoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

HDFS-9754. Avoid unnecessary getBlockCollection calls in BlockManager. Contributed by Jing Zhao.

Browse Source
This commit is contained in:
Jing Zhao 2016-02-12 11:07:52 -08:00
parent f3c91a41a5
commit 972782d956
11 changed files with 76 additions and 118 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
View File

@ -1002,6 +1002,9 @@ Release 2.9.0 - UNRELEASED
HDFS-9780. RollingFileSystemSink doesn't work on secure clusters.
(Daniel Templeton via kasha)
HDFS-9754. Avoid unnecessary getBlockCollection calls in BlockManager.
(jing9)
OPTIMIZATIONS
BUG FIXES

10
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockInfo.java
View File

@ -96,6 +96,10 @@ public void setBlockCollectionId(long id) {
this.bcId = id;
}
public void delete() {
setBlockCollectionId(INVALID_INODE_ID);
}
public boolean isDeleted() {
return bcId == INVALID_INODE_ID;
}
@ -245,6 +249,12 @@ public boolean isComplete() {
return getBlockUCState().equals(BlockUCState.COMPLETE);
}
public final boolean isCompleteOrCommitted() {
final BlockUCState state = getBlockUCState();
return state.equals(BlockUCState.COMPLETE) ||
state.equals(BlockUCState.COMMITTED);
}
/**
* Add/Update the under construction feature.
*/

86
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java
View File

@ -743,7 +743,7 @@ public boolean commitOrCompleteLastBlock(BlockCollection bc,
}
if (hasMinStorage(lastBlock)) {
if (committed) {
addExpectedReplicasToPending(lastBlock, bc);
addExpectedReplicasToPending(lastBlock);
}
completeBlock(lastBlock, false);
}
@ -755,26 +755,21 @@ public boolean commitOrCompleteLastBlock(BlockCollection bc,
* pendingReplications in order to keep ReplicationMonitor from scheduling
* the block.
*/
public void addExpectedReplicasToPending(BlockInfo blk, BlockCollection bc) {
if (!bc.isStriped()) {
addExpectedReplicasToPending(blk);
}
}
private void addExpectedReplicasToPending(BlockInfo lastBlock) {
DatanodeStorageInfo[] expectedStorages =
lastBlock.getUnderConstructionFeature().getExpectedStorageLocations();
if (expectedStorages.length - lastBlock.numNodes() > 0) {
ArrayList<DatanodeDescriptor> pendingNodes =
new ArrayList<DatanodeDescriptor>();
for (DatanodeStorageInfo storage : expectedStorages) {
DatanodeDescriptor dnd = storage.getDatanodeDescriptor();
if (lastBlock.findStorageInfo(dnd) == null) {
pendingNodes.add(dnd);
public void addExpectedReplicasToPending(BlockInfo blk) {
if (!blk.isStriped()) {
DatanodeStorageInfo[] expectedStorages =
blk.getUnderConstructionFeature().getExpectedStorageLocations();
if (expectedStorages.length - blk.numNodes() > 0) {
ArrayList<DatanodeDescriptor> pendingNodes = new ArrayList<>();
for (DatanodeStorageInfo storage : expectedStorages) {
DatanodeDescriptor dnd = storage.getDatanodeDescriptor();
if (blk.findStorageInfo(dnd) == null) {
pendingNodes.add(dnd);
}
}
pendingReplications.increment(blk,
pendingNodes.toArray(new DatanodeDescriptor[pendingNodes.size()]));
}
pendingReplications.increment(lastBlock,
pendingNodes.toArray(new DatanodeDescriptor[pendingNodes.size()]));
}
}
@ -962,13 +957,13 @@ private LocatedBlock createLocatedBlock(final BlockInfo blk, final long pos)
final BlockUnderConstructionFeature uc = blk.getUnderConstructionFeature();
if (blk.isStriped()) {
final DatanodeStorageInfo[] storages = uc.getExpectedStorageLocations();
final ExtendedBlock eb = new ExtendedBlock(namesystem.getBlockPoolId(),
final ExtendedBlock eb = new ExtendedBlock(getBlockPoolId(),
blk);
return newLocatedStripedBlock(eb, storages, uc.getBlockIndices(), pos,
false);
} else {
final DatanodeStorageInfo[] storages = uc.getExpectedStorageLocations();
final ExtendedBlock eb = new ExtendedBlock(namesystem.getBlockPoolId(),
final ExtendedBlock eb = new ExtendedBlock(getBlockPoolId(),
blk);
return newLocatedBlock(eb, storages, pos, false);
}
@ -1011,7 +1006,7 @@ private LocatedBlock createLocatedBlock(final BlockInfo blk, final long pos)
" numNodes: " + numNodes +
" numCorrupt: " + numCorruptNodes +
" numCorruptRepls: " + numCorruptReplicas;
final ExtendedBlock eb = new ExtendedBlock(namesystem.getBlockPoolId(), blk);
final ExtendedBlock eb = new ExtendedBlock(getBlockPoolId(), blk);
return blockIndices == null ?
newLocatedBlock(eb, machines, pos, isCorrupt) :
newLocatedStripedBlock(eb, machines, blockIndices, pos, isCorrupt);
@ -1578,11 +1573,8 @@ boolean hasEnoughEffectiveReplicas(BlockInfo block,
private BlockReconstructionWork scheduleReconstruction(BlockInfo block,
int priority) {
// block should belong to a file
BlockCollection bc = getBlockCollection(block);
// abandoned block or block reopened for append
if (bc == null
|| (bc.isUnderConstruction() && block.equals(bc.getLastBlock()))) {
// skip abandoned block or block reopened for append
if (block.isDeleted() || !block.isCompleteOrCommitted()) {
// remove from neededReplications
neededReplications.remove(block, priority);
return null;
@ -1626,6 +1618,7 @@ private BlockReconstructionWork scheduleReconstruction(BlockInfo block,
additionalReplRequired = 1; // Needed on a new rack
}
final BlockCollection bc = getBlockCollection(block);
if (block.isStriped()) {
if (pendingNum > 0) {
// Wait the previous reconstruction to finish.
@ -1649,11 +1642,8 @@ private boolean validateReconstructionWork(BlockReconstructionWork rw) {
BlockInfo block = rw.getBlock();
int priority = rw.getPriority();
// Recheck since global lock was released
// block should belong to a file
BlockCollection bc = getBlockCollection(block);
// abandoned block or block reopened for append
if (bc == null
|| (bc.isUnderConstruction() && block.equals(bc.getLastBlock()))) {
// skip abandoned block or block reopened for append
if (block.isDeleted() || !block.isCompleteOrCommitted()) {
neededReplications.remove(block, priority);
rw.resetTargets();
return false;
@ -1688,23 +1678,12 @@ private boolean validateReconstructionWork(BlockReconstructionWork rw) {
assert rw.getTargets().length > 0;
assert pendingNum == 0 : "Should wait the previous reconstruction"
+ " to finish";
String src = getBlockCollection(block).getName();
ErasureCodingPolicy ecPolicy = null;
try {
ecPolicy = namesystem.getErasureCodingPolicyForPath(src);
} catch (IOException e) {
blockLog
.warn("Failed to get EC policy for the file {} ", src);
}
if (ecPolicy == null) {
blockLog.warn("No erasure coding policy found for the file {}. "
+ "So cannot proceed for reconstruction", src);
// TODO: we may have to revisit later for what we can do better to
// handle this case.
return false;
}
final ErasureCodingPolicy ecPolicy =
((BlockInfoStriped) block).getErasureCodingPolicy();
assert ecPolicy != null;
rw.getTargets()[0].getDatanodeDescriptor().addBlockToBeErasureCoded(
new ExtendedBlock(namesystem.getBlockPoolId(), block),
new ExtendedBlock(getBlockPoolId(), block),
rw.getSrcNodes(), rw.getTargets(),
((ErasureCodingWork) rw).getLiveBlockIndicies(), ecPolicy);
} else {
@ -2870,8 +2849,6 @@ private Block addStoredBlock(final BlockInfo block,
// it will happen in next block report otherwise.
return block;
}
BlockCollection bc = getBlockCollection(storedBlock);
assert bc != null : "Block must belong to a file";
// add block to the datanode
AddBlockResult result = storageInfo.addBlock(storedBlock, reportedBlock);
@ -2907,7 +2884,7 @@ private Block addStoredBlock(final BlockInfo block,
if(storedBlock.getBlockUCState() == BlockUCState.COMMITTED &&
hasMinStorage(storedBlock, numLiveReplicas)) {
addExpectedReplicasToPending(storedBlock, bc);
addExpectedReplicasToPending(storedBlock);
completeBlock(storedBlock, false);
} else if (storedBlock.isComplete() && result == AddBlockResult.ADDED) {
// check whether safe replication is reached for the block
@ -2918,8 +2895,8 @@ private Block addStoredBlock(final BlockInfo block,
bmSafeMode.incrementSafeBlockCount(numCurrentReplica, storedBlock);
}
// if file is under construction, then done for now
if (bc.isUnderConstruction()) {
// if block is still under construction, then done for now
if (!storedBlock.isCompleteOrCommitted()) {
return storedBlock;
}
@ -3444,8 +3421,7 @@ public void removeStoredBlock(BlockInfo storedBlock, DatanodeDescriptor node) {
// necessary. In that case, put block on a possibly-will-
// be-replicated list.
//
BlockCollection bc = getBlockCollection(storedBlock);
if (bc != null) {
if (!storedBlock.isDeleted()) {
bmSafeMode.decrementSafeBlockCount(storedBlock);
updateNeededReplications(storedBlock, -1, 0);
}

2
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlocksMap.java
View File

@ -91,7 +91,7 @@ void removeBlock(Block block) {
if (blockInfo == null)
return;
blockInfo.setBlockCollectionId(INodeId.INVALID_INODE_ID);
assert blockInfo.getBlockCollectionId() == INodeId.INVALID_INODE_ID;
final int size = blockInfo.isStriped() ?
blockInfo.getCapacity() : blockInfo.numNodes();
for(int idx = size - 1; idx >= 0; idx--) {

8
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirTruncateOp.java
View File

@ -18,6 +18,7 @@
package org.apache.hadoop.hdfs.server.namenode;
import java.io.IOException;
import java.util.Set;
import org.apache.hadoop.HadoopIllegalArgumentException;
import org.apache.hadoop.fs.UnresolvedLinkException;
@ -186,6 +187,7 @@ static void unprotectedTruncate(final FSNamesystem fsn, final String src,
"Should be the same block.";
if (oldBlock.getBlockId() != tBlk.getBlockId()
&& !file.isBlockInLatestSnapshot(oldBlock)) {
oldBlock.delete();
fsd.getBlockManager().removeBlockFromMap(oldBlock);
}
}
@ -298,9 +300,9 @@ private static boolean unprotectedTruncate(FSNamesystem fsn,
verifyQuotaForTruncate(fsn, iip, file, newLength, delta);
long remainingLength =
file.collectBlocksBeyondMax(newLength, collectedBlocks);
file.excludeSnapshotBlocks(latestSnapshot, collectedBlocks);
Set<BlockInfo> toRetain = file.getSnapshotBlocksToRetain(latestSnapshot);
long remainingLength = file.collectBlocksBeyondMax(newLength,
collectedBlocks, toRetain);
file.setModificationTime(mtime);
// return whether on a block boundary
return (remainingLength - newLength) == 0;

14
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
View File

@ -3199,7 +3199,7 @@ void addCommittedBlocksToPending(final INodeFile pendingFile) {
final BlockInfo b = blocks[i];
if (b != null && b.getBlockUCState() == BlockUCState.COMMITTED) {
// b is COMMITTED but not yet COMPLETE, add it to pending replication.
blockManager.addExpectedReplicasToPending(b, pendingFile);
blockManager.addExpectedReplicasToPending(b);
}
}
}
@ -4308,9 +4308,7 @@ PermissionStatus createFsOwnerPermissions(FsPermission permission) {
return new PermissionStatus(fsOwner.getShortUserName(), supergroup, permission);
}
@Override
public void checkSuperuserPrivilege()
throws AccessControlException {
void checkSuperuserPrivilege() throws AccessControlException {
if (isPermissionEnabled) {
FSPermissionChecker pc = getPermissionChecker();
pc.checkSuperuserPrivilege();
@ -6573,7 +6571,7 @@ ErasureCodingPolicy getErasureCodingPolicy(String src)
readLock();
try {
checkOperation(OperationCategory.READ);
return getErasureCodingPolicyForPath(src);
return FSDirErasureCodingOp.getErasureCodingPolicy(this, src);
} finally {
readUnlock();
}
@ -6837,12 +6835,6 @@ public String getTotalSyncTimes() {
}
}
@Override
public ErasureCodingPolicy getErasureCodingPolicyForPath(String src)
throws IOException {
return FSDirErasureCodingOp.getErasureCodingPolicy(this, src);
}
/**
* Gets number of bytes in the blocks in future generation stamps.
*

6
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INode.java
View File

@ -1011,14 +1011,10 @@ public List<UpdatedReplicationInfo> toUpdateReplicationInfo() {
*/
public void addDeleteBlock(BlockInfo toDelete) {
assert toDelete != null : "toDelete is null";
toDelete.delete();
toDeleteList.add(toDelete);
}
public void removeDeleteBlock(BlockInfo block) {
assert block != null : "block is null";
toDeleteList.remove(block);
}
public void addUpdateReplicationFactor(BlockInfo block, short targetRepl) {
toUpdateReplicationInfo.add(
new UpdatedReplicationInfo(targetRepl, block));

38
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodeFile.java
View File

@ -25,6 +25,7 @@
import java.io.IOException;
import java.io.PrintWriter;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@ -314,12 +315,13 @@ BlockInfo removeLastBlock(Block oldblock) {
return null;
}
BlockInfo ucBlock = blocks[size_1];
BlockInfo lastBlock = blocks[size_1];
//copy to a new list
BlockInfo[] newlist = new BlockInfo[size_1];
System.arraycopy(blocks, 0, newlist, 0, size_1);
setBlocks(newlist);
return ucBlock;
lastBlock.delete();
return lastBlock;
}
/* End of Under-Construction Feature */
@ -629,7 +631,6 @@ public void clearFile(ReclaimContext reclaimContext) {
if (blocks != null && reclaimContext.collectedBlocks != null) {
for (BlockInfo blk : blocks) {
reclaimContext.collectedBlocks.addDeleteBlock(blk);
blk.setBlockCollectionId(INodeId.INVALID_INODE_ID);
}
}
clearBlocks();
@ -905,7 +906,7 @@ public void dumpTreeRecursively(PrintWriter out, StringBuilder prefix,
* @return sum of sizes of the remained blocks
*/
public long collectBlocksBeyondMax(final long max,
final BlocksMapUpdateInfo collectedBlocks) {
final BlocksMapUpdateInfo collectedBlocks, Set<BlockInfo> toRetain) {
final BlockInfo[] oldBlocks = getBlocks();
if (oldBlocks == null) {
return 0;
@ -927,7 +928,10 @@ public long collectBlocksBeyondMax(final long max,
// collect the blocks beyond max
if (collectedBlocks != null) {
for(; n < oldBlocks.length; n++) {
collectedBlocks.addDeleteBlock(oldBlocks[n]);
final BlockInfo del = oldBlocks[n];
if (toRetain == null || !toRetain.contains(del)) {
collectedBlocks.addDeleteBlock(del);
}
}
}
return size;
@ -1026,22 +1030,18 @@ public void collectBlocksBeyondSnapshot(BlockInfo[] snapshotBlocks,
}
/** Exclude blocks collected for deletion that belong to a snapshot. */
void excludeSnapshotBlocks(int snapshotId,
BlocksMapUpdateInfo collectedBlocks) {
if(collectedBlocks == null || collectedBlocks.getToDeleteList().isEmpty())
return;
Set<BlockInfo> getSnapshotBlocksToRetain(int snapshotId) {
FileWithSnapshotFeature sf = getFileWithSnapshotFeature();
if(sf == null)
return;
BlockInfo[] snapshotBlocks =
getDiffs().findEarlierSnapshotBlocks(snapshotId);
if(snapshotBlocks == null)
return;
List<BlockInfo> toDelete = collectedBlocks.getToDeleteList();
for(BlockInfo blk : snapshotBlocks) {
if(toDelete.contains(blk))
collectedBlocks.removeDeleteBlock(blk);
if(sf == null) {
return null;
}
BlockInfo[] snapshotBlocks = getDiffs().findEarlierSnapshotBlocks(snapshotId);
if(snapshotBlocks == null) {
return null;
}
Set<BlockInfo> toRetain = new HashSet<>(snapshotBlocks.length);
Collections.addAll(toRetain, snapshotBlocks);
return toRetain;
}
/**

23
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/Namesystem.java
View File

@ -17,17 +17,10 @@
*/
package org.apache.hadoop.hdfs.server.namenode;
import java.io.IOException;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.hdfs.protocol.Block;
import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockCollection;
import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory;
import org.apache.hadoop.hdfs.server.namenode.ha.HAContext;
import org.apache.hadoop.hdfs.util.RwLock;
import org.apache.hadoop.ipc.StandbyException;
import org.apache.hadoop.security.AccessControlException;
/** Namesystem operations. */
@InterfaceAudience.Private
@ -35,26 +28,10 @@ public interface Namesystem extends RwLock, SafeMode {
/** Is this name system running? */
boolean isRunning();
/** Check if the user has superuser privilege. */
void checkSuperuserPrivilege() throws AccessControlException;
/** @return the block pool ID */
String getBlockPoolId();
BlockCollection getBlockCollection(long id);
void startSecretManagerIfNecessary();
/**
* Gets the erasure coding policy for the path
* @param src
* - path
* @return {@link ErasureCodingPolicy}
* @throws IOException
*/
ErasureCodingPolicy getErasureCodingPolicyForPath(String src)
throws IOException;
boolean isInSnapshot(long blockCollectionID);
CacheManager getCacheManager();

2
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/FileWithSnapshotFeature.java
View File

@ -220,7 +220,7 @@ public void collectBlocksAndClear(
FileDiff last = diffs.getLast();
BlockInfo[] snapshotBlocks = last == null ? null : last.getBlocks();
if(snapshotBlocks == null)
file.collectBlocksBeyondMax(max, reclaimContext.collectedBlocks());
file.collectBlocksBeyondMax(max, reclaimContext.collectedBlocks(), null);
else
file.collectBlocksBeyondSnapshot(snapshotBlocks,
reclaimContext.collectedBlocks());

2
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestNameNodeMetadataConsistency.java
View File

@ -95,6 +95,7 @@ public void testGenerationStampInFuture() throws
cluster.getNameNode().getNamesystem().writeLock();
BlockInfo bInfo = cluster.getNameNode().getNamesystem().getBlockManager()
.getStoredBlock(block.getLocalBlock());
bInfo.delete();
cluster.getNameNode().getNamesystem().getBlockManager()
.removeBlock(bInfo);
cluster.getNameNode().getNamesystem().writeUnlock();
@ -146,6 +147,7 @@ public void testEnsureGenStampsIsStartupOnly() throws
BlockInfo bInfo = cluster.getNameNode().getNamesystem().getBlockManager
().getStoredBlock(block.getLocalBlock());
cluster.getNameNode().getNamesystem().writeLock();
bInfo.delete();
cluster.getNameNode().getNamesystem().getBlockManager()
.removeBlock(bInfo);
cluster.getNameNode().getNamesystem().writeUnlock();