HDFS-8882. Erasure Coding: Use datablocks, parityblocks and cell size from ErasureCodingPolicy. Contributed by Vinayakumar B.

Change-Id: Ic56da0b426f47c63dac440aef6f5fc8554f6cf13

Author: Zhe Zhang

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/HdfsConstants.java

@@ -86,17 +86,6 @@ public final class HdfsConstants {
   public static final String CLIENT_NAMENODE_PROTOCOL_NAME =
       "org.apache.hadoop.hdfs.protocol.ClientProtocol";
 
-  /*
-   * These values correspond to the values used by the system default erasure
-   * coding policy.
-   * TODO: get these values from ec policy of the associated INodeFile
-   */
-
-  public static final byte NUM_DATA_BLOCKS = 6;
-  public static final byte NUM_PARITY_BLOCKS = 3;
-  // The chunk size for striped block which is used by erasure coding
-  public static final int BLOCK_STRIPED_CELL_SIZE = 64 * 1024;
-
   // Timeouts for communicating with DataNode for streaming writes/reads
   public static final int READ_TIMEOUT = 60 * 1000;
   public static final int READ_TIMEOUT_EXTENSION = 5 * 1000;

hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/hdfs.proto

@@ -536,6 +536,7 @@ message BlockWithLocationsProto {
 
   optional bytes indices = 5;
   optional uint32 dataBlockNum = 6;
+  optional uint32 cellSize = 7;
 }
 
 /**

hadoop-hdfs-project/hadoop-hdfs/CHANGES-HDFS-EC-7285.txt

@@ -446,4 +446,7 @@
     getAllErasureCodingPolicies. (Rakesh R via zhz)
 
     HDFS-8920. Erasure Coding: when recovering lost blocks, logs can be too
-    verbose and hurt performance. (Rui Li via Kai Zheng)
+    verbose and hurt performance. (Rui Li via Kai Zheng)
+
+    HDFS-8882. Erasure Coding: Use datablocks, parityblocks and cell size from
+    ErasureCodingPolicy (Vinayakumar B via zhz)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/StripedDataStreamer.java

@@ -18,25 +18,20 @@
 
 package org.apache.hadoop.hdfs;
 
-import static org.apache.hadoop.hdfs.protocol.HdfsConstants.BLOCK_STRIPED_CELL_SIZE;
-import static org.apache.hadoop.hdfs.protocol.HdfsConstants.NUM_DATA_BLOCKS;
-import static org.apache.hadoop.hdfs.protocol.HdfsConstants.NUM_PARITY_BLOCKS;
-
 import java.io.IOException;
 import java.io.InterruptedIOException;
 import java.util.concurrent.atomic.AtomicReference;
-
 import org.apache.hadoop.hdfs.DFSStripedOutputStream.Coordinator;
 import org.apache.hadoop.hdfs.DFSStripedOutputStream.MultipleBlockingQueue;
 import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
 import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
-import org.apache.hadoop.hdfs.protocol.HdfsConstants;
 import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
 import org.apache.hadoop.hdfs.protocol.LocatedBlock;
 import org.apache.hadoop.hdfs.protocol.LocatedStripedBlock;
 import org.apache.hadoop.hdfs.server.datanode.CachingStrategy;
 import org.apache.hadoop.hdfs.util.ByteArrayManager;
 import org.apache.hadoop.hdfs.util.StripedBlockUtil;
+import org.apache.hadoop.io.erasurecode.ECSchema;
 import org.apache.hadoop.util.DataChecksum;
 import org.apache.hadoop.util.Progressable;
 
@@ -109,6 +104,8 @@ private static void sleep(long ms, String op) throws InterruptedIOException {
   private final Coordinator coordinator;
   private final int index;
   private volatile boolean failed;
+  private final ECSchema schema;
+  private final int cellSize;
 
   StripedDataStreamer(HdfsFileStatus stat,
                       DFSClient dfsClient, String src,
@@ -120,6 +117,8 @@ private static void sleep(long ms, String op) throws InterruptedIOException {
         byteArrayManage, favoredNodes);
     this.index = index;
     this.coordinator = coordinator;
+    this.schema = stat.getErasureCodingPolicy().getSchema();
+    this.cellSize = stat.getErasureCodingPolicy().getCellSize();
   }
 
   int getIndex() {
@@ -135,7 +134,7 @@ boolean isFailed() {
   }
 
   private boolean isParityStreamer() {
-    return index >= NUM_DATA_BLOCKS;
+    return index >= schema.getNumDataUnits();
   }
 
   @Override
@@ -168,7 +167,7 @@ void populate() throws IOException {
         if (block != null) {
           // set numByte for the previous block group
           long bytes = 0;
-          for (int i = 0; i < NUM_DATA_BLOCKS; i++) {
+          for (int i = 0; i < schema.getNumDataUnits(); i++) {
             final ExtendedBlock b = coordinator.takeEndBlock(i);
             StripedBlockUtil.checkBlocks(index, block, i, b);
             bytes += b.getNumBytes();
@@ -183,15 +182,15 @@ void populate() throws IOException {
 
         final LocatedBlock lb = StripedDataStreamer.super.locateFollowingBlock(
             excludedNodes);
-        if (lb.getLocations().length < HdfsConstants.NUM_DATA_BLOCKS) {
+        if (lb.getLocations().length < schema.getNumDataUnits()) {
           throw new IOException(
               "Failed to get datablocks number of nodes from namenode: blockGroupSize= "
-                  + (HdfsConstants.NUM_DATA_BLOCKS + HdfsConstants.NUM_PARITY_BLOCKS)
+                  + (schema.getNumDataUnits() + schema.getNumParityUnits())
                   + ", blocks.length= " + lb.getLocations().length);
         }
-        final LocatedBlock[] blocks = StripedBlockUtil.parseStripedBlockGroup(
-            (LocatedStripedBlock)lb,
-            BLOCK_STRIPED_CELL_SIZE, NUM_DATA_BLOCKS, NUM_PARITY_BLOCKS);
+        final LocatedBlock[] blocks =
+            StripedBlockUtil.parseStripedBlockGroup((LocatedStripedBlock) lb,
+                cellSize, schema.getNumDataUnits(), schema.getNumParityUnits());
 
         for (int i = 0; i < blocks.length; i++) {
           StripedDataStreamer si = coordinator.getStripedDataStreamer(i);
@@ -233,9 +232,10 @@ void populate() throws IOException {
         final LocatedBlock updated = callUpdateBlockForPipeline(bg);
         final long newGS = updated.getBlock().getGenerationStamp();
         final LocatedBlock[] updatedBlks = StripedBlockUtil
-            .parseStripedBlockGroup((LocatedStripedBlock) updated,
-                BLOCK_STRIPED_CELL_SIZE, NUM_DATA_BLOCKS, NUM_PARITY_BLOCKS);
-        for (int i = 0; i < NUM_DATA_BLOCKS + NUM_PARITY_BLOCKS; i++) {
+            .parseStripedBlockGroup((LocatedStripedBlock) updated, cellSize,
+                schema.getNumDataUnits(), schema.getNumParityUnits());
+        for (int i = 0; i < schema.getNumDataUnits()
+            + schema.getNumParityUnits(); i++) {
           StripedDataStreamer si = coordinator.getStripedDataStreamer(i);
           if (si.isFailed()) {
             continue; // skipping failed data streamer
@@ -280,7 +280,7 @@ void populate() throws IOException {
         final ExtendedBlock bg = coordinator.getBlockGroup();
         final ExtendedBlock newBG = newBlock(bg, newGS);
 
-        final int n = NUM_DATA_BLOCKS + NUM_PARITY_BLOCKS;
+        final int n = schema.getNumDataUnits() + schema.getNumParityUnits();
         final DatanodeInfo[] newNodes = new DatanodeInfo[n];
         final String[] newStorageIDs = new String[n];
         for (int i = 0; i < n; i++) {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java

@@ -192,6 +192,7 @@ public static BlockWithLocationsProto convert(BlockWithLocations blk) {
       StripedBlockWithLocations sblk = (StripedBlockWithLocations) blk;
       builder.setIndices(PBHelperClient.getByteString(sblk.getIndices()));
       builder.setDataBlockNum(sblk.getDataBlockNum());
+      builder.setCellSize(sblk.getCellSize());
     }
     return builder.build();
   }
@@ -207,7 +208,7 @@ public static BlockWithLocations convert(BlockWithLocationsProto b) {
         PBHelperClient.convertStorageTypes(storageTypes, storageUuids.size()));
     if (b.hasIndices()) {
       blk = new StripedBlockWithLocations(blk, b.getIndices().toByteArray(),
-          (short) b.getDataBlockNum());
+          (short) b.getDataBlockNum(), b.getCellSize());
     }
     return blk;
   }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java

@@ -415,11 +415,14 @@ public static class DBlockStriped extends DBlock {
 
     final byte[] indices;
     final short dataBlockNum;
+    final int cellSize;
 
-    public DBlockStriped(Block block, byte[] indices, short dataBlockNum) {
+    public DBlockStriped(Block block, byte[] indices, short dataBlockNum,
+        int cellSize) {
       super(block);
       this.indices = indices;
       this.dataBlockNum = dataBlockNum;
+      this.cellSize = cellSize;
     }
 
     public DBlock getInternalBlock(StorageGroup storage) {
@@ -429,8 +432,8 @@ public DBlock getInternalBlock(StorageGroup storage) {
       }
       byte idxInGroup = indices[idxInLocs];
       long blkId = getBlock().getBlockId() + idxInGroup;
-      long numBytes = getInternalBlockLength(getNumBytes(),
-          HdfsConstants.BLOCK_STRIPED_CELL_SIZE, dataBlockNum, idxInGroup);
+      long numBytes = getInternalBlockLength(getNumBytes(), cellSize,
+          dataBlockNum, idxInGroup);
       Block blk = new Block(getBlock());
       blk.setBlockId(blkId);
       blk.setNumBytes(numBytes);
@@ -717,8 +720,8 @@ private long getBlockList() throws IOException {
           bytesReceived += sblkLocs.getBlock().getNumBytes() /
               sblkLocs.getDataBlockNum();
           block = new DBlockStriped(sblkLocs.getBlock(), sblkLocs.getIndices(),
-              sblkLocs.getDataBlockNum());
-        } else{
+              sblkLocs.getDataBlockNum(), sblkLocs.getCellSize());
+        } else {
           bytesReceived += blkLocs.getBlock().getNumBytes();
           block = new DBlock(blkLocs.getBlock());
         }

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

@@ -22,8 +22,6 @@
 import org.apache.hadoop.hdfs.util.StripedBlockUtil;
 import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy;
 
-import static org.apache.hadoop.hdfs.protocol.HdfsConstants.BLOCK_STRIPED_CELL_SIZE;
-
 /**
  * Subclass of {@link BlockInfo}, presenting a block group in erasure coding.
  *
@@ -65,6 +63,10 @@ public short getParityBlockNum() {
     return (short) ecPolicy.getNumParityUnits();
   }
 
+  public int getCellSize() {
+    return ecPolicy.getCellSize();
+  }
+
   /**
    * If the block is committed/completed and its length is less than a full
    * stripe, it returns the the number of actual data blocks.
@@ -73,7 +75,7 @@ public short getParityBlockNum() {
   public short getRealDataBlockNum() {
     if (isComplete() || getBlockUCState() == BlockUCState.COMMITTED) {
       return (short) Math.min(getDataBlockNum(),
-          (getNumBytes() - 1) / BLOCK_STRIPED_CELL_SIZE + 1);
+          (getNumBytes() - 1) / ecPolicy.getCellSize() + 1);
     } else {
       return getDataBlockNum();
     }
@@ -200,7 +202,7 @@ public long spaceConsumed() {
     // `getNumBytes` is the total of actual data block size.
     return StripedBlockUtil.spaceConsumedByStripedBlock(getNumBytes(),
         ecPolicy.getNumDataUnits(), ecPolicy.getNumParityUnits(),
-        BLOCK_STRIPED_CELL_SIZE);
+        ecPolicy.getCellSize());
     }
 
   @Override

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

@@ -92,7 +92,6 @@
 import org.apache.hadoop.hdfs.util.LightWeightLinkedSet;
 import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy;
 
-import static org.apache.hadoop.hdfs.protocol.HdfsConstants.BLOCK_STRIPED_CELL_SIZE;
 import static org.apache.hadoop.hdfs.util.StripedBlockUtil.getInternalBlockLength;
 
 import org.apache.hadoop.metrics2.util.MBeans;
@@ -2554,10 +2553,9 @@ private BlockToMarkCorrupt checkReplicaCorrupt(
               BlockIdManager.convertToStripedID(reported.getBlockId());
           BlockInfoStriped stripedBlock = (BlockInfoStriped) storedBlock;
           int reportedBlkIdx = BlockIdManager.getBlockIndex(reported);
-          wrongSize = reported.getNumBytes() !=
-              getInternalBlockLength(stripedBlock.getNumBytes(),
-                  BLOCK_STRIPED_CELL_SIZE,
-                  stripedBlock.getDataBlockNum(), reportedBlkIdx);
+          wrongSize = reported.getNumBytes() != getInternalBlockLength(
+              stripedBlock.getNumBytes(), stripedBlock.getCellSize(),
+              stripedBlock.getDataBlockNum(), reportedBlkIdx);
         } else {
           wrongSize = storedBlock.getNumBytes() != reported.getNumBytes();
         }
@@ -3413,7 +3411,7 @@ private long addBlock(BlockInfo block, List<BlockWithLocations> results) {
               (byte) blockStriped.getStorageBlockIndex(locations.get(i));
         }
         results.add(new StripedBlockWithLocations(blkWithLocs, indices,
-            blockStriped.getDataBlockNum()));
+            blockStriped.getDataBlockNum(), blockStriped.getCellSize()));
         // approximate size
         return block.getNumBytes() / blockStriped.getDataBlockNum();
       }else{

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/mover/Mover.java

@@ -185,7 +185,8 @@ DBlock newDBlock(LocatedBlock lb, List<MLocation> locations,
       for (int i = 0; i < indices.length; i++) {
         indices[i] = (byte) lsb.getBlockIndices()[i];
       }
-      db = new DBlockStriped(blk, indices, (short) ecPolicy.getNumDataUnits());
+      db = new DBlockStriped(blk, indices, (short) ecPolicy.getNumDataUnits(),
+          ecPolicy.getCellSize());
     } else {
       db = new DBlock(blk);
     }

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

@@ -204,9 +204,14 @@ static ValidateAddBlockResult validateAddBlock(
     clientMachine = pendingFile.getFileUnderConstructionFeature()
         .getClientMachine();
     isStriped = pendingFile.isStriped();
-    numTargets = isStriped ?
-        HdfsConstants.NUM_DATA_BLOCKS + HdfsConstants.NUM_PARITY_BLOCKS :
-        pendingFile.getFileReplication();
+    ErasureCodingPolicy ecPolicy = null;
+    if (isStriped) {
+      ecPolicy = FSDirErasureCodingOp.getErasureCodingPolicy(fsn, src);
+      numTargets = (short) (ecPolicy.getSchema().getNumDataUnits()
+          + ecPolicy.getSchema().getNumParityUnits());
+    } else {
+      numTargets = pendingFile.getFileReplication();
+    }
     storagePolicyID = pendingFile.getStoragePolicyID();
     return new ValidateAddBlockResult(blockSize, numTargets, storagePolicyID,
                                       clientMachine, isStriped);