MAPREDUCE-4907. TrackerDistributedCacheManager issues too many getFileStatus calls. (sandyr via tucu)

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

CHANGES.txt

@@ -154,6 +154,9 @@ Release 1.2.0 - unreleased
     HDFS-4320. Add a separate configuration for namenode rpc address instead
     of using fs.default.name. (Mostafa Elhemali via suresh)
 
+    MAPREDUCE-4907. TrackerDistributedCacheManager issues too many 
+    getFileStatus calls. (sandyr via tucu)
+
   OPTIMIZATIONS
 
     HDFS-2533. Backport: Remove needless synchronization on some FSDataSet

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

@@ -320,14 +320,15 @@ public class TrackerDistributedCacheManager {
    * @return true if the path in the uri is visible to all, false otherwise
    * @throws IOException
    */
-  static boolean isPublic(Configuration conf, URI uri) throws IOException {
+  static boolean isPublic(Configuration conf, URI uri,
+      Map<URI, FileStatus> statCache) throws IOException {
     FileSystem fs = FileSystem.get(uri, conf);
     Path current = new Path(uri.getPath());
     //the leaf level file should be readable by others
-    if (!checkPermissionOfOther(fs, current, FsAction.READ)) {
+    if (!checkPermissionOfOther(fs, current, FsAction.READ, statCache)) {
       return false;
     }
-    return ancestorsHaveExecutePermissions(fs, current.getParent());
+    return ancestorsHaveExecutePermissions(fs, current.getParent(), statCache);
   }
 
   /**
@@ -335,12 +336,12 @@ public class TrackerDistributedCacheManager {
    * permission set for all users (i.e. that other users can traverse
    * the directory heirarchy to the given path)
    */
-  static boolean ancestorsHaveExecutePermissions(FileSystem fs, Path path)
-    throws IOException {
+  static boolean ancestorsHaveExecutePermissions(FileSystem fs, Path path,
+      Map<URI, FileStatus> statCache) throws IOException {
     Path current = path;
     while (current != null) {
       //the subdirs in the path should have execute permissions for others
-      if (!checkPermissionOfOther(fs, current, FsAction.EXECUTE)) {
+      if (!checkPermissionOfOther(fs, current, FsAction.EXECUTE, statCache)) {
         return false;
       }
       current = current.getParent();
@@ -358,8 +359,8 @@ public class TrackerDistributedCacheManager {
    * @throws IOException
    */
   private static boolean checkPermissionOfOther(FileSystem fs, Path path,
-      FsAction action) throws IOException {
-    FileStatus status = fs.getFileStatus(path);
+      FsAction action, Map<URI, FileStatus> statCache) throws IOException {
+    FileStatus status = getFileStatus(fs, path, statCache);
     FsPermission perms = status.getPermission();
     FsAction otherAction = perms.getOtherAction();
     if (otherAction.implies(action)) {
@@ -711,26 +712,70 @@ public class TrackerDistributedCacheManager {
     return jobArchives.get(jobId);
   }
 
+  /**
+   * Gets the file status for the given URI.  If the URI is in the cache,
+   * returns it.  Otherwise, fetches it and adds it to the cache.
+   */
+  private static FileStatus getFileStatus(Configuration job, URI uri,
+      Map<URI, FileStatus> statCache) throws IOException {
+    FileStatus stat = statCache.get(uri);
+    if (stat == null) {
+      stat = DistributedCache.getFileStatus(job, uri);
+      statCache.put(uri, stat);
+    }
+    return stat;
+  }
+  
+  private static FileStatus getFileStatus(FileSystem fs, Path path,
+      Map<URI, FileStatus> statCache) throws IOException {
+    URI uri = path.toUri();
+    FileStatus stat = statCache.get(uri);
+    if (stat == null) {
+      stat = fs.getFileStatus(path);
+      statCache.put(uri, stat);
+    }
+    return stat;
+  }
+  
   /**
    * Determines timestamps of files to be cached, and stores those
-   * in the configuration.  This is intended to be used internally by JobClient
-   * after all cache files have been added.
+   * in the configuration. Determines the visibilities of the distributed cache
+   * files and archives. The visibility of a cache path is "public" if the leaf
+   * component has READ permissions for others, and the parent subdirs have 
+   * EXECUTE permissions for others.
    * 
    * This is an internal method!
    * 
+   * @param job
+   * @throws IOException
+   */
+  public static void determineTimestampsAndCacheVisibilities(Configuration job)
+  throws IOException {
+    Map<URI, FileStatus> statCache = new HashMap<URI, FileStatus>();
+    determineTimestamps(job, statCache);
+    determineCacheVisibilities(job, statCache);
+  }
+  
+  /**
+   * Determines timestamps of files to be cached, and stores those
+   * in the configuration.
+   * 
    * @param job Configuration of a job.
+   * @param statCache a cache of FileStatuses so that redundant remote
+   *    calls can be avoided
    * @throws IOException
    */
-  public static void determineTimestamps(Configuration job) throws IOException {
+  static void determineTimestamps(Configuration job,
+      Map<URI, FileStatus> statCache) throws IOException {
     URI[] tarchives = DistributedCache.getCacheArchives(job);
     if (tarchives != null) {
-      FileStatus status = DistributedCache.getFileStatus(job, tarchives[0]);
+      FileStatus status = getFileStatus(job, tarchives[0], statCache);
       StringBuffer archiveFileSizes = 
         new StringBuffer(String.valueOf(status.getLen()));      
       StringBuffer archiveTimestamps = 
         new StringBuffer(String.valueOf(status.getModificationTime()));
       for (int i = 1; i < tarchives.length; i++) {
-        status = DistributedCache.getFileStatus(job, tarchives[i]);
+        status = getFileStatus(job, tarchives[i], statCache);
         archiveFileSizes.append(",");
         archiveFileSizes.append(String.valueOf(status.getLen()));
         archiveTimestamps.append(",");
@@ -744,7 +789,7 @@ public class TrackerDistributedCacheManager {
   
     URI[] tfiles = DistributedCache.getCacheFiles(job);
     if (tfiles != null) {
-      FileStatus status = DistributedCache.getFileStatus(job, tfiles[0]);
+      FileStatus status = getFileStatus(job, tfiles[0], statCache);
       StringBuffer fileSizes = 
         new StringBuffer(String.valueOf(status.getLen()));      
       StringBuffer fileTimestamps = new StringBuffer(String.valueOf(
@@ -766,27 +811,29 @@ public class TrackerDistributedCacheManager {
    * has READ permissions for others, and the parent subdirs have 
    * EXECUTE permissions for others
    * @param job
+   * @param statCache a cache of FileStatuses so that redundant remote
+   *    calls can be avoided
    * @throws IOException
    */
-  public static void determineCacheVisibilities(Configuration job) 
-  throws IOException {
+  static void determineCacheVisibilities(Configuration job,
+      Map<URI, FileStatus> statCache) throws IOException {
     URI[] tarchives = DistributedCache.getCacheArchives(job);
     if (tarchives != null) {
       StringBuffer archiveVisibilities = 
-        new StringBuffer(String.valueOf(isPublic(job, tarchives[0])));
+        new StringBuffer(String.valueOf(isPublic(job, tarchives[0], statCache)));
       for (int i = 1; i < tarchives.length; i++) {
         archiveVisibilities.append(",");
-        archiveVisibilities.append(String.valueOf(isPublic(job, tarchives[i])));
+        archiveVisibilities.append(String.valueOf(isPublic(job, tarchives[i], statCache)));
       }
       setArchiveVisibilities(job, archiveVisibilities.toString());
     }
     URI[] tfiles = DistributedCache.getCacheFiles(job);
     if (tfiles != null) {
       StringBuffer fileVisibilities = 
-        new StringBuffer(String.valueOf(isPublic(job, tfiles[0])));
+        new StringBuffer(String.valueOf(isPublic(job, tfiles[0], statCache)));
       for (int i = 1; i < tfiles.length; i++) {
         fileVisibilities.append(",");
-        fileVisibilities.append(String.valueOf(isPublic(job, tfiles[i])));
+        fileVisibilities.append(String.valueOf(isPublic(job, tfiles[i], statCache)));
       }
       setFileVisibilities(job, fileVisibilities.toString());
     }

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

@@ -827,10 +827,9 @@ public class JobClient extends Configured implements MRConstants, Tool  {
     
     // First we check whether the cached archives and files are legal.
     TrackerDistributedCacheManager.validate(job);
-    //  set the timestamps of the archives and files
-    TrackerDistributedCacheManager.determineTimestamps(job);
-    //  set the public/private visibility of the archives and files
-    TrackerDistributedCacheManager.determineCacheVisibilities(job);
+    //  set the timestamps of the archives and files and set the
+    //  public/private visibility of the archives and files
+    TrackerDistributedCacheManager.determineTimestampsAndCacheVisibilities(job);
     // get DelegationTokens for cache files
     TrackerDistributedCacheManager.getDelegationTokens(job, 
                                                        job.getCredentials());

src/test/org/apache/hadoop/filecache/TestTrackerDistributedCacheManager.java

@@ -24,6 +24,8 @@ import java.io.FileOutputStream;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Random;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -108,7 +110,7 @@ public class TestTrackerDistributedCacheManager extends TestCase {
     assertTrue("Test root directory " + TEST_ROOT + " and all of its " +
                "parent directories must have a+x permissions",
                TrackerDistributedCacheManager.ancestorsHaveExecutePermissions(
-                 fs, new Path(TEST_ROOT.toString())));
+                 fs, new Path(TEST_ROOT.toString()), new HashMap<URI, FileStatus>()));
 
     // Prepare the tests' mapred-local-dir
     ROOT_MAPRED_LOCAL_DIR = new File(TEST_ROOT_DIR, "mapred/local");
@@ -187,8 +189,11 @@ public class TestTrackerDistributedCacheManager extends TestCase {
     DistributedCache.addCacheFile(firstCacheFile.toUri(), subConf);
     DistributedCache.addFileToClassPath(secondCacheFile, subConf, 
                                         FileSystem.get(subConf));
-    TrackerDistributedCacheManager.determineTimestamps(subConf);
-    TrackerDistributedCacheManager.determineCacheVisibilities(subConf);
+
+    Map<URI, FileStatus> statCache = new HashMap<URI, FileStatus>();
+    TrackerDistributedCacheManager.determineTimestamps(subConf, statCache);
+    TrackerDistributedCacheManager.determineCacheVisibilities(subConf, statCache);
+    assertEquals(2, statCache.size());
     // ****** End of imitating JobClient code
 
     Path jobFile = new Path(TEST_ROOT_DIR, "job.xml");
@@ -273,8 +278,7 @@ public class TestTrackerDistributedCacheManager extends TestCase {
     conf1.set("user.name", userName);
     DistributedCache.addCacheFile(secondCacheFile.toUri(), conf1);
     
-    TrackerDistributedCacheManager.determineTimestamps(conf1);
-    TrackerDistributedCacheManager.determineCacheVisibilities(conf1);
+    TrackerDistributedCacheManager.determineTimestampsAndCacheVisibilities(conf1);
 
     // Task localizing for first job
     JobID jobId = new JobID("jt", 1);
@@ -302,8 +306,7 @@ public class TestTrackerDistributedCacheManager extends TestCase {
     DistributedCache.addCacheFile(firstCacheDirPublic.toUri(), conf2);
     DistributedCache.addCacheFile(firstCacheDirPrivate.toUri(), conf2);
 
-    TrackerDistributedCacheManager.determineTimestamps(conf2);
-    TrackerDistributedCacheManager.determineCacheVisibilities(conf2);
+    TrackerDistributedCacheManager.determineTimestampsAndCacheVisibilities(conf2);
 
     // Task localizing for second job
     JobID job2Id = new JobID("jt", 2);
@@ -339,8 +342,7 @@ public class TestTrackerDistributedCacheManager extends TestCase {
     // add a file that is never localized
     DistributedCache.addCacheFile(thirdCacheFile.toUri(), conf3);
     
-    TrackerDistributedCacheManager.determineTimestamps(conf3);
-    TrackerDistributedCacheManager.determineCacheVisibilities(conf3);
+    TrackerDistributedCacheManager.determineTimestampsAndCacheVisibilities(conf3);
 
     // Task localizing for third job
     // localization for the "firstCacheFile" will fail.
@@ -379,7 +381,7 @@ public class TestTrackerDistributedCacheManager extends TestCase {
    * @throws LoginException
    */
   public void testPublicPrivateCache() 
-  throws IOException, LoginException, InterruptedException {
+  throws IOException, LoginException, InterruptedException, URISyntaxException {
     if (!canRun()) {
       return;
     }
@@ -404,8 +406,7 @@ public class TestTrackerDistributedCacheManager extends TestCase {
 
     DistributedCache.addCacheFile(cacheFile.toUri(), conf1);
     DistributedCache.addCacheArchive(cacheFile.toUri(), conf1);
-    TrackerDistributedCacheManager.determineTimestamps(conf1);
-    TrackerDistributedCacheManager.determineCacheVisibilities(conf1);
+    TrackerDistributedCacheManager.determineTimestampsAndCacheVisibilities(conf1);
     dumpState(conf1);
 
     TaskDistributedCacheManager handle = manager
@@ -491,7 +492,7 @@ public class TestTrackerDistributedCacheManager extends TestCase {
   }
   
   private void checkLocalizedPath(boolean visibility) 
-  throws IOException, LoginException, InterruptedException {
+  throws IOException, LoginException, InterruptedException, URISyntaxException {
     TrackerDistributedCacheManager manager = 
       new TrackerDistributedCacheManager(conf, taskController);
     String userName = getJobOwnerName();
@@ -506,8 +507,7 @@ public class TestTrackerDistributedCacheManager extends TestCase {
     Configuration conf1 = new Configuration(conf);
     conf1.set("user.name", userName);
     DistributedCache.addCacheFile(cacheFile.toUri(), conf1);
-    TrackerDistributedCacheManager.determineTimestamps(conf1);
-    TrackerDistributedCacheManager.determineCacheVisibilities(conf1);
+    TrackerDistributedCacheManager.determineTimestampsAndCacheVisibilities(conf1);
     dumpState(conf1);
 
     // Task localizing for job
@@ -894,8 +894,7 @@ public class TestTrackerDistributedCacheManager extends TestCase {
     createPrivateTempFile(thirdCacheFile);
     createPrivateTempFile(fourthCacheFile);
     DistributedCache.setCacheFiles(new URI[]{thirdCacheFile.toUri()}, conf2);
-    TrackerDistributedCacheManager.determineCacheVisibilities(conf2);
-    TrackerDistributedCacheManager.determineTimestamps(conf2);
+    TrackerDistributedCacheManager.determineTimestampsAndCacheVisibilities(conf2);
     stat = fs.getFileStatus(thirdCacheFile);
     CacheFile cfile3 = new CacheFile(thirdCacheFile.toUri(), 
             CacheFile.FileType.REGULAR, false, 
@@ -922,8 +921,7 @@ public class TestTrackerDistributedCacheManager extends TestCase {
     
     DistributedCache.setCacheFiles(new URI[]{fourthCacheFile.toUri()}, conf2);
     DistributedCache.setLocalFiles(conf2, thirdCacheFile.toUri().toString());
-    TrackerDistributedCacheManager.determineCacheVisibilities(conf2);
-    TrackerDistributedCacheManager.determineTimestamps(conf2);
+    TrackerDistributedCacheManager.determineTimestampsAndCacheVisibilities(conf2);
     Path fourthLocalCache = manager.getLocalCache(fourthCacheFile.toUri(), conf2, 
         TaskTracker.getPrivateDistributedCacheDir(userName),
         fs.getFileStatus(fourthCacheFile), false, 
@@ -1100,8 +1098,7 @@ public class TestTrackerDistributedCacheManager extends TestCase {
     Configuration subConf = new Configuration(myConf);
     subConf.set("user.name", userName);
     DistributedCache.addCacheFile(firstCacheFile.toUri(), subConf);
-    TrackerDistributedCacheManager.determineTimestamps(subConf);
-    TrackerDistributedCacheManager.determineCacheVisibilities(subConf);
+    TrackerDistributedCacheManager.determineTimestampsAndCacheVisibilities(subConf);
     // ****** End of imitating JobClient code
 
     // ****** Imitate TaskRunner code.
@@ -1150,8 +1147,7 @@ public class TestTrackerDistributedCacheManager extends TestCase {
     Configuration subConf2 = new Configuration(myConf);
     subConf2.set("user.name", userName);
     DistributedCache.addCacheFile(firstCacheFile.toUri(), subConf2);
-    TrackerDistributedCacheManager.determineTimestamps(subConf2);
-    TrackerDistributedCacheManager.determineCacheVisibilities(subConf2);
+    TrackerDistributedCacheManager.determineTimestampsAndCacheVisibilities(subConf2);
     
     handle =
       manager.newTaskDistributedCacheManager(new JobID("jt", 2), subConf2);