HDFS-9542. Move BlockIdManager from FSNamesystem to BlockManager. Con…

…tributed by Jing Zhao.

hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

@@ -939,6 +939,8 @@ Release 2.9.0 - UNRELEASED
     HDFS-9576: HTrace: collect position/length information on read operations
     (zhz via cmccabe)
 
+    HDFS-9542. Move BlockIdManager from FSNamesystem to BlockManager. (jing9)
+
   OPTIMIZATIONS
 
   BUG FIXES

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockIdManager.java

@@ -163,7 +163,7 @@ public long getGenerationStampV2() {
   /**
    * Increments, logs and then returns the stamp
    */
-  public long nextGenerationStamp(boolean legacyBlock) throws IOException {
+  long nextGenerationStamp(boolean legacyBlock) throws IOException {
     return legacyBlock ? getNextGenerationStampV1() :
       getNextGenerationStampV2();
   }
@@ -199,30 +199,27 @@ public long getGenerationStampV1Limit() {
    *
    * @return true if the block ID was randomly generated, false otherwise.
    */
-  public boolean isLegacyBlock(Block block) {
+  boolean isLegacyBlock(Block block) {
     return block.getGenerationStamp() < getGenerationStampV1Limit();
   }
 
   /**
    * Increments, logs and then returns the block ID
    */
-  public long nextContiguousBlockId() {
-    return blockIdGenerator.nextValue();
+  long nextBlockId(boolean isStriped) {
+    return isStriped ? blockGroupIdGenerator.nextValue() :
+        blockIdGenerator.nextValue();
   }
 
-  public long nextStripedBlockId() {
-    return blockGroupIdGenerator.nextValue();
-  }
-
-  public boolean isGenStampInFuture(Block block) {
+  boolean isGenStampInFuture(Block block) {
     if (isLegacyBlock(block)) {
       return block.getGenerationStamp() > getGenerationStampV1();
     } else {
       return block.getGenerationStamp() > getGenerationStampV2();
     }
   }
 
-  public void clear() {
+  void clear() {
     generationStampV1.setCurrentValue(GenerationStamp.LAST_RESERVED_STAMP);
     generationStampV2.setCurrentValue(GenerationStamp.LAST_RESERVED_STAMP);
     getBlockIdGenerator().setCurrentValue(SequentialBlockIdGenerator
@@ -240,7 +237,7 @@ public static boolean isStripedBlockID(long id) {
    * and the other 60 bits are 1. Group ID is the first 60 bits of any
    * data/parity block id in the same striped block group.
    */
-  public static long convertToStripedID(long id) {
+  static long convertToStripedID(long id) {
     return id & (~HdfsServerConstants.BLOCK_GROUP_INDEX_MASK);
   }
 

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

@@ -78,7 +78,6 @@
 import org.apache.hadoop.hdfs.server.namenode.CachedBlock;
 import org.apache.hadoop.hdfs.server.namenode.INode.BlocksMapUpdateInfo;
 import org.apache.hadoop.hdfs.server.namenode.NameNode;
-import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory;
 import org.apache.hadoop.hdfs.server.namenode.Namesystem;
 import org.apache.hadoop.hdfs.server.namenode.ha.HAContext;
 import org.apache.hadoop.hdfs.server.namenode.metrics.NameNodeMetrics;
@@ -308,11 +307,14 @@ public int getPendingDataNodeMessageCount() {
   /** Check whether there are any non-EC blocks using StripedID */
   private boolean hasNonEcBlockUsingStripedID = false;
 
-  public BlockManager(final Namesystem namesystem, final Configuration conf)
-    throws IOException {
+  private final BlockIdManager blockIdManager;
+
+  public BlockManager(final Namesystem namesystem, boolean haEnabled,
+      final Configuration conf) throws IOException {
     this.namesystem = namesystem;
     datanodeManager = new DatanodeManager(this, namesystem, conf);
     heartbeatManager = datanodeManager.getHeartbeatManager();
+    this.blockIdManager = new BlockIdManager(this);
 
     startupDelayBlockDeletionInMs = conf.getLong(
         DFSConfigKeys.DFS_NAMENODE_STARTUP_DELAY_BLOCK_DELETION_SEC_KEY,
@@ -387,7 +389,7 @@ public BlockManager(final Namesystem namesystem, final Configuration conf)
         DFSConfigKeys.DFS_BLOCK_MISREPLICATION_PROCESSING_LIMIT_DEFAULT);
     this.blockReportLeaseManager = new BlockReportLeaseManager(conf);
 
-    bmSafeMode = new BlockManagerSafeMode(this, namesystem, conf);
+    bmSafeMode = new BlockManagerSafeMode(this, namesystem, haEnabled, conf);
 
     LOG.info("defaultReplication         = " + defaultReplication);
     LOG.info("maxReplication             = " + maxReplication);
@@ -498,8 +500,7 @@ private boolean isBlockTokenEnabled() {
 
   /** Should the access keys be updated? */
   boolean shouldUpdateBlockKey(final long updateTime) throws IOException {
-    return isBlockTokenEnabled()? blockTokenSecretManager.updateKeys(updateTime)
-        : false;
+    return isBlockTokenEnabled() && blockTokenSecretManager.updateKeys(updateTime);
   }
 
   public void activate(Configuration conf, long blockTotal) {
@@ -538,7 +539,7 @@ public BlockPlacementPolicy getBlockPlacementPolicy() {
 
   /** Dump meta data to out. */
   public void metaSave(PrintWriter out) {
-    assert namesystem.hasWriteLock();
+    assert namesystem.hasWriteLock(); // TODO: block manager read lock and NS write lock
     final List<DatanodeDescriptor> live = new ArrayList<DatanodeDescriptor>();
     final List<DatanodeDescriptor> dead = new ArrayList<DatanodeDescriptor>();
     datanodeManager.fetchDatanodes(live, dead, false);
@@ -1108,27 +1109,8 @@ public boolean isSufficientlyReplicated(BlockInfo b) {
     return countNodes(b).liveReplicas() >= replication;
   }
 
-  /**
-   * return a list of blocks & their locations on <code>datanode</code> whose
-   * total size is <code>size</code>
-   * 
-   * @param datanode on which blocks are located
-   * @param size total size of blocks
-   */
-  public BlocksWithLocations getBlocks(DatanodeID datanode, long size
-      ) throws IOException {
-    namesystem.checkOperation(OperationCategory.READ);
-    namesystem.readLock();
-    try {
-      namesystem.checkOperation(OperationCategory.READ);
-      return getBlocksWithLocations(datanode, size);  
-    } finally {
-      namesystem.readUnlock();
-    }
-  }
-
   /** Get all blocks with location information from a datanode. */
-  private BlocksWithLocations getBlocksWithLocations(final DatanodeID datanode,
+  public BlocksWithLocations getBlocksWithLocations(final DatanodeID datanode,
       final long size) throws UnregisteredNodeException {
     final DatanodeDescriptor node = getDatanodeManager().getDatanode(datanode);
     if (node == null) {
@@ -2353,8 +2335,7 @@ private void processFirstBlockReport(
             + " on " + storageInfo.getDatanodeDescriptor() + " size " +
             iblk.getNumBytes() + " replicaState = " + reportedState);
       }
-      if (shouldPostponeBlocksFromFuture &&
-          namesystem.isGenStampInFuture(iblk)) {
+      if (shouldPostponeBlocksFromFuture && isGenStampInFuture(iblk)) {
         queueReportedBlock(storageInfo, iblk, reportedState,
             QUEUE_REASON_FUTURE_GENSTAMP);
         continue;
@@ -2499,8 +2480,7 @@ private BlockInfo processReportedBlock(
           + " replicaState = " + reportedState);
     }
 
-    if (shouldPostponeBlocksFromFuture &&
-        namesystem.isGenStampInFuture(block)) {
+    if (shouldPostponeBlocksFromFuture && isGenStampInFuture(block)) {
       queueReportedBlock(storageInfo, block, reportedState,
           QUEUE_REASON_FUTURE_GENSTAMP);
       return null;
@@ -3360,8 +3340,7 @@ private void addToExcessReplicate(DatanodeInfo dn, BlockInfo storedBlock) {
 
   private void removeStoredBlock(DatanodeStorageInfo storageInfo, Block block,
       DatanodeDescriptor node) {
-    if (shouldPostponeBlocksFromFuture &&
-        namesystem.isGenStampInFuture(block)) {
+    if (shouldPostponeBlocksFromFuture && isGenStampInFuture(block)) {
       queueReportedBlock(storageInfo, block, null,
           QUEUE_REASON_FUTURE_GENSTAMP);
       return;
@@ -4201,6 +4180,7 @@ public void shutdown() {
   }
 
   public void clear() {
+    blockIdManager.clear();
     clearQueues();
     blocksMap.clear();
   }
@@ -4364,4 +4344,24 @@ void enqueue(Runnable action) throws InterruptedException {
       }
     }
   }
+
+  public BlockIdManager getBlockIdManager() {
+    return blockIdManager;
+  }
+
+  public long nextGenerationStamp(boolean legacyBlock) throws IOException {
+    return blockIdManager.nextGenerationStamp(legacyBlock);
+  }
+
+  public boolean isLegacyBlock(Block block) {
+    return blockIdManager.isLegacyBlock(block);
+  }
+
+  public long nextBlockId(boolean isStriped) {
+    return blockIdManager.nextBlockId(isStriped);
+  }
+
+  boolean isGenStampInFuture(Block block) {
+    return blockIdManager.isGenStampInFuture(block);
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManagerSafeMode.java

@@ -115,10 +115,10 @@ enum BMSafeModeStatus {
   private final boolean inRollBack;
 
   BlockManagerSafeMode(BlockManager blockManager, Namesystem namesystem,
-      Configuration conf) {
+      boolean haEnabled, Configuration conf) {
     this.blockManager = blockManager;
     this.namesystem = namesystem;
-    this.haEnabled = namesystem.isHaEnabled();
+    this.haEnabled = haEnabled;
     this.threshold = conf.getFloat(DFS_NAMENODE_SAFEMODE_THRESHOLD_PCT_KEY,
         DFS_NAMENODE_SAFEMODE_THRESHOLD_PCT_DEFAULT);
     if (this.threshold > 1.0) {
@@ -473,7 +473,7 @@ void checkBlocksWithFutureGS(BlockReportReplica brr) {
 
     if (!blockManager.getShouldPostponeBlocksFromFuture() &&
         !inRollBack &&
-        namesystem.isGenStampInFuture(brr)) {
+        blockManager.isGenStampInFuture(brr)) {
       numberOfBytesInFutureBlocks.addAndGet(brr.getBytesOnDisk());
     }
   }

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

@@ -228,7 +228,7 @@ static Block prepareFileForTruncate(FSNamesystem fsn, INodesInPath iip,
     if (newBlock == null) {
       newBlock = (shouldCopyOnTruncate) ? fsn.createNewBlock(false)
           : new Block(oldBlock.getBlockId(), oldBlock.getNumBytes(),
-          fsn.nextGenerationStamp(fsn.getBlockIdManager().isLegacyBlock(
+          fsn.nextGenerationStamp(fsn.getBlockManager().isLegacyBlock(
               oldBlock)));
     }
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLogLoader.java

@@ -48,6 +48,7 @@
 import org.apache.hadoop.hdfs.protocol.LastBlockWithStatus;
 import org.apache.hadoop.hdfs.protocol.LayoutVersion;
 import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfoContiguous;
+import org.apache.hadoop.hdfs.server.blockmanagement.BlockManager;
 import org.apache.hadoop.hdfs.server.common.HdfsServerConstants;
 import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState;
 import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.RollingUpgradeStartupOption;
@@ -116,12 +117,14 @@ public class FSEditLogLoader {
   static final long REPLAY_TRANSACTION_LOG_INTERVAL = 1000; // 1sec
 
   private final FSNamesystem fsNamesys;
+  private final BlockManager blockManager;
   private long lastAppliedTxId;
   /** Total number of end transactions loaded. */
   private int totalEdits = 0;
 
   public FSEditLogLoader(FSNamesystem fsNamesys, long lastAppliedTxId) {
     this.fsNamesys = fsNamesys;
+    this.blockManager = fsNamesys.getBlockManager();
     this.lastAppliedTxId = lastAppliedTxId;
   }
 
@@ -586,7 +589,7 @@ fsDir, renameReservedPathsOnUpgrade(deleteOp.path, logVersion),
     }
     case OP_SET_GENSTAMP_V1: {
       SetGenstampV1Op setGenstampV1Op = (SetGenstampV1Op)op;
-      fsNamesys.getBlockIdManager().setGenerationStampV1(
+      blockManager.getBlockIdManager().setGenerationStampV1(
           setGenstampV1Op.genStampV1);
       break;
     }
@@ -794,7 +797,7 @@ fsDir, renameReservedPathsOnUpgrade(timesOp.path, logVersion),
     }
     case OP_SET_GENSTAMP_V2: {
       SetGenstampV2Op setGenstampV2Op = (SetGenstampV2Op) op;
-      fsNamesys.getBlockIdManager().setGenerationStampV2(
+      blockManager.getBlockIdManager().setGenerationStampV2(
           setGenstampV2Op.genStampV2);
       break;
     }
@@ -803,10 +806,10 @@ fsDir, renameReservedPathsOnUpgrade(timesOp.path, logVersion),
       if (BlockIdManager.isStripedBlockID(allocateBlockIdOp.blockId)) {
         // ALLOCATE_BLOCK_ID is added for sequential block id, thus if the id
         // is negative, it must belong to striped blocks
-        fsNamesys.getBlockIdManager().setLastAllocatedStripedBlockId(
+        blockManager.getBlockIdManager().setLastAllocatedStripedBlockId(
             allocateBlockIdOp.blockId);
       } else {
-        fsNamesys.getBlockIdManager().setLastAllocatedContiguousBlockId(
+        blockManager.getBlockIdManager().setLastAllocatedContiguousBlockId(
             allocateBlockIdOp.blockId);
       }
       break;

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageFormat.java

@@ -48,11 +48,11 @@
 import org.apache.hadoop.fs.UnresolvedLinkException;
 import org.apache.hadoop.fs.permission.PermissionStatus;
 import org.apache.hadoop.hdfs.DFSUtil;
-import org.apache.hadoop.hdfs.protocol.Block;
 import org.apache.hadoop.hdfs.protocol.HdfsConstants;
 import org.apache.hadoop.hdfs.protocol.LayoutFlags;
 import org.apache.hadoop.hdfs.protocol.LayoutVersion;
 import org.apache.hadoop.hdfs.protocol.LayoutVersion.Feature;
+import org.apache.hadoop.hdfs.server.blockmanagement.BlockIdManager;
 import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfo;
 import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfoContiguous;
 import org.apache.hadoop.hdfs.server.blockmanagement.BlockManager;
@@ -344,27 +344,26 @@ public void load(File curFile) throws IOException {
 
         // read in the last generation stamp for legacy blocks.
         long genstamp = in.readLong();
-        namesystem.getBlockIdManager().setGenerationStampV1(genstamp);
+        final BlockIdManager blockIdManager = namesystem.getBlockManager()
+            .getBlockIdManager();
+        blockIdManager.setGenerationStampV1(genstamp);
 
         if (NameNodeLayoutVersion.supports(
             LayoutVersion.Feature.SEQUENTIAL_BLOCK_ID, imgVersion)) {
           // read the starting generation stamp for sequential block IDs
           genstamp = in.readLong();
-          namesystem.getBlockIdManager().setGenerationStampV2(genstamp);
+          blockIdManager.setGenerationStampV2(genstamp);
 
           // read the last generation stamp for blocks created after
           // the switch to sequential block IDs.
           long stampAtIdSwitch = in.readLong();
-          namesystem.getBlockIdManager().setGenerationStampV1Limit(stampAtIdSwitch);
+          blockIdManager.setGenerationStampV1Limit(stampAtIdSwitch);
 
           // read the max sequential block ID.
           long maxSequentialBlockId = in.readLong();
-          namesystem.getBlockIdManager().setLastAllocatedContiguousBlockId(
-              maxSequentialBlockId);
+          blockIdManager.setLastAllocatedContiguousBlockId(maxSequentialBlockId);
         } else {
-
-          long startingGenStamp = namesystem.getBlockIdManager()
-            .upgradeGenerationStampToV2();
+          long startingGenStamp = blockIdManager.upgradeGenerationStampToV2();
           // This is an upgrade.
           LOG.info("Upgrading to sequential block IDs. Generation stamp " +
                    "for new blocks set to " + startingGenStamp);
@@ -1269,10 +1268,12 @@ void save(File newFile, FSImageCompression compression) throws IOException {
         out.writeInt(sourceNamesystem.unprotectedGetNamespaceInfo()
             .getNamespaceID());
         out.writeLong(numINodes);
-        out.writeLong(sourceNamesystem.getBlockIdManager().getGenerationStampV1());
-        out.writeLong(sourceNamesystem.getBlockIdManager().getGenerationStampV2());
-        out.writeLong(sourceNamesystem.getBlockIdManager().getGenerationStampAtblockIdSwitch());
-        out.writeLong(sourceNamesystem.getBlockIdManager().getLastAllocatedContiguousBlockId());
+        final BlockIdManager blockIdManager = sourceNamesystem.getBlockManager()
+            .getBlockIdManager();
+        out.writeLong(blockIdManager.getGenerationStampV1());
+        out.writeLong(blockIdManager.getGenerationStampV2());
+        out.writeLong(blockIdManager.getGenerationStampAtblockIdSwitch());
+        out.writeLong(blockIdManager.getLastAllocatedContiguousBlockId());
         out.writeLong(context.getTxId());
         out.writeLong(sourceNamesystem.dir.getLastInodeId());
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageFormatProtobuf.java

@@ -293,7 +293,7 @@ public int compare(FileSummary.Section s1, FileSummary.Section s2) {
 
     private void loadNameSystemSection(InputStream in) throws IOException {
       NameSystemSection s = NameSystemSection.parseDelimitedFrom(in);
-      BlockIdManager blockIdManager = fsn.getBlockIdManager();
+      BlockIdManager blockIdManager = fsn.getBlockManager().getBlockIdManager();
       blockIdManager.setGenerationStampV1(s.getGenstampV1());
       blockIdManager.setGenerationStampV2(s.getGenstampV2());
       blockIdManager.setGenerationStampV1Limit(s.getGenstampV1Limit());
@@ -548,7 +548,7 @@ private void saveNameSystemSection(FileSummary.Builder summary)
         throws IOException {
       final FSNamesystem fsn = context.getSourceNamesystem();
       OutputStream out = sectionOutputStream;
-      BlockIdManager blockIdManager = fsn.getBlockIdManager();
+      BlockIdManager blockIdManager = fsn.getBlockManager().getBlockIdManager();
       NameSystemSection.Builder b = NameSystemSection.newBuilder()
           .setGenstampV1(blockIdManager.getGenerationStampV1())
           .setGenstampV1Limit(blockIdManager.getGenerationStampV1Limit())