BlockManager implements BlockStatsMXBean { @VisibleForTesting DatanodeDescriptor[] chooseSourceDatanodes(BlockInfo block, List<DatanodeDescriptor> containingNodes, List<DatanodeStorageInfo> nodesContainingLiveReplicas, NumberReplicas numReplicas, List<Byte> liveBlockIndices, List<Byte> liveBusyBlockIndices, int priority) { containingNodes.clear(); nodesContainingLiveReplicas.clear(); List<DatanodeDescriptor> srcNodes = new ArrayList<>(); liveBlockIndices.clear(); final boolean isStriped = block.isStriped(); DatanodeDescriptor decommissionedSrc = null; BitSet liveBitSet = null; BitSet decommissioningBitSet = null; if (isStriped) { int blockNum = ((BlockInfoStriped) block).getTotalBlockNum(); liveBitSet = new BitSet(blockNum); decommissioningBitSet = new BitSet(blockNum); } for (DatanodeStorageInfo storage : blocksMap.getStorages(block)) { final DatanodeDescriptor node = getDatanodeDescriptorFromStorage(storage); final StoredReplicaState state = checkReplicaOnStorage(numReplicas, block, storage, corruptReplicas.getNodes(block), false); if (state == StoredReplicaState.LIVE) { if (storage.getStorageType() == StorageType.PROVIDED) { storage = new DatanodeStorageInfo(node, storage.getStorageID(), storage.getStorageType(), storage.getState()); } nodesContainingLiveReplicas.add(storage); } containingNodes.add(node); if (state == StoredReplicaState.CORRUPT || state == StoredReplicaState.EXCESS) { continue; } if (state == null || state == StoredReplicaState.MAINTENANCE_NOT_FOR_READ) { continue; } if (state == StoredReplicaState.DECOMMISSIONED) { if (decommissionedSrc == null || ThreadLocalRandom.current().nextBoolean()) { decommissionedSrc = node; } continue; } byte blockIndex = -1; if (isStriped) { blockIndex = ((BlockInfoStriped) block) .getStorageBlockIndex(storage); countLiveAndDecommissioningReplicas(numReplicas, state, liveBitSet, decommissioningBitSet, blockIndex); } if (priority != LowRedundancyBlocks.QUEUE_HIGHEST_PRIORITY && (!node.isDecommissionInProgress() && !node.isEnteringMaintenance()) && node.getNumberOfBlocksToBeReplicated() >= maxReplicationStreams) { if (isStriped && (state == StoredReplicaState.LIVE || state == StoredReplicaState.DECOMMISSIONING)) { liveBusyBlockIndices.add(blockIndex); } continue; } if (node.getNumberOfBlocksToBeReplicated() >= replicationStreamsHardLimit) { if (isStriped && (state == StoredReplicaState.LIVE || state == StoredReplicaState.DECOMMISSIONING)) { liveBusyBlockIndices.add(blockIndex); } continue; } if(isStriped || srcNodes.isEmpty()) { srcNodes.add(node); if (isStriped) { liveBlockIndices.add(blockIndex); } continue; } if (ThreadLocalRandom.current().nextBoolean()) { srcNodes.set(0, node); } } if (!isStriped && nodesContainingLiveReplicas.isEmpty() && srcNodes.isEmpty() && decommissionedSrc != null) { srcNodes.add(decommissionedSrc); } return srcNodes.toArray(new DatanodeDescriptor[srcNodes.size()]); } BlockManager(final Namesystem namesystem, boolean haEnabled,
      final Configuration conf); long getPendingReconstructionBlocksCount(); long getLowRedundancyBlocksCount(); long getCorruptReplicaBlocksCount(); long getScheduledReplicationBlocksCount(); long getPendingDeletionBlocksCount(); long getStartupDelayBlockDeletionInMs(); long getExcessBlocksCount(); long getPostponedMisreplicatedBlocksCount(); int getPendingDataNodeMessageCount(); long getNumTimedOutPendingReconstructions(); long getLowRedundancyBlocks(); long getCorruptBlocks(); long getMissingBlocks(); long getMissingReplicationOneBlocks(); long getPendingDeletionReplicatedBlocks(); long getTotalReplicatedBlocks(); long getLowRedundancyECBlockGroups(); long getCorruptECBlockGroups(); long getMissingECBlockGroups(); long getPendingDeletionECBlocks(); long getTotalECBlockGroups(); BlockStoragePolicy getStoragePolicy(final String policyName); BlockStoragePolicy getStoragePolicy(final byte policyId); BlockStoragePolicy[] getStoragePolicies(); void setBlockPoolId(String blockPoolId); String getBlockPoolId(); BlockStoragePolicySuite getStoragePolicySuite(); @VisibleForTesting BlockTokenSecretManager getBlockTokenSecretManager(); void activate(Configuration conf, long blockTotal); void close(); DatanodeManager getDatanodeManager(); @VisibleForTesting BlockPlacementPolicy getBlockPlacementPolicy(); void refreshBlockPlacementPolicy(Configuration conf); void metaSave(PrintWriter out); int getMaxReplicationStreams(); void setMaxReplicationStreams(int newVal); int getReplicationStreamsHardLimit(); void setReplicationStreamsHardLimit(int newVal); int getBlocksReplWorkMultiplier(); void setBlocksReplWorkMultiplier(int newVal); int getDefaultStorageNum(BlockInfo block); short getMinReplication(); short getMinStorageNum(BlockInfo block); short getMinReplicationToBeInMaintenance(); boolean hasMinStorage(BlockInfo block); boolean hasMinStorage(BlockInfo block, int liveNum); boolean commitOrCompleteLastBlock(BlockCollection bc,
      Block commitBlock, INodesInPath iip); void addExpectedReplicasToPending(BlockInfo blk); void forceCompleteBlock(final BlockInfo block); LocatedBlock convertLastBlockToUnderConstruction(
      BlockCollection bc, long bytesToRemove); LocatedBlocks createLocatedBlocks(final BlockInfo[] blocks,
      final long fileSizeExcludeBlocksUnderConstruction,
      final boolean isFileUnderConstruction, final long offset,
      final long length, final boolean needBlockToken,
      final boolean inSnapshot, FileEncryptionInfo feInfo,
      ErasureCodingPolicy ecPolicy); 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(BlockInfo b); BlocksWithLocations getBlocksWithLocations(final DatanodeID datanode,
      final long size, final long minBlockSize); void findAndMarkBlockAsCorrupt(final ExtendedBlock blk,
      final DatanodeInfo dn, String storageID, String reason); void setPostponeBlocksFromFuture(boolean postpone); int getUnderReplicatedNotMissingBlocks(); 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,
      BlockType blockType); 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,
      final BlockType blockType,
      final ErasureCodingPolicy ecPolicy,
      final EnumSet<AddBlockFlag> flags); long requestBlockReportLeaseId(DatanodeRegistration nodeReg); void registerDatanode(DatanodeRegistration nodeReg); void setBlockTotal(long total); boolean isInSafeMode(); String getSafeModeTip(); boolean leaveSafeMode(boolean force); void checkSafeMode(); long getBytesInFuture(); long getBytesInFutureReplicatedBlocks(); long getBytesInFutureECBlockGroups(); void removeBlocksAndUpdateSafemodeTotal(BlocksMapUpdateInfo blocks); long getProvidedCapacity(); boolean checkBlockReportLease(BlockReportContext context,
      final DatanodeID nodeID); boolean processReport(final DatanodeID nodeID,
      final DatanodeStorage storage,
      final BlockListAsLongs newReport,
      BlockReportContext context); void removeBRLeaseIfNeeded(final DatanodeID nodeID,
      final BlockReportContext context); void markBlockReplicasAsCorrupt(Block oldBlock,
      BlockInfo block,
      long oldGenerationStamp, long oldNumBytes, 
      DatanodeStorageInfo[] newStorages); void processQueuedMessagesForBlock(Block b); void processAllPendingDNMessages(); void processMisReplicatedBlocks(); double getReconstructionQueuesInitProgress(); boolean hasNonEcBlockUsingStripedID(); int processMisReplicatedBlocks(List<BlockInfo> blocks); void setReplication(
      final short oldRepl, final short newRepl, final BlockInfo b); void removeStoredBlock(BlockInfo storedBlock, DatanodeDescriptor node); @VisibleForTesting void addBlock(DatanodeStorageInfo storageInfo, Block block,
      String delHint); void processIncrementalBlockReport(final DatanodeID nodeID,
      final StorageReceivedDeletedBlocks srdb); NumberReplicas countNodes(BlockInfo b); boolean isExcess(DatanodeDescriptor dn, BlockInfo blk); int getActiveBlockCount(); DatanodeStorageInfo[] getStorages(BlockInfo block); Iterable<DatanodeStorageInfo> getStorages(final Block block); int getTotalBlocks(); void removeBlock(BlockInfo block); BlockInfo getStoredBlock(Block block); void updateLastBlock(BlockInfo lastBlock, ExtendedBlock newBlock); void checkRedundancy(BlockCollection bc); @VisibleForTesting boolean containsInvalidateBlock(final DatanodeInfo dn,
      final Block block); short getExpectedLiveRedundancyNum(BlockInfo block,
      NumberReplicas numberReplicas); short getExpectedRedundancyNum(BlockInfo block); long getMissingBlocksCount(); long getMissingReplOneBlocksCount(); long getHighestPriorityReplicatedBlockCount(); long getHighestPriorityECBlockCount(); BlockInfo addBlockCollection(BlockInfo block,
      BlockCollection bc); BlockInfo addBlockCollectionWithCheck(
      BlockInfo block, BlockCollection bc); int numCorruptReplicas(Block block); void removeBlockFromMap(BlockInfo block); int getCapacity(); Iterator<BlockInfo> getCorruptReplicaBlockIterator(); Collection<DatanodeDescriptor> getCorruptReplicas(Block block); String getCorruptReason(Block block, DatanodeDescriptor node); int numOfUnderReplicatedBlocks(); @VisibleForTesting long getLastRedundancyMonitorTS(); void clearQueues(); static LocatedBlock newLocatedBlock(
      ExtendedBlock b, DatanodeStorageInfo[] storages,
      long startOffset, boolean corrupt); static LocatedStripedBlock newLocatedStripedBlock(
      ExtendedBlock b, DatanodeStorageInfo[] storages,
      byte[] indices, long startOffset, boolean corrupt); static LocatedBlock newLocatedBlock(ExtendedBlock eb, BlockInfo info,
      DatanodeStorageInfo[] locs, long offset); void shutdown(); void clear(); BlockReportLeaseManager getBlockReportLeaseManager(); @Override // BlockStatsMXBean Map<StorageType, StorageTypeStats> getStorageTypeStats(); void initializeReplQueues(); boolean isPopulatingReplQueues(); void setInitializedReplQueues(boolean v); boolean shouldPopulateReplQueues(); void enqueueBlockOp(final Runnable action); T runBlockOp(final Callable<T> action); void successfulBlockRecovery(BlockInfo block); boolean addBlockRecoveryAttempt(BlockInfo b); @VisibleForTesting void flushBlockOps(); int getBlockOpQueueLength(); BlockIdManager getBlockIdManager(); long nextGenerationStamp(boolean legacyBlock); boolean isLegacyBlock(Block block); long nextBlockId(BlockType blockType); @VisibleForTesting void setBlockRecoveryTimeout(long blockRecoveryTimeout); @VisibleForTesting ProvidedStorageMap getProvidedStorageMap(); boolean createSPSManager(final Configuration conf,
      final String spsMode); void disableSPS(); StoragePolicySatisfyManager getSPSManager(); static final Logger LOG; static final Logger blockLog; final LowRedundancyBlocks neededReconstruction; final short maxReplication; final short minReplication; final int defaultReplication; }
