HDFS-565. Introduce block committing logic during new block allocation and file close. Contributed by Konstantin Shvachko.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/hdfs/branches/HDFS-265@808718 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -26,6 +26,9 @@ Trunk (unreleased changes)
 
     HDFS-544. Add a "rbw" subdir to DataNode data directory. (hairong)
 
+    HDFS-565. Introduce block committing logic during new block allocation
+    and file close. (shv)
+
   IMPROVEMENTS
 
     HDFS-381. Remove blocks from DataNode maps when corresponding file

src/java/org/apache/hadoop/hdfs/DFSClient.java

@@ -2432,6 +2432,8 @@ public class DFSClient implements FSConstants, java.io.Closeable {
                   } catch (InterruptedException  e) {
                   }
                 }
+                // update block length
+                block.setNumBytes(offsetInBlock + one.dataPos - one.dataStart);
               }
               
               if (ackQueue.isEmpty()) { // done receiving all acks
@@ -2442,6 +2444,9 @@ public class DFSClient implements FSConstants, java.io.Closeable {
                 blockStream.writeInt(0);
                 blockStream.flush();
               }
+            } else {
+              // update block length
+              block.setNumBytes(offsetInBlock + one.dataPos - one.dataStart);
             }
             if (LOG.isDebugEnabled()) {
               LOG.debug("DataStreamer block " + block +
@@ -2801,6 +2806,7 @@ public class DFSClient implements FSConstants, java.io.Closeable {
           if (!success) {
             LOG.info("Abandoning block " + block);
             namenode.abandonBlock(block, src, clientName);
+            block = null;
 
             // Connection failed.  Let's wait a little bit and retry
             retry = true;
@@ -2908,7 +2914,7 @@ public class DFSClient implements FSConstants, java.io.Closeable {
           long localstart = System.currentTimeMillis();
           while (true) {
             try {
-              return namenode.addBlock(src, clientName);
+              return namenode.addBlock(src, clientName, block);
             } catch (RemoteException e) {
               IOException ue = 
                 e.unwrapRemoteException(FileNotFoundException.class,
@@ -2997,6 +3003,10 @@ public class DFSClient implements FSConstants, java.io.Closeable {
         processDatanodeError(true, true);
       }
 
+      Block getBlock() {
+        return block;
+      }
+
       DatanodeInfo[] getNodes() {
         return nodes;
       }
@@ -3364,8 +3374,10 @@ public class DFSClient implements FSConstants, java.io.Closeable {
         }
 
         flushInternal();             // flush all data to Datanodes
+        // get last block before destroying the streamer
+        Block lastBlock = streamer.getBlock();
         closeThreads(false);
-        completeFile();
+        completeFile(lastBlock);
         leasechecker.remove(src);
       } finally {
         closed = true;
@@ -3374,11 +3386,11 @@ public class DFSClient implements FSConstants, java.io.Closeable {
 
     // should be called holding (this) lock since setTestFilename() may 
     // be called during unit tests
-    private void completeFile() throws IOException {
+    private void completeFile(Block last) throws IOException {
       long localstart = System.currentTimeMillis();
       boolean fileComplete = false;
       while (!fileComplete) {
-        fileComplete = namenode.complete(src, clientName);
+        fileComplete = namenode.complete(src, clientName, last);
         if (!fileComplete) {
           if (!clientRunning ||
                 (hdfsTimeout > 0 &&

src/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java

@@ -43,9 +43,9 @@ public interface ClientProtocol extends VersionedProtocol {
    * Compared to the previous version the following changes have been introduced:
    * (Only the latest change is reflected.
    * The log of historical changes can be retrieved from the svn).
-   * 45: add create flag for create command, see Hadoop-5438
+   * 46: add Block parameter to addBlock() and complete()
    */
-  public static final long versionID = 45L;
+  public static final long versionID = 46L;
   
   ///////////////////////////////////////
   // File contents
@@ -85,8 +85,8 @@ public interface ClientProtocol extends VersionedProtocol {
    * {@link #rename(String, String)} it until the file is completed
    * or explicitly as a result of lease expiration.
    * <p>
-   * Blocks have a maximum size.  Clients that intend to
-   * create multi-block files must also use {@link #addBlock(String, String)}.
+   * Blocks have a maximum size.  Clients that intend to create
+   * multi-block files must also use {@link #addBlock(String, String, Block)}.
    *
    * @param src path of the file being created.
    * @param masked masked permission.
@@ -177,9 +177,14 @@ public interface ClientProtocol extends VersionedProtocol {
    * addBlock() allocates a new block and datanodes the block data
    * should be replicated to.
    * 
+   * addBlock() also commits the previous block by reporting
+   * to the name-node the actual generation stamp and the length
+   * of the block that the client has transmitted to data-nodes.
+   * 
    * @return LocatedBlock allocated block information.
    */
-  public LocatedBlock addBlock(String src, String clientName) throws IOException;
+  public LocatedBlock addBlock(String src, String clientName,
+                               Block previous) throws IOException;
 
   /**
    * The client is done writing data to the given filename, and would 
@@ -187,13 +192,18 @@ public interface ClientProtocol extends VersionedProtocol {
    *
    * The function returns whether the file has been closed successfully.
    * If the function returns false, the caller should try again.
+   * 
+   * close() also commits the last block of the file by reporting
+   * to the name-node the actual generation stamp and the length
+   * of the block that the client has transmitted to data-nodes.
    *
    * A call to complete() will not return true until all the file's
    * blocks have been replicated the minimum number of times.  Thus,
    * DataNode failures may cause a client to call complete() several
    * times before succeeding.
    */
-  public boolean complete(String src, String clientName) throws IOException;
+  public boolean complete(String src, String clientName,
+                          Block last) throws IOException;
 
   /**
    * The client wants to report corrupted blocks (blocks with specified

src/java/org/apache/hadoop/hdfs/server/namenode/BlockInfo.java

@@ -85,7 +85,7 @@ class BlockInfo extends Block {
     assert index >= 0 && index*3+2 < triplets.length : "Index is out of bound";
     BlockInfo info = (BlockInfo)triplets[index*3+2];
     assert info == null || 
-        BlockInfo.class.getName().equals(info.getClass().getName()) : 
+        info.getClass().getName().startsWith(BlockInfo.class.getName()) : 
               "BlockInfo is expected at " + index*3;
     return info;
   }
@@ -295,9 +295,18 @@ class BlockInfo extends Block {
    * 
    * @return BlockInfoUnderConstruction -  an under construction block.
    */
-  BlockInfoUnderConstruction convertToBlockUnderConstruction() {
-    assert !isUnderConstruction() : "the block is already under construction";
-    return new BlockInfoUnderConstruction(this, getINode().getReplication());
+  BlockInfoUnderConstruction convertToBlockUnderConstruction(
+      BlockUCState s, DatanodeDescriptor[] targets) {
+    if(!isUnderConstruction()) {
+      return new BlockInfoUnderConstruction(
+          this, getINode().getReplication(), s, targets);
+    }
+    // the block is already under construction
+    BlockInfoUnderConstruction ucBlock = (BlockInfoUnderConstruction)this;
+    ucBlock.setBlockUCState(s);
+    ucBlock.setLocations(targets);
+    ucBlock.setLastRecoveryTime(0);
+    return ucBlock;
   }
 
   @Override

src/java/org/apache/hadoop/hdfs/server/namenode/BlockInfoUnderConstruction.java

@@ -104,13 +104,17 @@ class BlockInfoUnderConstruction extends BlockInfo {
 
   /**
    * Return the state of the block under construction.
-   * @see {@link BlockUCState}
+   * @see BlockUCState
    */
   @Override // BlockInfo
   BlockUCState getBlockUCState() {
     return blockUCState;
   }
 
+  void setBlockUCState(BlockUCState s) {
+    blockUCState = s;
+  }
+
   /**
    * Commit block's length and generation stamp as reported by the client.
    * Set block state to {@link BlockUCState#COMMITTED}.

src/java/org/apache/hadoop/hdfs/server/namenode/BlockManager.java

@@ -240,9 +240,14 @@ public class BlockManager {
    */
   void commitLastBlock(INodeFileUnderConstruction fileINode, 
                        Block commitBlock) throws IOException {
+    if(commitBlock == null)
+      return; // not committing, this is a block allocation retry
     BlockInfoUnderConstruction lastBlock = fileINode.getLastBlock();
     if(lastBlock == null)
-      return;
+      return; // no blocks in file yet
+    assert lastBlock.getNumBytes() <= commitBlock.getNumBytes() :
+      "commitBlock length is less than the stored one "
+      + commitBlock.getNumBytes() + " vs. " + lastBlock.getNumBytes();
     lastBlock.commitBlock(commitBlock);
 
     // complete the penultimate block
@@ -274,17 +279,18 @@ public class BlockManager {
   }
 
   /**
-   * Convert the last block of the file to an under constroction block.
+   * Convert the last block of the file to an under construction block.
    * @param fileINode file
+   * @param targets data-nodes that will form the pipeline for this block
    */
-  void convertLastBlockToUnderConstruction(INodeFile fileINode)
-  throws IOException {
+  void convertLastBlockToUnderConstruction(
+      INodeFileUnderConstruction fileINode,
+      DatanodeDescriptor[] targets) throws IOException {
     BlockInfo oldBlock = fileINode.getLastBlock();
-    if(oldBlock == null || oldBlock.isUnderConstruction())
+    if(oldBlock == null)
       return;
     BlockInfoUnderConstruction ucBlock =
-      oldBlock.convertToBlockUnderConstruction();
-    fileINode.setBlock(fileINode.numBlocks()-1, ucBlock);
+      fileINode.setLastBlock(oldBlock, targets);
     blocksMap.replaceBlock(ucBlock);
   }
 

src/java/org/apache/hadoop/hdfs/server/namenode/FSDirectory.java

@@ -279,10 +279,6 @@ class FSDirectory implements Closeable {
       INodeFileUnderConstruction fileINode = 
         (INodeFileUnderConstruction)inodes[inodes.length-1];
 
-      // commit the last block and complete the penultimate block
-      // SHV !!! second parameter should be a block reported by client
-      getBlockManager().commitLastBlock(fileINode, fileINode.getLastBlock());
-
       // check quota limits and updated space consumed
       updateCount(inodes, inodes.length-1, 0, 
                   fileINode.getPreferredBlockSize()*fileINode.getReplication());

src/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -956,7 +956,6 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
         // Recreate in-memory lease record.
         //
         INodeFile node = (INodeFile) myFile;
-        blockManager.convertLastBlockToUnderConstruction(node);
         INodeFileUnderConstruction cons = new INodeFileUnderConstruction(
                                         node.getLocalNameBytes(),
                                         node.getReplication(),
@@ -1018,7 +1017,7 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
     LocatedBlock lb = null;
     synchronized (this) {
       INodeFileUnderConstruction file = (INodeFileUnderConstruction)dir.getFileINode(src);
-      BlockInfoUnderConstruction lastBlock = file.getLastBlock();
+      BlockInfo lastBlock = file.getLastBlock();
       if (lastBlock != null) {
         assert lastBlock == blockManager.getStoredBlock(lastBlock) :
           "last block of the file is not in blocksMap";
@@ -1030,7 +1029,7 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
             targets[i].removeBlock(lastBlock);
           }
           // convert last block to under-construction and set its locations
-          file.setLastBlock(lastBlock, targets);
+          blockManager.convertLastBlockToUnderConstruction(file, targets);
 
           lb = new LocatedBlock(lastBlock, targets, 
                                 fileLength-lastBlock.getNumBytes());
@@ -1081,7 +1080,8 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
    * client to "try again later".
    */
   public LocatedBlock getAdditionalBlock(String src, 
-                                         String clientName
+                                         String clientName,
+                                         Block previous
                                          ) throws IOException {
     long fileLength, blockSize;
     int replication;
@@ -1101,6 +1101,9 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
 
       INodeFileUnderConstruction pendingFile  = checkLease(src, clientName);
 
+      // commit the last block and complete the penultimate block
+      blockManager.commitLastBlock(pendingFile, previous);
+
       //
       // If we fail this, bad things happen!
       //
@@ -1215,15 +1218,18 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
     COMPLETE_SUCCESS
   }
   
-  public CompleteFileStatus completeFile(String src, String holder) throws IOException {
-    CompleteFileStatus status = completeFileInternal(src, holder);
+  public CompleteFileStatus completeFile(String src,
+                                         String holder,
+                                         Block last) throws IOException {
+    CompleteFileStatus status = completeFileInternal(src, holder, last);
     getEditLog().logSync();
     return status;
   }
 
-
-  private synchronized CompleteFileStatus completeFileInternal(String src, 
-                                                String holder) throws IOException {
+  private synchronized CompleteFileStatus completeFileInternal(
+                                            String src, 
+                                            String holder,
+                                            Block last) throws IOException {
     NameNode.stateChangeLog.debug("DIR* NameSystem.completeFile: " + src + " for " + holder);
     if (isInSafeMode())
       throw new SafeModeException("Cannot complete file " + src, safeMode);
@@ -1244,7 +1250,12 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
                                      ("from " + pendingFile.getClientMachine()))
                                   );                      
       return CompleteFileStatus.OPERATION_FAILED;
-    } else if (!checkFileProgress(pendingFile, true)) {
+    } 
+
+    // commit the last block and complete the penultimate block
+    blockManager.commitLastBlock(pendingFile, last);
+
+    if (!checkFileProgress(pendingFile, true)) {
       return CompleteFileStatus.STILL_WAITING;
     }
 
@@ -1578,14 +1589,11 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
     leaseManager.renewLease(lease);
   }
 
-  private void finalizeINodeFileUnderConstruction(String src,
+  private void finalizeINodeFileUnderConstruction(
+      String src,
       INodeFileUnderConstruction pendingFile) throws IOException {
     leaseManager.removeLease(pendingFile.clientName, src);
 
-    // commit the last block and complete the penultimate block
-    // SHV !!! second parameter should be a block reported by client
-    blockManager.commitLastBlock(pendingFile, pendingFile.getLastBlock());
-
     // The file is no longer pending.
     // Create permanent INode, update blocks
     INodeFile newFile = pendingFile.convertToInodeFile();
@@ -1671,6 +1679,9 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean {
       LOG.info("commitBlockSynchronization(" + lastblock + ") successful");
       return;
     }
+
+    // commit the last block and complete the penultimate block
+    blockManager.commitLastBlock(pendingFile, lastblock);
     
     //remove lease, close file
     finalizeINodeFileUnderConstruction(src, pendingFile);

src/java/org/apache/hadoop/hdfs/server/namenode/INodeFileUnderConstruction.java

@@ -21,6 +21,7 @@ import java.io.IOException;
 
 import org.apache.hadoop.fs.permission.PermissionStatus;
 import org.apache.hadoop.hdfs.protocol.Block;
+import org.apache.hadoop.hdfs.server.common.HdfsConstants.BlockUCState;
 
 
 class INodeFileUnderConstruction extends INodeFile {
@@ -117,19 +118,18 @@ class INodeFileUnderConstruction extends INodeFile {
    * Convert the last block of the file to an under-construction block.
    * Set its locations.
    */
-  void setLastBlock(BlockInfo lastBlock, DatanodeDescriptor[] targets)
+  BlockInfoUnderConstruction setLastBlock(BlockInfo lastBlock,
+                                          DatanodeDescriptor[] targets)
   throws IOException {
     if (blocks == null || blocks.length == 0) {
       throw new IOException("Trying to update non-existant block. " +
       		"File is empty.");
     }
-    BlockInfoUnderConstruction ucBlock;
-    if(lastBlock.isUnderConstruction())
-      ucBlock = (BlockInfoUnderConstruction)lastBlock;
-    else
-      ucBlock = new BlockInfoUnderConstruction(lastBlock, getReplication());
-    ucBlock.setLocations(targets);
-    ucBlock.setLastRecoveryTime(0);
-    blocks[blocks.length - 1] = ucBlock;
+    BlockInfoUnderConstruction ucBlock =
+      lastBlock.convertToBlockUnderConstruction(
+          BlockUCState.UNDER_CONSTRUCTION, targets);
+    ucBlock.setINode(this);
+    setBlock(numBlocks()-1, ucBlock);
+    return ucBlock;
   }
 }

src/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java

@@ -603,11 +603,12 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
 
   /**
    */
-  public LocatedBlock addBlock(String src, 
-                               String clientName) throws IOException {
+  public LocatedBlock addBlock(String src, String clientName,
+                               Block previous) throws IOException {
     stateChangeLog.debug("*BLOCK* NameNode.addBlock: file "
                          +src+" for "+clientName);
-    LocatedBlock locatedBlock = namesystem.getAdditionalBlock(src, clientName);
+    LocatedBlock locatedBlock = 
+      namesystem.getAdditionalBlock(src, clientName, previous);
     if (locatedBlock != null)
       myMetrics.numAddBlockOps.inc();
     return locatedBlock;
@@ -626,9 +627,11 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
   }
 
   /** {@inheritDoc} */
-  public boolean complete(String src, String clientName) throws IOException {
+  public boolean complete(String src, String clientName,
+                          Block last) throws IOException {
     stateChangeLog.debug("*DIR* NameNode.complete: " + src + " for " + clientName);
-    CompleteFileStatus returnCode = namesystem.completeFile(src, clientName);
+    CompleteFileStatus returnCode =
+      namesystem.completeFile(src, clientName, last);
     if (returnCode == CompleteFileStatus.STILL_WAITING) {
       return false;
     } else if (returnCode == CompleteFileStatus.COMPLETE_SUCCESS) {

src/test/hdfs/org/apache/hadoop/hdfs/TestDFSClientRetries.java

@@ -136,7 +136,7 @@ public class TestDFSClientRetries extends TestCase {
       return versionID;
     }
 
-    public LocatedBlock addBlock(String src, String clientName)
+    public LocatedBlock addBlock(String src, String clientName, Block previous)
     throws IOException
     {
       num_calls++;
@@ -167,7 +167,7 @@ public class TestDFSClientRetries extends TestCase {
 
     public void abandonBlock(Block b, String src, String holder) throws IOException {}
 
-    public boolean complete(String src, String clientName) throws IOException { return false; }
+    public boolean complete(String src, String clientName, Block last) throws IOException { return false; }
 
     public void reportBadBlocks(LocatedBlock[] blocks) throws IOException {}
 

src/test/hdfs/org/apache/hadoop/hdfs/TestFileCreation.java

@@ -416,9 +416,9 @@ public class TestFileCreation extends junit.framework.TestCase {
       System.out.println("testFileCreationError2: "
           + "The file has " + locations.locatedBlockCount() + " blocks.");
 
-      // add another block to the file
+      // add one block to the file
       LocatedBlock location = client.getNamenode().addBlock(file1.toString(), 
-          client.clientName);
+          client.clientName, null);
       System.out.println("testFileCreationError2: "
           + "Added block " + location.getBlock());
 

src/test/hdfs/org/apache/hadoop/hdfs/server/namenode/NNThroughputBenchmark.java

@@ -519,7 +519,8 @@ public class NNThroughputBenchmark {
               .of(CreateFlag.OVERWRITE)), replication, BLOCK_SIZE);
       long end = System.currentTimeMillis();
       for(boolean written = !closeUponCreate; !written; 
-        written = nameNode.complete(fileNames[daemonId][inputIdx], clientName));
+        written = nameNode.complete(fileNames[daemonId][inputIdx],
+                                    clientName, null));
       return end-start;
     }
 
@@ -889,8 +890,8 @@ public class NNThroughputBenchmark {
         nameNode.create(fileName, FsPermission.getDefault(), clientName,
             new EnumSetWritable<CreateFlag>(EnumSet.of(CreateFlag.OVERWRITE)), replication,
             BLOCK_SIZE);
-        addBlocks(fileName, clientName);
-        nameNode.complete(fileName, clientName);
+        Block lastBlock = addBlocks(fileName, clientName);
+        nameNode.complete(fileName, clientName, lastBlock);
       }
       // prepare block reports
       for(int idx=0; idx < nrDatanodes; idx++) {
@@ -898,9 +899,12 @@ public class NNThroughputBenchmark {
       }
     }
 
-    private void addBlocks(String fileName, String clientName) throws IOException {
+    private Block addBlocks(String fileName, String clientName)
+    throws IOException {
+      Block prevBlock = null;
       for(int jdx = 0; jdx < blocksPerFile; jdx++) {
-        LocatedBlock loc = nameNode.addBlock(fileName, clientName);
+        LocatedBlock loc = nameNode.addBlock(fileName, clientName, prevBlock);
+        prevBlock = loc.getBlock();
         for(DatanodeInfo dnInfo : loc.getLocations()) {
           int dnIdx = Arrays.binarySearch(datanodes, dnInfo.getName());
           datanodes[dnIdx].addBlock(loc.getBlock());
@@ -910,6 +914,7 @@ public class NNThroughputBenchmark {
               new String[] {""});
         }
       }
+      return prevBlock;
     }
 
     /**