Author: omalley
Date: Fri Mar 4 04:24:33 2011
New Revision: 1077527
URL: http://svn.apache.org/viewvc?rev=1077527&view=rev
Log:
commit d1d1713459678e945d527ee318c3076bf176a4c8
Author: Arun C Murthy <[email protected]>
Date: Fri Jul 2 16:50:55 2010 -0700
MAPREDUCE-1914. Ensure unique sub-directories for artifacts in the
DistributedCache are cleaned up. Contributed by Dick King.
+++ b/YAHOO-CHANGES.txt
+ MAPREDUCE-1914. Ensure unique sub-directories for artifacts in the
+ DistributedCache are cleaned up. (Dick King via acmurthy)
+
Modified:
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/filecache/TrackerDistributedCacheManager.java
hadoop/common/branches/branch-0.20-security-patches/src/test/org/apache/hadoop/filecache/TestTrackerDistributedCacheManager.java
Modified:
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/filecache/TrackerDistributedCacheManager.java
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/filecache/TrackerDistributedCacheManager.java?rev=1077527&r1=1077526&r2=1077527&view=diff
==============================================================================
---
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/filecache/TrackerDistributedCacheManager.java
(original)
+++
hadoop/common/branches/branch-0.20-security-patches/src/mapred/org/apache/hadoop/filecache/TrackerDistributedCacheManager.java
Fri Mar 4 04:24:33 2011
@@ -23,6 +23,8 @@ import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashSet;
import java.util.Iterator;
+import java.util.List;
+import java.util.LinkedList;
import java.util.Map;
import java.util.Random;
import java.util.Set;
@@ -146,7 +148,10 @@ public class TrackerDistributedCacheMana
lcacheStatus = cachedArchives.get(key);
if (lcacheStatus == null) {
// was never localized
- String uniqueString = String.valueOf(random.nextLong());
+ String uniqueString
+ = (String.valueOf(random.nextLong())
+ + "_" + cache.hashCode()
+ + "_" + (confFileStamp % Integer.MAX_VALUE));
String cachePath = new Path (subDir,
new Path(uniqueString, makeRelative(cache, conf))).toString();
Path localPath = lDirAllocator.getLocalPathForWrite(cachePath,
@@ -194,7 +199,7 @@ public class TrackerDistributedCacheMana
if (allowedCacheSize < size || allowedCacheSubdirs < numberSubdirs) {
// try some cache deletions
- deleteCache(conf);
+ compactCache(conf);
}
initSuccessful = true;
return localizedPath;
@@ -248,8 +253,8 @@ public class TrackerDistributedCacheMana
// To delete the caches which have a refcount of zero
- private void deleteCache(Configuration conf) throws IOException {
- Set<CacheStatus> deleteSet = new HashSet<CacheStatus>();
+ private void compactCache(Configuration conf) throws IOException {
+ List<CacheStatus> deleteList = new LinkedList<CacheStatus>();
// try deleting cache Status with refcount of zero
synchronized (cachedArchives) {
for (Iterator<String> it = cachedArchives.keySet().iterator();
@@ -262,19 +267,30 @@ public class TrackerDistributedCacheMana
if (lcacheStatus.refcount == 0) {
// delete this cache entry from the global list
// and mark the localized file for deletion
- deleteSet.add(lcacheStatus);
+ deleteList.add(lcacheStatus);
it.remove();
}
}
}
// do the deletion, after releasing the global lock
- for (CacheStatus lcacheStatus : deleteSet) {
+ for (CacheStatus lcacheStatus : deleteList) {
synchronized (lcacheStatus) {
- FileSystem.getLocal(conf).delete(lcacheStatus.localizedLoadPath, true);
+ FileSystem localFS = FileSystem.getLocal(conf);
+
+ Path potentialDeletee = lcacheStatus.localizedLoadPath;
+
+ localFS.delete(potentialDeletee, true);
// Update the maps baseDirSize and baseDirNumberSubDir
- LOG.info("Deleted path " + lcacheStatus.localizedLoadPath);
+ LOG.info("Deleted path " + potentialDeletee);
+
+ try {
+ localFS.delete(lcacheStatus.getLocalizedUniqueDir(), true);
+ } catch (IOException e) {
+ LOG.warn("Could not delete distributed cache empty directory "
+ + lcacheStatus.getLocalizedUniqueDir());
+ }
deleteCacheInfoUpdate(lcacheStatus);
}
Modified:
hadoop/common/branches/branch-0.20-security-patches/src/test/org/apache/hadoop/filecache/TestTrackerDistributedCacheManager.java
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-0.20-security-patches/src/test/org/apache/hadoop/filecache/TestTrackerDistributedCacheManager.java?rev=1077527&r1=1077526&r2=1077527&view=diff
==============================================================================
---
hadoop/common/branches/branch-0.20-security-patches/src/test/org/apache/hadoop/filecache/TestTrackerDistributedCacheManager.java
(original)
+++
hadoop/common/branches/branch-0.20-security-patches/src/test/org/apache/hadoop/filecache/TestTrackerDistributedCacheManager.java
Fri Mar 4 04:24:33 2011
@@ -65,7 +65,7 @@ public class TestTrackerDistributedCache
private static final int TEST_FILE_SIZE = 4 * 1024; // 4K
private static final int LOCAL_CACHE_LIMIT = 5 * 1024; //5K
- private static final int LOCAL_CACHE_SUBDIR = 2;
+ private static final int LOCAL_CACHE_SUBDIR_LIMIT = 2;
protected Configuration conf;
protected Path firstCacheFile;
protected Path secondCacheFile;
@@ -481,7 +481,7 @@ public class TestTrackerDistributedCache
conf2.set("mapred.local.dir", ROOT_MAPRED_LOCAL_DIR.toString());
conf2.setLong("local.cache.size", LOCAL_CACHE_LIMIT);
conf2.setLong("mapreduce.tasktracker.local.cache.numberdirectories",
- LOCAL_CACHE_SUBDIR);
+ LOCAL_CACHE_SUBDIR_LIMIT);
refreshConf(conf2);
TrackerDistributedCacheManager manager =
new TrackerDistributedCacheManager(conf2, taskController);
@@ -491,7 +491,7 @@ public class TestTrackerDistributedCache
conf2.set("user.name", userName);
// We first test the size limit
- Path localCache = manager.getLocalCache(firstCacheFile.toUri(), conf2,
+ Path firstLocalCache = manager.getLocalCache(firstCacheFile.toUri(),
conf2,
TaskTracker.getPrivateDistributedCacheDir(userName),
fs.getFileStatus(firstCacheFile), false,
now, new Path(TEST_ROOT_DIR), false, false);
@@ -500,13 +500,41 @@ public class TestTrackerDistributedCache
// which will cause the cache be deleted when the limit goes out.
// The below code localize another cache which's designed to
//sweep away the first cache.
- manager.getLocalCache(secondCacheFile.toUri(), conf2,
+ Path secondLocalCache = manager.getLocalCache(secondCacheFile.toUri(),
conf2,
TaskTracker.getPrivateDistributedCacheDir(userName),
fs.getFileStatus(secondCacheFile), false,
System.currentTimeMillis(), new Path(TEST_ROOT_DIR), false, false);
assertFalse("DistributedCache failed deleting old" +
" cache when the cache store is full.",
- localfs.exists(localCache));
+ localfs.exists(firstLocalCache));
+
+ // find the root directory of distributed caches
+ Path firstCursor = firstLocalCache;
+ Path secondCursor = secondLocalCache;
+
+ while (!firstCursor.equals(secondCursor)) {
+ // Debug code, to see what these things look like
+ System.err.println("cursors: " + firstCursor);
+ System.err.println(" and " + secondCursor);
+
+ firstCursor = firstCursor.getParent();
+ secondCursor = secondCursor.getParent();
+ }
+
+ System.err.println("The final cursor is " + firstCursor);
+
+ System.err.println("That directory ends up with "
+ + localfs.listStatus(firstCursor).length
+ + " subdirectories");
+
+ Path cachesBase = firstCursor;
+
+ assertFalse
+ ("DistributedCache did not delete the gensym'ed distcache "
+ + "directory names when it deleted the files they contained "
+ + "because they collectively exceeded the size limit.",
+ localfs.listStatus(cachesBase).length > 1);
+
// Now we test the number of sub directories limit
// Create the temporary cache files to be used in the tests.
@@ -524,13 +552,20 @@ public class TestTrackerDistributedCache
manager.releaseCache(thirdCacheFile.toUri(), conf2, now);
// Getting the fourth cache will make the number of sub directories becomes
// 3 which is greater than 2. So the released cache will be deleted.
- manager.getLocalCache(fourthCacheFile.toUri(), conf2,
+ Path fourthLocalCache = manager.getLocalCache(fourthCacheFile.toUri(),
conf2,
TaskTracker.getPrivateDistributedCacheDir(userName),
fs.getFileStatus(fourthCacheFile), false,
System.currentTimeMillis(), new Path(TEST_ROOT_DIR), false, false);
assertFalse("DistributedCache failed deleting old" +
" cache when the cache exceeds the number of sub directories limit.",
localfs.exists(thirdLocalCache));
+
+ assertFalse
+ ("DistributedCache did not delete the gensym'ed distcache "
+ + "directory names when it deleted the files they contained "
+ + "because there were too many.",
+ localfs.listStatus(cachesBase).length > LOCAL_CACHE_SUBDIR_LIMIT);
+
// Clean up the files created in this test
new File(thirdCacheFile.toString()).delete();
new File(fourthCacheFile.toString()).delete();