@Test public void testChooseSrcDatanodesWithDupEC() throws Exception { bm.maxReplicationStreams = 4; long blockId = -9223372036854775776L; Block aBlock = new Block(blockId, 0, 0); ECSchema rsSchema = new ECSchema("rs", 3, 2); String policyName = "RS-3-2-128k"; int cellSize = 128 * 1024; ErasureCodingPolicy ecPolicy = new ErasureCodingPolicy(policyName, rsSchema, cellSize, (byte) -1); BlockInfoStriped aBlockInfoStriped = new BlockInfoStriped(aBlock, ecPolicy); DatanodeStorageInfo ds1 = DFSTestUtil.createDatanodeStorageInfo( "storage1", "1.1.1.1", "rack1", "host1"); DatanodeStorageInfo ds2 = DFSTestUtil.createDatanodeStorageInfo( "storage2", "2.2.2.2", "rack2", "host2"); DatanodeStorageInfo ds3 = DFSTestUtil.createDatanodeStorageInfo( "storage3", "3.3.3.3", "rack3", "host3"); DatanodeStorageInfo ds4 = DFSTestUtil.createDatanodeStorageInfo( "storage4", "4.4.4.4", "rack4", "host4"); DatanodeStorageInfo ds5 = DFSTestUtil.createDatanodeStorageInfo( "storage5", "5.5.5.5", "rack5", "host5"); aBlockInfoStriped.addStorage(ds1, aBlock); aBlockInfoStriped.addStorage(ds2, new Block(blockId + 1, 0, 0)); aBlockInfoStriped.addStorage(ds3, new Block(blockId + 2, 0, 0)); aBlockInfoStriped.addStorage(ds4, new Block(blockId + 3, 0, 0)); aBlockInfoStriped.addStorage(ds5, new Block(blockId + 3, 0, 0)); for(int i = 0; i < 4; i++){ ds4.getDatanodeDescriptor().incrementPendingReplicationWithoutTargets(); } addEcBlockToBM(blockId, ecPolicy); List<DatanodeDescriptor> cntNodes = new LinkedList<DatanodeDescriptor>(); List<DatanodeStorageInfo> liveNodes = new LinkedList<DatanodeStorageInfo>(); NumberReplicas numReplicas = new NumberReplicas(); List<Byte> liveBlockIndices = new ArrayList<>(); List<Byte> liveBusyBlockIndices = new ArrayList<>(); bm.chooseSourceDatanodes( aBlockInfoStriped, cntNodes, liveNodes, numReplicas, liveBlockIndices, liveBusyBlockIndices, LowRedundancyBlocks.QUEUE_VERY_LOW_REDUNDANCY); assertEquals("Choose the source node for reconstruction with one node reach" + " the MAX maxReplicationStreams, the numReplicas still return the" + " correct live replicas.", 4, numReplicas.liveReplicas()); assertEquals("Choose the source node for reconstruction with one node reach" + " the MAX maxReplicationStreams, the numReplicas should return" + " the correct redundant Internal Blocks.", 1, numReplicas.redundantInternalBlocks()); }
@Test public void testChooseSrcDNWithDupECInDecommissioningNode() throws Exception { long blockId = -9223372036854775776L; Block aBlock = new Block(blockId, 0, 0); ErasureCodingPolicy ecPolicy = SystemErasureCodingPolicies.getPolicies().get(1); BlockInfoStriped aBlockInfoStriped = new BlockInfoStriped(aBlock, ecPolicy); DatanodeStorageInfo ds1 = DFSTestUtil.createDatanodeStorageInfo( "storage1", "1.1.1.1", "rack1", "host1"); DatanodeStorageInfo ds2 = DFSTestUtil.createDatanodeStorageInfo( "storage2", "2.2.2.2", "rack2", "host2"); DatanodeStorageInfo ds3 = DFSTestUtil.createDatanodeStorageInfo( "storage3", "3.3.3.3", "rack3", "host3"); DatanodeStorageInfo ds4 = DFSTestUtil.createDatanodeStorageInfo( "storage4", "4.4.4.4", "rack4", "host4"); DatanodeStorageInfo ds5 = DFSTestUtil.createDatanodeStorageInfo( "storage5", "5.5.5.5", "rack5", "host5"); DatanodeStorageInfo ds6 = DFSTestUtil.createDatanodeStorageInfo( "storage6", "6.6.6.6", "rack6", "host6"); aBlockInfoStriped.addStorage(ds1, aBlock); aBlockInfoStriped.addStorage(ds2, new Block(blockId + 1, 0, 0)); aBlockInfoStriped.addStorage(ds3, new Block(blockId + 2, 0, 0)); aBlockInfoStriped.addStorage(ds4, new Block(blockId + 3, 0, 0)); aBlockInfoStriped.addStorage(ds5, new Block(blockId + 4, 0, 0)); aBlockInfoStriped.addStorage(ds6, aBlock); addEcBlockToBM(blockId, ecPolicy); ds1.getDatanodeDescriptor().startDecommission(); List<DatanodeDescriptor> containingNodes = new LinkedList<DatanodeDescriptor>(); List<DatanodeStorageInfo> nodesContainingLiveReplicas = new LinkedList<DatanodeStorageInfo>(); NumberReplicas numReplicas = new NumberReplicas(); List<Byte> liveBlockIndices = new ArrayList<>(); List<Byte> liveBusyBlockIndices = new ArrayList<>(); bm.chooseSourceDatanodes( aBlockInfoStriped, containingNodes, nodesContainingLiveReplicas, numReplicas, liveBlockIndices, liveBusyBlockIndices, LowRedundancyBlocks.QUEUE_HIGHEST_PRIORITY); assertEquals("There are 5 live replicas in " + "[ds2, ds3, ds4, ds5, ds6] datanodes ", 5, numReplicas.liveReplicas()); assertEquals("The ds1 datanode is in decommissioning, " + "so there is no redundant replica", 0, numReplicas.redundantInternalBlocks()); }