| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.hadoop.mapreduce.filecache; |
| |
| import java.io.File; |
| import java.io.IOException; |
| import java.net.URI; |
| import java.util.Collection; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.LinkedHashMap; |
| import java.util.LinkedList; |
| import java.util.Map; |
| import java.util.Random; |
| import java.util.TreeMap; |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.mapred.TaskController; |
| import org.apache.hadoop.mapred.TaskController.DistributedCacheFileContext; |
| import org.apache.hadoop.mapreduce.MRJobConfig; |
| import org.apache.hadoop.mapreduce.server.tasktracker.TTConfig; |
| import org.apache.hadoop.mapreduce.util.MRAsyncDiskService; |
| import org.apache.hadoop.fs.FileStatus; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.FileUtil; |
| import org.apache.hadoop.fs.LocalDirAllocator; |
| import org.apache.hadoop.fs.LocalFileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.security.UserGroupInformation; |
| import org.apache.hadoop.util.RunJar; |
| import org.apache.hadoop.classification.InterfaceAudience; |
| |
| /** |
| * Manages a single machine's instance of a cross-job |
| * cache. This class would typically be instantiated |
| * by a TaskTracker (or something that emulates it, |
| * like LocalJobRunner). |
| * |
| */ |
| @InterfaceAudience.Private |
| public class TrackerDistributedCacheManager { |
| // cacheID to cacheStatus mapping |
| private LinkedHashMap<String, CacheStatus> cachedArchives = |
| new LinkedHashMap<String, CacheStatus>(); |
| |
| // default total cache size (10GB) |
| private static final long DEFAULT_CACHE_SIZE = 10737418240L; |
| private static final long DEFAULT_CACHE_SUBDIR_LIMIT = 10000; |
| // This default of 75 % was just pulled out of the air, it should be set |
| // based off of read world use cases. |
| private static final float DEFAULT_CACHE_KEEP_AROUND_PCT = 0.75f; |
| private long allowedCacheSize; |
| private long allowedCacheSubdirs; |
| private long allowedCacheSizeCleanupGoal; |
| private long allowedCacheSubdirsCleanupGoal; |
| |
| private static final Log LOG = |
| LogFactory.getLog(TrackerDistributedCacheManager.class); |
| |
| private final LocalFileSystem localFs; |
| |
| private LocalDirAllocator lDirAllocator; |
| |
| private TaskController taskController; |
| |
| private Configuration trackerConf; |
| |
| private Random random = new Random(); |
| |
| private MRAsyncDiskService asyncDiskService; |
| |
| protected BaseDirManager baseDirManager = new BaseDirManager(); |
| protected CleanupThread cleanupThread; |
| |
| public TrackerDistributedCacheManager(Configuration conf, |
| TaskController taskController) throws IOException { |
| this.localFs = FileSystem.getLocal(conf); |
| this.trackerConf = conf; |
| this.lDirAllocator = new LocalDirAllocator(TTConfig.LOCAL_DIR); |
| this.taskController = taskController; |
| // setting the cache size to a default of 10GB |
| this.allowedCacheSize = conf.getLong(TTConfig.TT_LOCAL_CACHE_SIZE, |
| DEFAULT_CACHE_SIZE); |
| // setting the cache number of subdirectories limit to a default of 10000 |
| this.allowedCacheSubdirs = conf.getLong( |
| TTConfig.TT_LOCAL_CACHE_SUBDIRS_LIMIT, DEFAULT_CACHE_SUBDIR_LIMIT); |
| double cleanupPct = conf.getFloat(TTConfig.TT_LOCAL_CACHE_KEEP_AROUND_PCT, |
| DEFAULT_CACHE_KEEP_AROUND_PCT); |
| this.allowedCacheSizeCleanupGoal = |
| (long)(this.allowedCacheSize * cleanupPct); |
| this.allowedCacheSubdirsCleanupGoal = |
| (long)(this.allowedCacheSubdirs * cleanupPct); |
| this.cleanupThread = new CleanupThread(conf); |
| } |
| |
| /** |
| * Creates a TrackerDistributedCacheManager with a MRAsyncDiskService. |
| * @param asyncDiskService Provides a set of ThreadPools for async disk |
| * operations. |
| */ |
| public TrackerDistributedCacheManager(Configuration conf, |
| TaskController taskController, MRAsyncDiskService asyncDiskService) |
| throws IOException { |
| this(conf, taskController); |
| this.asyncDiskService = asyncDiskService; |
| } |
| |
| /** |
| * Get the locally cached file or archive; it could either be |
| * previously cached (and valid) or copy it from the {@link FileSystem} now. |
| * |
| * @param cache the cache to be localized, this should be specified as |
| * new URI(scheme://scheme-specific-part/absolute_path_to_file#LINKNAME). |
| * @param conf The Configuration file which contains the filesystem |
| * @param subDir The base cache subDir where you want to localize the |
| * files/archives |
| * @param fileStatus The file status on the dfs. |
| * @param isArchive if the cache is an archive or a file. In case it is an |
| * archive with a .zip or .jar or .tar or .tgz or .tar.gz extension it will |
| * be unzipped/unjarred/untarred automatically |
| * and the directory where the archive is unzipped/unjarred/untarred is |
| * returned as the Path. |
| * In case of a file, the path to the file is returned |
| * @param confFileStamp this is the hdfs file modification timestamp to verify |
| * that the file to be cached hasn't changed since the job started |
| * @param currentWorkDir this is the directory where you would want to create |
| * symlinks for the locally cached files/archives |
| * @param honorSymLinkConf if this is false, then the symlinks are not |
| * created even if conf says so (this is required for an optimization in task |
| * launches |
| * NOTE: This is effectively always on since r696957, since there is no code |
| * path that does not use this. |
| * @param isPublic to know the cache file is accessible to public or private |
| * @return the path to directory where the archives are unjarred in case of |
| * archives, the path to the file where the file is copied locally |
| * @throws IOException |
| */ |
| Path getLocalCache(URI cache, Configuration conf, |
| String subDir, FileStatus fileStatus, |
| boolean isArchive, long confFileStamp, |
| Path currentWorkDir, boolean honorSymLinkConf, boolean isPublic) |
| throws IOException { |
| String key; |
| key = getKey(cache, conf, confFileStamp, getLocalizedCacheOwner(isPublic)); |
| CacheStatus lcacheStatus; |
| Path localizedPath = null; |
| synchronized (cachedArchives) { |
| lcacheStatus = cachedArchives.get(key); |
| if (lcacheStatus == null) { |
| // was never localized |
| String uniqueString = String.valueOf(random.nextLong()); |
| String cachePath = new Path (subDir, |
| new Path(uniqueString, makeRelative(cache, conf))).toString(); |
| Path localPath = lDirAllocator.getLocalPathForWrite(cachePath, |
| fileStatus.getLen(), trackerConf); |
| lcacheStatus = new CacheStatus(new Path(localPath.toString().replace( |
| cachePath, "")), localPath, new Path(subDir), uniqueString, key); |
| cachedArchives.put(key, lcacheStatus); |
| } |
| |
| //mark the cache for use. |
| lcacheStatus.incRefCount(); |
| } |
| |
| boolean initSuccessful = false; |
| try { |
| // do the localization, after releasing the global lock |
| synchronized (lcacheStatus) { |
| if (!lcacheStatus.isInited()) { |
| FileSystem fs = FileSystem.get(cache, conf); |
| checkStampSinceJobStarted(conf, fs, cache, confFileStamp, |
| lcacheStatus, fileStatus); |
| localizedPath = localizeCache(conf, cache, confFileStamp, |
| lcacheStatus, isArchive, isPublic); |
| lcacheStatus.initComplete(); |
| } else { |
| localizedPath = checkCacheStatusValidity(conf, cache, confFileStamp, |
| lcacheStatus, fileStatus, isArchive); |
| } |
| createSymlink(conf, cache, lcacheStatus, isArchive, currentWorkDir, |
| honorSymLinkConf); |
| } |
| initSuccessful = true; |
| return localizedPath; |
| } finally { |
| if (!initSuccessful) { |
| lcacheStatus.decRefCount(); |
| } |
| } |
| } |
| |
| /** |
| * This is the opposite of getlocalcache. When you are done with |
| * using the cache, you need to release the cache |
| * @param cache The cache URI to be released |
| * @param conf configuration which contains the filesystem the cache |
| * @param timeStamp the timestamp on the file represented by the cache URI |
| * @param owner the owner of the localized file |
| * is contained in. |
| * @throws IOException |
| */ |
| void releaseCache(URI cache, Configuration conf, long timeStamp, |
| String owner) throws IOException { |
| String key = getKey(cache, conf, timeStamp, owner); |
| synchronized (cachedArchives) { |
| CacheStatus lcacheStatus = cachedArchives.get(key); |
| if (lcacheStatus == null) { |
| LOG.warn("Cannot find localized cache: " + cache + |
| " (key: " + key + ") in releaseCache!"); |
| return; |
| } |
| |
| // decrement ref count |
| lcacheStatus.decRefCount(); |
| } |
| } |
| |
| /* |
| * This method is called from unit tests. |
| */ |
| int getReferenceCount(URI cache, Configuration conf, long timeStamp, |
| String owner) throws IOException { |
| String key = getKey(cache, conf, timeStamp, owner); |
| synchronized (cachedArchives) { |
| CacheStatus lcacheStatus = cachedArchives.get(key); |
| if (lcacheStatus == null) { |
| throw new IOException("Cannot find localized cache: " + cache); |
| } |
| return lcacheStatus.getRefCount(); |
| } |
| } |
| |
| /** |
| * Get the user who should "own" the localized distributed cache file. |
| * If the cache is public, the tasktracker user is the owner. If private, |
| * the user that the task is running as, is the owner. |
| * @param isPublic |
| * @return the owner as a shortname string |
| * @throws IOException |
| */ |
| static String getLocalizedCacheOwner(boolean isPublic) throws IOException { |
| String user; |
| if (isPublic) { |
| user = UserGroupInformation.getLoginUser().getShortUserName(); |
| } else { |
| user = UserGroupInformation.getCurrentUser().getShortUserName(); |
| } |
| return user; |
| } |
| |
| /** |
| * Delete a local path with asyncDiskService if available, |
| * or otherwise synchronously with local file system. |
| */ |
| private static void deleteLocalPath(MRAsyncDiskService asyncDiskService, |
| LocalFileSystem fs, Path path) throws IOException { |
| boolean deleted = false; |
| if (asyncDiskService != null) { |
| // Try to delete using asyncDiskService |
| String localPathToDelete = |
| path.toUri().getPath(); |
| deleted = asyncDiskService.moveAndDeleteAbsolutePath(localPathToDelete); |
| if (!deleted) { |
| LOG.warn("Cannot find DistributedCache path " + localPathToDelete |
| + " on any of the asyncDiskService volumes!"); |
| } |
| } |
| if (!deleted) { |
| // If no asyncDiskService, we will delete the files synchronously |
| fs.delete(path, true); |
| } |
| LOG.info("Deleted path " + path); |
| } |
| |
| /* |
| * Returns the relative path of the dir this cache will be localized in |
| * relative path that this cache will be localized in. For |
| * hdfs://hostname:port/absolute_path -- the relative path is |
| * hostname/absolute path -- if it is just /absolute_path -- then the |
| * relative path is hostname of DFS this mapred cluster is running |
| * on/absolute_path |
| */ |
| String makeRelative(URI cache, Configuration conf) |
| throws IOException { |
| String host = cache.getHost(); |
| if (host == null) { |
| host = cache.getScheme(); |
| } |
| if (host == null) { |
| URI defaultUri = FileSystem.get(conf).getUri(); |
| host = defaultUri.getHost(); |
| if (host == null) { |
| host = defaultUri.getScheme(); |
| } |
| } |
| String path = host + cache.getPath(); |
| path = path.replace(":/","/"); // remove windows device colon |
| return path; |
| } |
| |
| String getKey(URI cache, Configuration conf, long timeStamp, String user) |
| throws IOException { |
| return makeRelative(cache, conf) + String.valueOf(timeStamp) + user; |
| } |
| |
| /** |
| * Returns {@link FileStatus} of a given cache file on hdfs. |
| * |
| * @param conf configuration |
| * @param cache cache file |
| * @return {@link FileStatus} of a given cache file on hdfs |
| * @throws IOException |
| */ |
| static FileStatus getFileStatus(Configuration conf, URI cache) |
| throws IOException { |
| return ClientDistributedCacheManager.getFileStatus(conf, cache); |
| } |
| |
| /** |
| * Returns mtime of a given cache file on hdfs. |
| * |
| * @param conf configuration |
| * @param cache cache file |
| * @return mtime of a given cache file on hdfs |
| * @throws IOException |
| */ |
| static long getTimestamp(Configuration conf, URI cache) |
| throws IOException { |
| return getFileStatus(conf, cache).getModificationTime(); |
| } |
| |
| private Path checkCacheStatusValidity(Configuration conf, |
| URI cache, long confFileStamp, |
| CacheStatus cacheStatus, |
| FileStatus fileStatus, |
| boolean isArchive |
| ) throws IOException { |
| FileSystem fs = FileSystem.get(cache, conf); |
| // Has to be |
| if (!ifExistsAndFresh(conf, fs, cache, confFileStamp, |
| cacheStatus, fileStatus)) { |
| throw new IOException("Stale cache file: " + cacheStatus.localizedLoadPath + |
| " for cache-file: " + cache); |
| } |
| |
| LOG.info(String.format("Using existing cache of %s->%s", |
| cache.toString(), cacheStatus.localizedLoadPath)); |
| return cacheStatus.localizedLoadPath; |
| } |
| |
| private void createSymlink(Configuration conf, URI cache, |
| CacheStatus cacheStatus, boolean isArchive, |
| Path currentWorkDir, boolean honorSymLinkConf) throws IOException { |
| boolean doSymlink = honorSymLinkConf && DistributedCache.getSymlink(conf); |
| if(cache.getFragment() == null) { |
| doSymlink = false; |
| } |
| String link = |
| currentWorkDir.toString() + Path.SEPARATOR + cache.getFragment(); |
| File flink = new File(link); |
| if (doSymlink){ |
| if (!flink.exists()) { |
| FileUtil.symLink(cacheStatus.localizedLoadPath.toString(), link); |
| } |
| } |
| } |
| |
| // the method which actually copies the caches locally and unjars/unzips them |
| // and does chmod for the files |
| Path localizeCache(Configuration conf, |
| URI cache, long confFileStamp, |
| CacheStatus cacheStatus, |
| boolean isArchive, boolean isPublic) |
| throws IOException { |
| FileSystem fs = FileSystem.get(cache, conf); |
| FileSystem localFs = FileSystem.getLocal(conf); |
| Path parchive = null; |
| if (isArchive) { |
| parchive = new Path(cacheStatus.localizedLoadPath, |
| new Path(cacheStatus.localizedLoadPath.getName())); |
| } else { |
| parchive = cacheStatus.localizedLoadPath; |
| } |
| |
| if (!localFs.mkdirs(parchive.getParent())) { |
| throw new IOException("Mkdirs failed to create directory " + |
| cacheStatus.localizedLoadPath.toString()); |
| } |
| |
| String cacheId = cache.getPath(); |
| fs.copyToLocalFile(new Path(cacheId), parchive); |
| if (isArchive) { |
| String tmpArchive = parchive.toString().toLowerCase(); |
| File srcFile = new File(parchive.toString()); |
| File destDir = new File(parchive.getParent().toString()); |
| LOG.info(String.format("Extracting %s to %s", |
| srcFile.toString(), destDir.toString())); |
| if (tmpArchive.endsWith(".jar")) { |
| RunJar.unJar(srcFile, destDir); |
| } else if (tmpArchive.endsWith(".zip")) { |
| FileUtil.unZip(srcFile, destDir); |
| } else if (isTarFile(tmpArchive)) { |
| FileUtil.unTar(srcFile, destDir); |
| } else { |
| LOG.warn(String.format( |
| "Cache file %s specified as archive, but not valid extension.", |
| srcFile.toString())); |
| // else will not do anyhting |
| // and copy the file into the dir as it is |
| } |
| } |
| |
| long cacheSize = |
| FileUtil.getDU(new File(parchive.getParent().toString())); |
| cacheStatus.size = cacheSize; |
| // Increase the size and sub directory count of the cache |
| // from baseDirSize and baseDirNumberSubDir. |
| baseDirManager.addCacheUpdate(cacheStatus); |
| |
| // set proper permissions for the localized directory |
| setPermissions(conf, cacheStatus, isPublic); |
| |
| // update cacheStatus to reflect the newly cached file |
| cacheStatus.mtime = getTimestamp(conf, cache); |
| |
| LOG.info(String.format("Cached %s as %s", |
| cache.toString(), cacheStatus.localizedLoadPath)); |
| return cacheStatus.localizedLoadPath; |
| } |
| |
| private void setPermissions(Configuration conf, CacheStatus cacheStatus, |
| boolean isPublic) throws IOException { |
| if (isPublic) { |
| Path localizedUniqueDir = cacheStatus.getLocalizedUniqueDir(); |
| LOG.info("Doing chmod on localdir :" + localizedUniqueDir); |
| try { |
| FileUtil.chmod(localizedUniqueDir.toString(), "ugo+rx", true); |
| } catch (InterruptedException e) { |
| LOG.warn("Exception in chmod" + e.toString()); |
| throw new IOException(e); |
| } |
| } else { |
| // invoke taskcontroller to set permissions |
| DistributedCacheFileContext context = new DistributedCacheFileContext( |
| conf.get(MRJobConfig.USER_NAME), new File(cacheStatus.localizedBaseDir |
| .toString()), cacheStatus.localizedBaseDir, |
| cacheStatus.uniqueString); |
| taskController.initializeDistributedCacheFile(context); |
| } |
| } |
| |
| private static boolean isTarFile(String filename) { |
| return (filename.endsWith(".tgz") || filename.endsWith(".tar.gz") || |
| filename.endsWith(".tar")); |
| } |
| |
| // ensure that the file on hdfs hasn't been modified since the job started |
| private long checkStampSinceJobStarted(Configuration conf, FileSystem fs, |
| URI cache, long confFileStamp, |
| CacheStatus lcacheStatus, |
| FileStatus fileStatus) |
| throws IOException { |
| long dfsFileStamp; |
| if (fileStatus != null) { |
| dfsFileStamp = fileStatus.getModificationTime(); |
| } else { |
| dfsFileStamp = getTimestamp(conf, cache); |
| } |
| |
| if (dfsFileStamp != confFileStamp) { |
| LOG.fatal("File: " + cache + " has changed on HDFS since job started"); |
| throw new IOException("File: " + cache + |
| " has changed on HDFS since job started"); |
| } |
| |
| return dfsFileStamp; |
| } |
| |
| // Checks if the cache has already been localized and is fresh |
| private boolean ifExistsAndFresh(Configuration conf, FileSystem fs, |
| URI cache, long confFileStamp, |
| CacheStatus lcacheStatus, |
| FileStatus fileStatus) |
| throws IOException { |
| long dfsFileStamp = checkStampSinceJobStarted(conf, fs, cache, |
| confFileStamp, lcacheStatus, fileStatus); |
| if (dfsFileStamp != lcacheStatus.mtime) { |
| return false; |
| } |
| |
| return true; |
| } |
| |
| /** |
| * This method create symlinks for all files in a given dir in another |
| * directory. |
| * |
| * Should not be used outside of DistributedCache code. |
| * |
| * @param conf the configuration |
| * @param jobCacheDir the target directory for creating symlinks |
| * @param workDir the directory in which the symlinks are created |
| * @throws IOException |
| */ |
| public static void createAllSymlink(Configuration conf, File jobCacheDir, |
| File workDir) |
| throws IOException{ |
| if ((jobCacheDir == null || !jobCacheDir.isDirectory()) || |
| workDir == null || (!workDir.isDirectory())) { |
| return; |
| } |
| boolean createSymlink = DistributedCache.getSymlink(conf); |
| if (createSymlink){ |
| File[] list = jobCacheDir.listFiles(); |
| for (int i=0; i < list.length; i++){ |
| String target = list[i].getAbsolutePath(); |
| String link = new File(workDir, list[i].getName()).toString(); |
| LOG.info(String.format("Creating symlink: %s <- %s", target, link)); |
| int ret = FileUtil.symLink(target, link); |
| if (ret != 0) { |
| LOG.warn(String.format("Failed to create symlink: %s <- %s", target, |
| link)); |
| } |
| } |
| } |
| } |
| |
| class CacheStatus { |
| // |
| // This field should be accessed under global cachedArchives lock. |
| // |
| private int refcount; // number of instances using this cache. |
| |
| // |
| // The following three fields should be accessed under |
| // individual cacheStatus lock. |
| // |
| long size; //the size of this cache. |
| long mtime; // the cache-file modification time |
| boolean inited = false; // is it initialized ? |
| |
| // |
| // The following four fields are Immutable. |
| // |
| // The sub directory (tasktracker/archive or tasktracker/user/archive), |
| // under which the file will be localized |
| final Path subDir; |
| // unique string used in the construction of local load path |
| final String uniqueString; |
| // the local load path of this cache |
| final Path localizedLoadPath; |
| //the base dir where the cache lies |
| final Path localizedBaseDir; |
| //The key of this in the cachedArchives. |
| private final String key; |
| |
| public CacheStatus(Path baseDir, Path localLoadPath, Path subDir, |
| String uniqueString, String key) { |
| super(); |
| this.localizedLoadPath = localLoadPath; |
| this.refcount = 0; |
| this.mtime = -1; |
| this.localizedBaseDir = baseDir; |
| this.size = 0; |
| this.subDir = subDir; |
| this.uniqueString = uniqueString; |
| this.key = key; |
| } |
| |
| public synchronized void incRefCount() { |
| refcount += 1; |
| } |
| |
| public void decRefCount() { |
| synchronized (cachedArchives) { |
| synchronized (this) { |
| refcount -= 1; |
| if(refcount <= 0) { |
| String key = this.key; |
| cachedArchives.remove(key); |
| cachedArchives.put(key, this); |
| } |
| } |
| } |
| } |
| |
| public int getRefCount() { |
| return refcount; |
| } |
| |
| public synchronized boolean isUsed() { |
| return refcount > 0; |
| } |
| |
| Path getBaseDir(){ |
| return this.localizedBaseDir; |
| } |
| |
| // mark it as initialized |
| void initComplete() { |
| inited = true; |
| } |
| |
| // is it initialized? |
| boolean isInited() { |
| return inited; |
| } |
| |
| Path getLocalizedUniqueDir() { |
| return new Path(localizedBaseDir, new Path(subDir, uniqueString)); |
| } |
| } |
| |
| /** |
| * Clear the entire contents of the cache and delete the backing files. This |
| * should only be used when the server is reinitializing, because the users |
| * are going to lose their files. |
| */ |
| public void purgeCache() { |
| synchronized (cachedArchives) { |
| for (Map.Entry<String,CacheStatus> f: cachedArchives.entrySet()) { |
| try { |
| deleteLocalPath(asyncDiskService, localFs, f.getValue().localizedLoadPath); |
| } catch (IOException ie) { |
| LOG.debug("Error cleaning up cache", ie); |
| } |
| } |
| cachedArchives.clear(); |
| } |
| } |
| |
| public TaskDistributedCacheManager newTaskDistributedCacheManager( |
| Configuration taskConf) throws IOException { |
| return new TaskDistributedCacheManager(this, taskConf); |
| } |
| |
| /** |
| * Get the booleans on whether the files are public or not. Used by |
| * internal DistributedCache and MapReduce code. |
| * @param conf The configuration which stored the timestamps |
| * @return a string array of booleans |
| * @throws IOException |
| */ |
| static String[] getFileVisibilities(Configuration conf) { |
| return conf.getStrings(MRJobConfig.CACHE_FILE_VISIBILITIES); |
| } |
| |
| /** |
| * Get the booleans on whether the archives are public or not. Used by |
| * internal DistributedCache and MapReduce code. |
| * @param conf The configuration which stored the timestamps |
| * @return a string array of booleans |
| */ |
| static String[] getArchiveVisibilities(Configuration conf) { |
| return conf.getStrings(MRJobConfig.CACHE_ARCHIVES_VISIBILITIES); |
| } |
| |
| /** |
| * Set the conf to contain the location for localized archives. |
| * |
| * @param conf The conf to modify to contain the localized caches |
| * @param str a comma separated list of local archives |
| */ |
| static void setLocalArchives(Configuration conf, String str) { |
| conf.set(MRJobConfig.CACHE_LOCALARCHIVES, str); |
| } |
| |
| /** |
| * Set the conf to contain the location for localized files. |
| * |
| * @param conf The conf to modify to contain the localized caches |
| * @param str a comma separated list of local files |
| */ |
| public static void setLocalFiles(Configuration conf, String str) { |
| conf.set(MRJobConfig.CACHE_LOCALFILES, str); |
| } |
| |
| /** |
| * A thread to check and cleanup the unused files periodically |
| */ |
| protected class CleanupThread extends Thread { |
| // How often do we check if we need to clean up cache files? |
| private long cleanUpCheckPeriod = 60000L; // 1 minute |
| public CleanupThread(Configuration conf) { |
| cleanUpCheckPeriod = |
| conf.getLong(TTConfig.TT_DISTRIBUTED_CACHE_CHECK_PERIOD, |
| cleanUpCheckPeriod); |
| } |
| |
| private volatile boolean running = true; |
| public void stopRunning() { |
| running = false; |
| } |
| @Override |
| public void run() { |
| while (running) { |
| try { |
| Thread.sleep(cleanUpCheckPeriod); |
| baseDirManager.checkAndCleanup(); |
| } catch (IOException e) { |
| LOG.error("Exception in DistributedCache CleanupThread.", e); |
| } catch (InterruptedException e) { |
| LOG.info("Cleanup...",e); |
| //To force us to exit cleanly |
| running = false; |
| } catch (Throwable t) { |
| exitTaskTracker(t); |
| } |
| } |
| } |
| |
| /** |
| * Exit the task tracker because of a fatal error. |
| */ |
| protected void exitTaskTracker(Throwable t) { |
| LOG.fatal("Distributed Cache cleanup thread received runtime exception." + |
| " Exiting the TaskTracker", t); |
| Runtime.getRuntime().exit(-1); |
| } |
| } |
| |
| private class CacheDir { |
| long size; |
| long subdirs; |
| } |
| |
| /** |
| * This class holds properties of each base directories and is responsible |
| * for clean up unused cache files in base directories. |
| */ |
| protected class BaseDirManager { |
| |
| private TreeMap<Path, CacheDir> properties = |
| new TreeMap<Path, CacheDir>(); |
| |
| /** |
| * Check each base directory to see if the size or number of subdirectories |
| * are exceed the limit. If the limit is exceeded, start deleting caches |
| * with zero reference count. This method synchronizes cachedArchives. |
| */ |
| public void checkAndCleanup() throws IOException { |
| Collection<CacheStatus> toBeDeletedCache = new LinkedList<CacheStatus>(); |
| HashMap<Path, CacheDir> toBeCleanedBaseDir = |
| new HashMap<Path, CacheDir>(); |
| synchronized (properties) { |
| for (Map.Entry<Path, CacheDir> baseDir : properties.entrySet()) { |
| CacheDir baseDirCounts = baseDir.getValue(); |
| if (allowedCacheSize < baseDirCounts.size || |
| allowedCacheSubdirs < baseDirCounts.subdirs) { |
| CacheDir tcc = new CacheDir(); |
| tcc.size = baseDirCounts.size - allowedCacheSizeCleanupGoal; |
| tcc.subdirs = baseDirCounts.subdirs - allowedCacheSubdirsCleanupGoal; |
| toBeCleanedBaseDir.put(baseDir.getKey(), tcc); |
| } |
| } |
| } |
| synchronized (cachedArchives) { |
| for (Iterator<Map.Entry<String, CacheStatus>> it = |
| cachedArchives.entrySet().iterator(); it.hasNext();) { |
| Map.Entry<String, CacheStatus> entry = it.next(); |
| String cacheId = entry.getKey(); |
| CacheStatus cacheStatus = cachedArchives.get(cacheId); |
| CacheDir leftToClean = toBeCleanedBaseDir.get(cacheStatus.getBaseDir()); |
| if (leftToClean != null |
| && (leftToClean.size > 0 || leftToClean.subdirs > 0)) { |
| // if reference count is zero mark the cache for deletion |
| if (!cacheStatus.isUsed()) { |
| leftToClean.size -= cacheStatus.size; |
| leftToClean.subdirs--; |
| // delete this cache entry from the global list |
| // and mark the localized file for deletion |
| toBeDeletedCache.add(cacheStatus); |
| it.remove(); |
| } |
| } |
| } |
| } |
| // do the deletion, after releasing the global lock |
| for (CacheStatus cacheStatus : toBeDeletedCache) { |
| synchronized (cacheStatus) { |
| deleteLocalPath(asyncDiskService, FileSystem.getLocal(trackerConf), |
| cacheStatus.getLocalizedUniqueDir()); |
| // Update the maps baseDirSize and baseDirNumberSubDir |
| deleteCacheUpdate(cacheStatus); |
| } |
| } |
| } |
| |
| /** |
| * Decrement the size and sub directory count of the cache from baseDirSize |
| * and baseDirNumberSubDir. Have to synchronize cacheStatus before calling |
| * this method |
| * @param cacheStatus cache status of the cache is deleted |
| */ |
| public void deleteCacheUpdate(CacheStatus cacheStatus) { |
| if (!cacheStatus.inited) { |
| // if it is not created yet, do nothing. |
| return; |
| } |
| synchronized (properties) { |
| CacheDir cacheDir = properties.get(cacheStatus.getBaseDir()); |
| if (cacheDir != null) { |
| cacheDir.size -= cacheStatus.size; |
| cacheDir.subdirs--; |
| } else { |
| LOG.warn("Cannot find size and number of subdirectories of" + |
| " baseDir: " + cacheStatus.getBaseDir()); |
| } |
| } |
| } |
| |
| /** |
| * Update the maps baseDirSize and baseDirNumberSubDir when adding cache. |
| * Increase the size and sub directory count of the cache from baseDirSize |
| * and baseDirNumberSubDir. Have to synchronize cacheStatus before calling |
| * this method. |
| * @param cacheStatus cache status of the cache is added |
| */ |
| public void addCacheUpdate(CacheStatus cacheStatus) { |
| long cacheSize = cacheStatus.size; |
| LOG.info("Adding in cache "+cacheStatus.localizedLoadPath+" at "+cacheStatus.localizedBaseDir+" size:"+cacheStatus.size); |
| synchronized (properties) { |
| CacheDir cacheDir = properties.get(cacheStatus.getBaseDir()); |
| if (cacheDir != null) { |
| cacheDir.size += cacheSize; |
| cacheDir.subdirs++; |
| } else { |
| cacheDir = new CacheDir(); |
| cacheDir.size = cacheSize; |
| cacheDir.subdirs = 1; |
| properties.put(cacheStatus.getBaseDir(), cacheDir); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Start the background thread |
| */ |
| public void startCleanupThread() { |
| this.cleanupThread.start(); |
| } |
| |
| /** |
| * Stop the background thread |
| */ |
| public void stopCleanupThread() { |
| cleanupThread.stopRunning(); |
| cleanupThread.interrupt(); |
| } |
| } |
