svn merge -c 1402599 from trunk for HDFS-4112. A few improvements on INodeDirectory include adding a utility method for casting; avoiding creation of new empty lists; cleaning up some code and rewriting some javadoc.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1402602 13f79535-47bb-0310-9956-ffa450edef68

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

@@ -79,6 +79,10 @@ Release 2.0.3-alpha - Unreleased
     HDFS-4107. Add utility methods for casting INode to INodeFile and
     INodeFileUnderConstruction. (szetszwo)
 
+    HDFS-4112. A few improvements on INodeDirectory include adding a utility
+    method for casting; avoiding creation of new empty lists; cleaning up 
+    some code and rewriting some javadoc. (szetszwo)
+
   OPTIMIZATIONS
 
   BUG FIXES 

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

@@ -696,7 +696,7 @@ public class FSDirectory implements Closeable {
         throw new FileAlreadyExistsException(error);
       }
       List<INode> children = dstInode.isDirectory() ? 
-          ((INodeDirectory) dstInode).getChildrenRaw() : null;
+          ((INodeDirectory) dstInode).getChildren() : null;
       if (children != null && children.size() != 0) {
         error = "rename cannot overwrite non empty destination directory "
             + dst;
@@ -1019,35 +1019,21 @@ public class FSDirectory implements Closeable {
     return true;
   }
   
-  /** Return if a directory is empty or not **/
+  /**
-  boolean isDirEmpty(String src) throws UnresolvedLinkException {
+   * @return true if the path is a non-empty directory; otherwise, return false.
-    boolean dirNotEmpty = true;
+   */
-    if (!isDir(src)) {
+  boolean isNonEmptyDirectory(String path) throws UnresolvedLinkException {
-      return true;
-    }
     readLock();
     try {
-      INode targetNode = rootDir.getNode(src, false);
+      final INode inode = rootDir.getNode(path, false);
-      assert targetNode != null : "should be taken care in isDir() above";
+      if (inode == null || !inode.isDirectory()) {
-      if (((INodeDirectory)targetNode).getChildren().size() != 0) {
+        //not found or not a directory
-        dirNotEmpty = false;
+        return false;
       }
+      return ((INodeDirectory)inode).getChildrenList().size() != 0;
     } finally {
       readUnlock();
     }
-    return dirNotEmpty;
-  }
-
-  boolean isEmpty() {
-    try {
-      return isDirEmpty("/");
-    } catch (UnresolvedLinkException e) {
-      if(NameNode.stateChangeLog.isDebugEnabled()) {
-        NameNode.stateChangeLog.debug("/ cannot be a symlink");
-      }
-      assert false : "/ cannot be a symlink";
-      return true;
-    }
   }
 
   /**
@@ -1171,7 +1157,7 @@ public class FSDirectory implements Closeable {
                 targetNode, needLocation)}, 0);
       }
       INodeDirectory dirInode = (INodeDirectory)targetNode;
-      List<INode> contents = dirInode.getChildren();
+      List<INode> contents = dirInode.getChildrenList();
       int startChild = dirInode.nextChild(startAfter);
       int totalNumChildren = contents.size();
       int numOfListing = Math.min(totalNumChildren-startChild, this.lsLimit);
@@ -1683,7 +1669,7 @@ public class FSDirectory implements Closeable {
       }
       if (maxDirItems != 0) {
         INodeDirectory parent = (INodeDirectory)pathComponents[pos-1];
-        int count = parent.getChildren().size();
+        int count = parent.getChildrenList().size();
         if (count >= maxDirItems) {
           throw new MaxDirectoryItemsExceededException(maxDirItems, count);
         }
@@ -1832,7 +1818,7 @@ public class FSDirectory implements Closeable {
      * INode. using 'parent' is not currently recommended. */
     nodesInPath.add(dir);
 
-    for (INode child : dir.getChildren()) {
+    for (INode child : dir.getChildrenList()) {
       if (child.isDirectory()) {
         updateCountForINodeWithQuota((INodeDirectory)child, 
                                      counts, nodesInPath);

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

@@ -246,10 +246,8 @@ class FSImageFormat {
    private int loadDirectory(DataInputStream in) throws IOException {
      String parentPath = FSImageSerialization.readString(in);
      FSDirectory fsDir = namesystem.dir;
-     INode parent = fsDir.rootDir.getNode(parentPath, true);
+     final INodeDirectory parent = INodeDirectory.valueOf(
-     if (parent == null || !parent.isDirectory()) {
+         fsDir.rootDir.getNode(parentPath, true), parentPath);
-       throw new IOException("Path " + parentPath + "is not a directory.");
-     }
 
      int numChildren = in.readInt();
      for(int i=0; i<numChildren; i++) {
@@ -259,7 +257,7 @@ class FSImageFormat {
        INode newNode = loadINode(in); // read rest of inode
 
        // add to parent
-       namesystem.dir.addToParent(localName, (INodeDirectory)parent, newNode, false);
+       namesystem.dir.addToParent(localName, parent, newNode, false);
      }
      return numChildren;
    }
@@ -532,7 +530,7 @@ class FSImageFormat {
     private void saveImage(ByteBuffer currentDirName,
                                   INodeDirectory current,
                                   DataOutputStream out) throws IOException {
-      List<INode> children = current.getChildrenRaw();
+      List<INode> children = current.getChildren();
       if (children == null || children.isEmpty())
         return;
       // print prefix (parent directory name)

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

@@ -2661,7 +2661,7 @@ public class FSNamesystem implements Namesystem, FSClusterStats,
       if (isInSafeMode()) {
         throw new SafeModeException("Cannot delete " + src, safeMode);
       }
-      if (!recursive && !dir.isDirEmpty(src)) {
+      if (!recursive && dir.isNonEmptyDirectory(src)) {
         throw new IOException(src + " is non empty");
       }
       if (enforcePermission && isPermissionEnabled) {

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

@@ -173,7 +173,7 @@ class FSPermissionChecker {
       INodeDirectory d = directories.pop();
       check(d, access);
 
-      for(INode child : d.getChildren()) {
+      for(INode child : d.getChildrenList()) {
         if (child.isDirectory()) {
           directories.push((INodeDirectory)child);
         }

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

@@ -17,7 +17,9 @@
  */
 package org.apache.hadoop.hdfs.server.namenode;
 
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 
 import org.apache.hadoop.classification.InterfaceAudience;
@@ -39,7 +41,8 @@ import com.google.common.primitives.SignedBytes;
  */
 @InterfaceAudience.Private
 abstract class INode implements Comparable<byte[]> {
-  /*
+  static final List<INode> EMPTY_LIST = Collections.unmodifiableList(new ArrayList<INode>());
+  /**
    *  The inode name is in java UTF8 encoding; 
    *  The name in HdfsFileStatus should keep the same encoding as this.
    *  if this encoding is changed, implicitly getFileInfo and listStatus in

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

@@ -18,6 +18,7 @@
 package org.apache.hadoop.hdfs.server.namenode;
 
 import java.io.FileNotFoundException;
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -32,6 +33,18 @@ import org.apache.hadoop.hdfs.protocol.UnresolvedPathException;
  * Directory INode class.
  */
 class INodeDirectory extends INode {
+  /** Cast INode to INodeDirectory. */
+  public static INodeDirectory valueOf(INode inode, String path
+      ) throws IOException {
+    if (inode == null) {
+      throw new IOException("Directory does not exist: " + path);
+    }
+    if (!inode.isDirectory()) {
+      throw new IOException("Path is not a directory: " + path);
+    }
+    return (INodeDirectory)inode; 
+  }
+
   protected static final int DEFAULT_FILES_PER_DIRECTORY = 5;
   final static String ROOT_NAME = "";
 
@@ -112,11 +125,11 @@ class INodeDirectory extends INode {
   }
 
   /**
-   * Return the INode of the last component in components, or null if the last
+   * @return the INode of the last component in components, or null if the last
    * component does not exist.
    */
-  private INode getNode(byte[][] components, boolean resolveLink) 
+  private INode getNode(byte[][] components, boolean resolveLink
-    throws UnresolvedLinkException {
+      ) throws UnresolvedLinkException {
     INode[] inode  = new INode[1];
     getExistingPathINodes(components, inode, resolveLink);
     return inode[0];
@@ -299,10 +312,7 @@ class INodeDirectory extends INode {
   <T extends INode> T addNode(String path, T newNode
       ) throws FileNotFoundException, UnresolvedLinkException  {
     byte[][] pathComponents = getPathComponents(path);        
-    if(addToParent(pathComponents, newNode,
+    return addToParent(pathComponents, newNode, true) == null? null: newNode;
-                    true) == null)
-      return null;
-    return newNode;
   }
 
   /**
@@ -326,10 +336,9 @@ class INodeDirectory extends INode {
     return parent;
   }
 
-  INodeDirectory getParent(byte[][] pathComponents)
+  INodeDirectory getParent(byte[][] pathComponents
-  throws FileNotFoundException, UnresolvedLinkException {
+      ) throws FileNotFoundException, UnresolvedLinkException {
-    int pathLen = pathComponents.length;
+    if (pathComponents.length < 2)  // add root
-    if (pathLen < 2)  // add root
       return null;
     // Gets the parent INode
     INode[] inodes  = new INode[2];
@@ -355,21 +364,15 @@ class INodeDirectory extends INode {
    * @throws  FileNotFoundException if parent does not exist or 
    *          is not a directory.
    */
-  INodeDirectory addToParent( byte[][] pathComponents,
+  INodeDirectory addToParent(byte[][] pathComponents, INode newNode,
-                              INode newNode,
+      boolean propagateModTime) throws FileNotFoundException, UnresolvedLinkException {
-                              boolean propagateModTime
+    if (pathComponents.length < 2) { // add root
-                            ) throws FileNotFoundException, 
-                                     UnresolvedLinkException {
-              
-    int pathLen = pathComponents.length;
-    if (pathLen < 2)  // add root
       return null;
-    newNode.name = pathComponents[pathLen-1];
+    }
+    newNode.name = pathComponents[pathComponents.length - 1];
     // insert into the parent children list
     INodeDirectory parent = getParent(pathComponents);
-    if(parent.addChild(newNode, propagateModTime) == null)
+    return parent.addChild(newNode, propagateModTime) == null? null: parent;
-      return null;
-    return parent;
   }
 
   @Override
@@ -415,11 +418,15 @@ class INodeDirectory extends INode {
   }
 
   /**
+   * @return an empty list if the children list is null;
+   *         otherwise, return the children list.
+   *         The returned list should not be modified.
    */
-  List<INode> getChildren() {
+  public List<INode> getChildrenList() {
-    return children==null ? new ArrayList<INode>() : children;
+    return children==null ? EMPTY_LIST : children;
   }
-  List<INode> getChildrenRaw() {
+  /** @return the children list which is possibly null. */
+  public List<INode> getChildren() {
     return children;
   }
 

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestINodeFile.java

@@ -234,6 +234,14 @@ public class TestINodeFile {
       } catch(FileNotFoundException fnfe) {
         assertTrue(fnfe.getMessage().contains("File does not exist"));
       }
+
+      //cast to INodeDirectory, should fail
+      try {
+        INodeDirectory.valueOf(from, path);
+        fail();
+      } catch(IOException ioe) {
+        assertTrue(ioe.getMessage().contains("Directory does not exist"));
+      }
     }
 
     {//cast from INodeFile
@@ -251,6 +259,14 @@ public class TestINodeFile {
       } catch(IOException ioe) {
         assertTrue(ioe.getMessage().contains("File is not under construction"));
       }
+
+      //cast to INodeDirectory, should fail
+      try {
+        INodeDirectory.valueOf(from, path);
+        fail();
+      } catch(IOException ioe) {
+        assertTrue(ioe.getMessage().contains("Path is not a directory"));
+      }
     }
 
     {//cast from INodeFileUnderConstruction
@@ -265,6 +281,14 @@ public class TestINodeFile {
       final INodeFileUnderConstruction u = INodeFileUnderConstruction.valueOf(
           from, path);
       assertTrue(u == from);
+
+      //cast to INodeDirectory, should fail
+      try {
+        INodeDirectory.valueOf(from, path);
+        fail();
+      } catch(IOException ioe) {
+        assertTrue(ioe.getMessage().contains("Path is not a directory"));
+      }
     }
 
     {//cast from INodeDirectory
@@ -285,6 +309,10 @@ public class TestINodeFile {
       } catch(FileNotFoundException fnfe) {
         assertTrue(fnfe.getMessage().contains("Path is not a file"));
       }
+
+      //cast to INodeDirectory, should success
+      final INodeDirectory d = INodeDirectory.valueOf(from, path);
+      assertTrue(d == from);
     }
   }
 }