BlockManager { @VisibleForTesting DatanodeDescriptor chooseSourceDatanode(Block b, List<DatanodeDescriptor> containingNodes, List<DatanodeStorageInfo> nodesContainingLiveReplicas, NumberReplicas numReplicas, int priority) throws IOException { containingNodes.clear(); nodesContainingLiveReplicas.clear(); DatanodeDescriptor srcNode = null; int live = 0; int decommissioned = 0; int decommissioning = 0; int corrupt = 0; int excess = 0; final BlockInfoContiguous block = getBlockInfo(b); Collection<DatanodeDescriptor> nodesCorrupt = corruptReplicas.getNodes(block); for(DatanodeStorageInfo storage : block.getStorages(datanodeManager)) { final DatanodeDescriptor node = storage.getDatanodeDescriptor(); int countableReplica = storage.getState() == State.NORMAL ? 1 : 0; if ((nodesCorrupt != null) && (nodesCorrupt.contains(node))) { corrupt += countableReplica; } else if (node.isDecommissionInProgress()) { decommissioning += countableReplica; } else if (node.isDecommissioned()) { decommissioned += countableReplica; } else if (excessReplicateMap.contains(storage, block)) { excess += countableReplica; } else { nodesContainingLiveReplicas.add(storage); live+= countableReplica; } if(!containingNodes.contains(node)) { containingNodes.add(node); } if ((nodesCorrupt != null) && nodesCorrupt.contains(node)) { continue; } if(priority != UnderReplicatedBlocks.QUEUE_HIGHEST_PRIORITY && !node.isDecommissionInProgress() && node.getNumberOfBlocksToBeReplicated() >= maxReplicationStreams) { continue; } if (node.getNumberOfBlocksToBeReplicated() >= replicationStreamsHardLimit) { continue; } if (excessReplicateMap.contains(storage, block)) { continue; } if (node.isDecommissioned()) { continue; } if (srcNode == null) { srcNode = node; continue; } if (DFSUtil.getRandom().nextBoolean()) { srcNode = node; } } if(numReplicas != null) numReplicas.initialize(live, decommissioned, decommissioning, corrupt, excess, 0); return srcNode; } BlockManager(final Namesystem namesystem, final Configuration conf); long getPendingReplicationBlocksCount(); long getUnderReplicatedBlocksCount(); long getCorruptReplicaBlocksCount(); long getScheduledReplicationBlocksCount(); long getPendingDeletionBlocksCount(); long getStartupDelayBlockDeletionInMs(); long getExcessBlocksCount(); long getPostponedMisreplicatedBlocksCount(); BlockStoragePolicy getDefaultStoragePolicy(); BlockStoragePolicy getStoragePolicy(final String policyName); BlockStoragePolicy getStoragePolicy(final byte policyId); BlockStoragePolicy[] getStoragePolicies(); void setBlockPoolId(String blockPoolId); BlockStoragePolicySuite getStoragePolicySuite(); @VisibleForTesting BlockTokenSecretManager getBlockTokenSecretManager(); void activate(Configuration conf); void close(); DatanodeManager getDatanodeManager(); @VisibleForTesting BlockPlacementPolicy getBlockPlacementPolicy(); void setBlockPlacementPolicy(BlockPlacementPolicy newpolicy); int getMaxReplicationStreams(); boolean checkMinReplication(Block block); boolean commitOrCompleteLastBlock(BlockCollection bc,
      Block commitBlock); BlockInfoContiguous forceCompleteBlock(final BlockCollection bc,
      final BlockInfoContiguousUnderConstruction block); LocatedBlock convertLastBlockToUnderConstruction(
      BlockCollection bc, long bytesToRemove); LocatedBlocks createPhantomLocatedBlocks(INodeFile file, byte[] data,
      final boolean isFileUnderConstruction,
      final boolean needBlockToken, FileEncryptionInfo feInfo); LocatedBlocks createLocatedBlocks(final BlockInfoContiguous[] blocks,
      final long fileSizeExcludeBlocksUnderConstruction,
      final boolean isFileUnderConstruction, final long offset,
      final long length, final boolean needBlockToken, FileEncryptionInfo feInfo); ExportedBlockKeys getBlockKeys(); void setBlockToken(final LocatedBlock b,
      final AccessMode mode); DataEncryptionKey generateDataEncryptionKey(); short adjustReplication(short replication); void verifyReplication(String src, short replication,
      String clientName); boolean isSufficientlyReplicated(BlockInfoContiguous b); BlocksWithLocations getBlocks(DatanodeID datanode, long size
      // used in the namenode protocol
  ); void findAndMarkBlockAsCorrupt(final ExtendedBlock blk,
      final DatanodeInfo dn, final String storageID, final String reason); int getUnderReplicatedNotMissingBlocks(); DatanodeStorageInfo[] chooseTarget4NewBlock(final String src,
      final int numOfReplicas, final Node client,
      final Set<Node> excludedNodes,
      final long blocksize,
      final List<String> favoredNodes,
      final byte storagePolicyID); DatanodeStorageInfo[] chooseTarget4WebHDFS(String src,
      DatanodeDescriptor clientnode, Set<Node> excludes, long blocksize); DatanodeStorageInfo[] chooseTarget4AdditionalDatanode(String src,
      int numAdditionalNodes,
      Node clientnode,
      List<DatanodeStorageInfo> chosen,
      Set<Node> excludes,
      long blocksize,
      byte storagePolicyID); DatanodeStorageInfo[] chooseTarget4ParityRepair(String src,  int numOfReplicas,
      Node clientnode,List<DatanodeStorageInfo> chosen, Set<Node> excludes, long blocksize,
      byte storagePolicyID); List<Integer> checkHashes(final DatanodeID nodeID,
                                   final DatanodeStorage storage,
                                   final BlockReport newReport); boolean processReport(final DatanodeID nodeID,
      final DatanodeStorage storage,
      final BlockReport newReport, BlockReportContext context,
      boolean lastStorageInRpc); void markBlockReplicasAsCorrupt(BlockInfoContiguous block, 
      long oldGenerationStamp, long oldNumBytes, 
      DatanodeStorageInfo[] newStorages); @VisibleForTesting ReportStatistics processReport(final DatanodeStorageInfo storage,final BlockReport report); @VisibleForTesting void removeBlocks(List<Long> allBlockIds, final DatanodeDescriptor node); void removeBlocks(Map<Long,Long> allBlocksAndInodesIds, final DatanodeDescriptor node); synchronized void processMisReplicatedBlocks(); double getReplicationQueuesInitProgress(); void setReplication(final short oldRepl, final short newRepl,
      final String src, final Block... blocks); void removeStoredBlock(Block block, DatanodeDescriptor node); void removeStoredBlock(Block block, int sid); void processIncrementalBlockReport(DatanodeRegistration nodeID,
      final StorageReceivedDeletedBlocks blockInfos); NumberReplicas countNodes(Block b); int getActiveBlockCount(); DatanodeStorageInfo[] getStorages(BlockInfoContiguous block); int getTotalBlocks(); void removeBlock(Block block); BlockInfoContiguous getStoredBlock(Block block); void checkReplication(BlockCollection bc); boolean checkBlocksProperlyReplicated(
      String src, BlockInfoContiguous[] blocks); long getMissingBlocksCount(); long getMissingReplOneBlocksCount(); BlockInfoContiguous addBlockCollection(BlockInfoContiguous block, 
      BlockCollection bc); BlockCollection getBlockCollection(Block b); Iterable<DatanodeStorageInfo> getStorages(final Block block); List<DatanodeStorageInfo> storageList(final Block block); int numCorruptReplicas(Block block); void removeBlockFromMap(Block block); int getCapacity(); long[] getCorruptReplicaBlockIds(int numExpectedBlocks,
      Long startingBlockId); Iterator<Block> getCorruptReplicaBlockIterator(); Collection<DatanodeDescriptor> getCorruptReplicas(Block block); String getCorruptReason(BlockInfoContiguous block, DatanodeDescriptor node); int numOfUnderReplicatedBlocks(); void clearQueues(); static LocatedBlock newLocatedBlock(
      ExtendedBlock b, DatanodeStorageInfo[] storages,
      long startOffset, boolean corrupt); BlockInfoContiguous tryToCompleteBlock(final BlockCollection bc,
      final int blkIndex); @VisibleForTesting void processTimedOutPendingBlock(final long timedOutItemId); int getTotalCompleteBlocks(); void shutdown(); int getNumBuckets(); int getBlockFetcherNBThreads(); int getBlockFetcherBucketsPerThread(); int getRemovalBatchSize(); int getRemovalNoThreads(); void blockReportCompleted(final DatanodeID nodeID, DatanodeStorage[] storages, boolean success); static final Logger LOG; static final Logger blockLog; final ExcessReplicasMap excessReplicateMap; final UnderReplicatedBlocks neededReplications; final short maxReplication; final short minReplication; final int defaultReplication; }
