commit b81a27720032053edf2fb6ef4e115c7d65ba9f90

Author: Owen O'Malley <omalley@apache.org>
Date:   Sat Sep 18 16:25:13 2010 -0700

    fix dc issues with private caches


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

src/core/org/apache/hadoop/fs/LocalDirAllocator.java

@@ -124,8 +124,26 @@ public class LocalDirAllocator {
    */
   public Path getLocalPathForWrite(String pathStr, long size, 
       Configuration conf) throws IOException {
+	return getLocalPathForWrite(pathStr, size, conf, true);
+  }
+
+  /** Get a path from the local FS. Pass size as 
+   *  SIZE_UNKNOWN if not known apriori. We
+   *  round-robin over the set of disks (via the configured dirs) and return
+   *  the first complete path which has enough space 
+   *  @param pathStr the requested path (this will be created on the first 
+   *  available disk)
+   *  @param size the size of the file that is going to be written
+   *  @param conf the Configuration object
+   *  @param checkWrite ensure that the path is writable
+   *  @return the complete path to the file on a local disk
+   *  @throws IOException
+   */
+  public Path getLocalPathForWrite(String pathStr, long size, 
+                                   Configuration conf,
+                                   boolean checkWrite) throws IOException {
     AllocatorPerContext context = obtainContext(contextCfgItemName);
-    return context.getLocalPathForWrite(pathStr, size, conf);
+    return context.getLocalPathForWrite(pathStr, size, conf, checkWrite);
   }
   
   /** Get a path from the local FS for reading. We search through all the
@@ -266,18 +284,21 @@ public class LocalDirAllocator {
       }
     }
 
-    private Path createPath(String path) throws IOException {
+    private Path createPath(String path, 
+    		boolean checkWrite) throws IOException {
       Path file = new Path(new Path(localDirs[dirNumLastAccessed]),
                                     path);
-      //check whether we are able to create a directory here. If the disk
-      //happens to be RDONLY we will fail
-      try {
-        DiskChecker.checkDir(new File(file.getParent().toUri().getPath()));
-        return file;
-      } catch (DiskErrorException d) {
-        LOG.warn(StringUtils.stringifyException(d));
-        return null;
+      if (checkWrite) {
+    	//check whether we are able to create a directory here. If the disk
+    	//happens to be RDONLY we will fail
+    	try {
+          DiskChecker.checkDir(new File(file.getParent().toUri().getPath()));
+    	} catch (DiskErrorException d) {
+          LOG.warn(StringUtils.stringifyException(d));
+          return null;
+    	}
       }
+      return file;
     }
 
     /**
@@ -295,8 +316,10 @@ public class LocalDirAllocator {
      *  If size is not known, use roulette selection -- pick directories
      *  with probability proportional to their available space.
      */
-    public synchronized Path getLocalPathForWrite(String pathStr, long size, 
-        Configuration conf) throws IOException {
+    public synchronized 
+    Path getLocalPathForWrite(String pathStr, long size, 
+    	                      Configuration conf, boolean checkWrite
+    	                      ) throws IOException {
       confChanged(conf);
       int numDirs = localDirs.length;
       int numDirsSearched = 0;
@@ -328,7 +351,7 @@ public class LocalDirAllocator {
             dir++;
           }
           dirNumLastAccessed = dir;
-          returnPath = createPath(pathStr);
+          returnPath = createPath(pathStr, checkWrite);
           if (returnPath == null) {
             totalAvailable -= availableOnDisk[dir];
             availableOnDisk[dir] = 0; // skip this disk
@@ -339,7 +362,7 @@ public class LocalDirAllocator {
         while (numDirsSearched < numDirs && returnPath == null) {
           long capacity = dirDF[dirNumLastAccessed].getAvailable();
           if (capacity > size) {
-            returnPath = createPath(pathStr);
+            returnPath = createPath(pathStr, checkWrite);
           }
           dirNumLastAccessed++;
           dirNumLastAccessed = dirNumLastAccessed % numDirs; 
@@ -365,7 +388,7 @@ public class LocalDirAllocator {
         Configuration conf) throws IOException {
 
       // find an appropriate directory
-      Path path = getLocalPathForWrite(pathStr, size, conf);
+      Path path = getLocalPathForWrite(pathStr, size, conf, true);
       File dir = new File(path.getParent().toUri().getPath());
       String prefix = path.getName();
 

src/mapred/org/apache/hadoop/filecache/TrackerDistributedCacheManager.java

@@ -160,7 +160,7 @@ public class TrackerDistributedCacheManager {
         String cachePath = new Path (subDir, 
           new Path(uniqueString, makeRelative(cache, conf))).toString();
         localPath = lDirAllocator.getLocalPathForWrite(cachePath,
-          fileStatus.getLen(), trackerConf);
+          fileStatus.getLen(), trackerConf, isPublic);
         lcacheStatus = 
           new CacheStatus(new Path(localPath.toString().replace(cachePath, "")), 
                           localPath, new Path(subDir), uniqueString, 
@@ -479,7 +479,8 @@ public class TrackerDistributedCacheManager {
     } else {
       parchive = destination;
     }
-    if (!localFs.mkdirs(destination.getParent())) {
+    LOG.info("Creating " + destination + " with " + permission);
+    if (!localFs.mkdirs(destination.getParent(), permission)) {
       throw new IOException("Mkdirs failed to create directory " +
                             destination);
     }

src/mapred/org/apache/hadoop/mapred/JobLocalizer.java

@@ -307,7 +307,7 @@ public class JobLocalizer {
                                              boolean isArchive
                                              ) throws IOException {
     if (null == sources && null == dests && null == times && null == isPublic) {
-      return new long[0];
+      return null;
     }
     if (sources.length != dests.length ||
         sources.length != times.length ||
@@ -381,7 +381,9 @@ public class JobLocalizer {
     TaskTracker.resetNumTasksPerJvm(jConf);
 
     long[] sizes = downloadPrivateCache(jConf);
-    taskTracker.updatePrivateDistributedCacheSizes(jobid, sizes);
+    if (sizes != null) {
+      taskTracker.updatePrivateDistributedCacheSizes(jobid, sizes);
+    }
 
     // Create job-acls.xml file in job userlog dir and write the needed
     // info for authorization of users for viewing task logs of this job.