HADOOP-2423. Code optimization in FSNamesystem.mkdirs.

(Tsz Wo (Nicholas), SZE via dhruba)



git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@636853 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -97,6 +97,9 @@ Trunk (unreleased changes)
     HADOOP-2399. Input key and value to combiner and reducer is reused.
     (Owen O'Malley via ddas). 
 
+    HADOOP-2423.  Code optimization in FSNamesystem.mkdirs.
+    (Tsz Wo (Nicholas), SZE via dhruba)
+
   BUG FIXES
 
     HADOOP-2195. '-mkdir' behaviour is now closer to Linux shell in case of

src/java/org/apache/hadoop/dfs/FSDirectory.java

@@ -636,50 +636,39 @@ class FSDirectory implements FSConstants {
   boolean mkdirs(String src, PermissionStatus permissions,
       boolean inheritPermission, long now) throws IOException {
     src = normalizePath(src);
+    String[] names = INode.getPathNames(src);
+    byte[][] components = INode.getPathComponents(names);
 
-    // Use this to collect all the dirs we need to construct
-    List<String> v = new ArrayList<String>();
-
-    // The dir itself
-    v.add(src);
+    synchronized(rootDir) {
+      INode[] inodes = new INode[components.length];
+      rootDir.getExistingPathINodes(components, inodes);
+
+      // find the index of the first null in inodes[]  
+      StringBuilder pathbuilder = new StringBuilder();
+      int i = 1;
+      for(; i < inodes.length && inodes[i] != null; i++) {
+        pathbuilder.append(Path.SEPARATOR + names[i]);
+        if (!inodes[i].isDirectory()) {
+          throw new FileNotFoundException("Parent path is not a directory: "
+              + pathbuilder);
+        }
+      }
 
-    // All its parents
-    Path parent = new Path(src).getParent();
-    while (parent != null) {
-      v.add(parent.toString());
-      parent = parent.getParent();
-    }
+      // create directories beginning from the first null index
+      for(; i < inodes.length; i++) {
+        pathbuilder.append(Path.SEPARATOR + names[i]);
+        String cur = pathbuilder.toString();
+  
+        inodes[i] = new INodeDirectory(permissions, now);
+        inodes[i].name = components[i];
+        INode inserted = ((INodeDirectory)inodes[i-1]).addChild(
+            inodes[i], inheritPermission || i != inodes.length-1);
 
-    // Now go backwards through list of dirs, creating along
-    // the way
-    int numElts = v.size();
-    for (int i = numElts - 1; i >= 0; i--) {
-      String cur = v.get(i);
-      try {
-        INode inserted = null;
-        synchronized (rootDir) {
-          inserted = rootDir.addNode(cur, 
-                             new INodeDirectory(permissions, now),
-                             inheritPermission || i != 0);
-          if (inserted != null) {
-            totalInodes++;
-          }
-        }
-        if (inserted != null) {
-          NameNode.stateChangeLog.debug("DIR* FSDirectory.mkdirs: "
-                                        +"created directory "+cur);
-          fsImage.getEditLog().logMkDir(cur, inserted);
-        } else { // otherwise cur exists, verify that it is a directory
-          if (!isDir(cur)) {
-            NameNode.stateChangeLog.debug("DIR* FSDirectory.mkdirs: "
-                                          +"path " + cur + " is not a directory ");
-            return false;
-          } 
-        }
-      } catch (FileNotFoundException e) {
-        NameNode.stateChangeLog.debug("DIR* FSDirectory.mkdirs: "
-                                      +"failed to create directory "+src);
-        return false;
+        assert inserted == inodes[i];
+        totalInodes++;
+        NameNode.stateChangeLog.debug(
+            "DIR* FSDirectory.mkdirs: created directory " + cur);
+        fsImage.getEditLog().logMkDir(cur, inserted);
       }
     }
     return true;

src/java/org/apache/hadoop/dfs/INode.java

@@ -232,21 +232,32 @@ abstract class INode implements Comparable<byte[]> {
    * a single path component.
    */
   static byte[][] getPathComponents(String path) {
-    if (path == null || !path.startsWith(Path.SEPARATOR)) {
-      return null;
-    }
-    if (Path.SEPARATOR.equals(path))  // root
+    return getPathComponents(getPathNames(path));
+  }
+
+  /** Convert strings to byte arrays for path components. */
+  static byte[][] getPathComponents(String[] strings) {
+    if (strings.length == 0) {
       return new byte[][]{null};
-    String[] strings = path.split(Path.SEPARATOR, -1);
-    int size = strings.length;
-    byte[][] bytes = new byte[size][];
-    for (int i = 0; i < size; i++)
+    }
+    byte[][] bytes = new byte[strings.length][];
+    for (int i = 0; i < strings.length; i++)
       bytes[i] = string2Bytes(strings[i]);
     return bytes;
   }
 
   /**
+   * Breaks file path into names.
+   * @param path
+   * @return array of names 
    */
+  static String[] getPathNames(String path) {
+    if (path == null || !path.startsWith(Path.SEPARATOR)) {
+      return null;
+    }
+    return path.split(Path.SEPARATOR);
+  }
+
   boolean removeNode() {
     if (parent == null) {
       return false;
@@ -423,7 +434,7 @@ class INodeDirectory extends INode {
    * @param existing INode array to fill with existing INodes
    * @return number of existing INodes in the path
    */
-  private int getExistingPathINodes(byte[][] components, INode[] existing) {
+  int getExistingPathINodes(byte[][] components, INode[] existing) {
     assert compareBytes(this.name, components[0]) == 0 :
       "Incorrect name " + getLocalName() + " expected " + components[0];
 
@@ -469,10 +480,21 @@ class INodeDirectory extends INode {
    * Add a child inode to the directory.
    * 
    * @param node INode to insert
+   * @param inheritPermission inherit permission from parent?
    * @return  null if the child with this name already exists; 
    *          inserted INode, otherwise
    */
-  private <T extends INode> T addChild(T node) {
+  <T extends INode> T addChild(final T node, boolean inheritPermission) {
+    if (inheritPermission) {
+      FsPermission p = getFsPermission();
+      //make sure the  permission has wx for the user
+      if (!p.getUserAction().implies(FsAction.WRITE_EXECUTE)) {
+        p = new FsPermission(p.getUserAction().or(FsAction.WRITE_EXECUTE),
+            p.getGroupAction(), p.getOtherAction());
+      }
+      node.setPermission(p);
+    }
+
     if (children == null) {
       children = new ArrayList<INode>(DEFAULT_FILES_PER_DIRECTORY);
     }
@@ -523,21 +545,10 @@ class INodeDirectory extends INode {
     if (!node.isDirectory()) {
       throw new FileNotFoundException("Parent path is not a directory: "+path);
     }
-    INodeDirectory parentNode = (INodeDirectory)node;
-
-    if (inheritPermission) {
-      FsPermission p = parentNode.getFsPermission();
-      //make sure the  permission has wx for the user
-      if (!p.getUserAction().implies(FsAction.WRITE_EXECUTE)) {
-        p = new FsPermission(p.getUserAction().or(FsAction.WRITE_EXECUTE),
-            p.getGroupAction(), p.getOtherAction());
-      }
-      newNode.setPermission(p);
-    }
 
     // insert into the parent children list
     newNode.name = pathComponents[pathLen-1];
-    return parentNode.addChild(newNode);
+    return ((INodeDirectory)node).addChild(newNode, inheritPermission);
   }
 
   /**