@Test public void testHighestPriReplSrcChosenDespiteMaxReplLimit() throws Exception { Configuration conf = new HdfsConfiguration(); HdfsStorageFactory.setConfiguration(conf); HdfsStorageFactory.reset(); formatStorage(conf); bm.maxReplicationStreams = 0; bm.replicationStreamsHardLimit = 1; final long blockId = 42; final Block aBlock = new Block(blockId, 0, 0); final List<DatanodeDescriptor> origNodes = getNodes(0, 1); addNodes(origNodes); addBlockOnNodes(blockId, origNodes.subList(0, 1)); final List<DatanodeDescriptor> cntNodes = new LinkedList<>(); final List<DatanodeStorageInfo> liveNodes = new LinkedList<>(); new HopsTransactionalRequestHandler(HDFSOperationType.TEST) { INodeIdentifier inodeIdentifier; @Override public void setUp() throws StorageException { inodeIdentifier = INodeUtil.resolveINodeFromBlock(aBlock); } @Override public void acquireLock(TransactionLocks locks) throws IOException { LockFactory lf = LockFactory.getInstance(); locks.add( lf.getIndividualINodeLock(INodeLockType.WRITE, inodeIdentifier)) .add( lf.getIndividualBlockLock(aBlock.getBlockId(), inodeIdentifier)) .add(lf.getBlockRelated(BLK.RE, BLK.ER, BLK.CR, BLK.UR, BLK.PE)); } @Override public Object performTask() throws IOException { assertNotNull("Chooses source node for a highest-priority replication" + " even if all available source nodes have reached their replication" + " limits below the hard limit.", bm.chooseSourceDatanode(aBlock, cntNodes, liveNodes, new NumberReplicas(), UnderReplicatedBlocks.QUEUE_HIGHEST_PRIORITY)); assertNull( "Does not choose a source node for a less-than-highest-priority" + " replication since all available source nodes have reached" + " their replication limits.", bm.chooseSourceDatanode(aBlock, cntNodes, liveNodes, new NumberReplicas(), UnderReplicatedBlocks.QUEUE_VERY_UNDER_REPLICATED)); DatanodeStorageInfo targets[] = { origNodes.get(1).getStorageInfos()[0] }; origNodes.get(0).addBlockToBeReplicated(aBlock, targets); assertNull("Does not choose a source node for a highest-priority" + " replication when all available nodes exceed the hard limit.", bm.chooseSourceDatanode(aBlock, cntNodes, liveNodes, new NumberReplicas(), UnderReplicatedBlocks.QUEUE_HIGHEST_PRIORITY)); return null; } }.handle(); }
@Test public void testFavorDecomUntilHardLimit() throws Exception { Configuration conf = new HdfsConfiguration(); HdfsStorageFactory.setConfiguration(conf); HdfsStorageFactory.reset(); formatStorage(conf); bm.maxReplicationStreams = 0; bm.replicationStreamsHardLimit = 1; long blockId = 42; final Block aBlock = new Block(blockId, 0, 0); final List<DatanodeDescriptor> origNodes = getNodes(0, 1); addNodes(origNodes); addBlockOnNodes(blockId,origNodes.subList(0,1)); origNodes.get(0).startDecommission(); final List<DatanodeDescriptor> cntNodes = new LinkedList<DatanodeDescriptor>(); final List<DatanodeStorageInfo> liveNodes = new LinkedList<DatanodeStorageInfo>(); new HopsTransactionalRequestHandler(HDFSOperationType.TEST) { INodeIdentifier inodeIdentifier; @Override public void setUp() throws StorageException { inodeIdentifier = INodeUtil.resolveINodeFromBlock(aBlock); } @Override public void acquireLock(TransactionLocks locks) throws IOException { LockFactory lf = LockFactory.getInstance(); locks.add( lf.getIndividualINodeLock(INodeLockType.WRITE, inodeIdentifier)) .add( lf.getIndividualBlockLock(aBlock.getBlockId(), inodeIdentifier)) .add(lf.getBlockRelated(BLK.RE, BLK.ER, BLK.CR, BLK.UR, BLK.PE)); } @Override public Object performTask() throws IOException { assertNotNull("Chooses decommissioning source node for a normal replication" + " if all available source nodes have reached their replication" + " limits below the hard limit.", bm.chooseSourceDatanode( aBlock, cntNodes, liveNodes, new NumberReplicas(), UnderReplicatedBlocks.QUEUE_UNDER_REPLICATED)); DatanodeStorageInfo targets[] = {origNodes.get(1).getStorageInfos()[0]}; origNodes.get(0).addBlockToBeReplicated(aBlock, targets); assertNull("Does not choose a source decommissioning node for a normal" + " replication when all available nodes exceed the hard limit.", bm.chooseSourceDatanode( aBlock, cntNodes, liveNodes, new NumberReplicas(), UnderReplicatedBlocks.QUEUE_UNDER_REPLICATED)); return null; } }.handle(); }