| /** |
| * 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.mapred; |
| |
| import java.io.ByteArrayOutputStream; |
| import java.io.File; |
| import java.io.IOException; |
| import java.io.OutputStream; |
| import java.io.PrintStream; |
| import java.net.InetSocketAddress; |
| import java.net.URI; |
| import java.security.PrivilegedExceptionAction; |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Vector; |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.mapreduce.MRConfig; |
| import org.apache.hadoop.mapreduce.MRJobConfig; |
| import org.apache.hadoop.mapreduce.filecache.DistributedCache; |
| import org.apache.hadoop.mapreduce.filecache.TaskDistributedCacheManager; |
| import org.apache.hadoop.mapreduce.filecache.TrackerDistributedCacheManager; |
| import org.apache.hadoop.mapreduce.security.TokenCache; |
| import org.apache.hadoop.fs.FSError; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.FileUtil; |
| import org.apache.hadoop.fs.LocalDirAllocator; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.fs.permission.FsPermission; |
| import org.apache.hadoop.security.UserGroupInformation; |
| import org.apache.hadoop.util.Shell; |
| import org.apache.hadoop.util.StringUtils; |
| import org.apache.log4j.Level; |
| |
| /** Base class that runs a task in a separate process. Tasks are run in a |
| * separate process in order to isolate the map/reduce system code from bugs in |
| * user supplied map and reduce functions. |
| */ |
| abstract class TaskRunner extends Thread { |
| public static final Log LOG = |
| LogFactory.getLog(TaskRunner.class); |
| |
| volatile boolean killed = false; |
| private TaskTracker.TaskInProgress tip; |
| private Task t; |
| private Object lock = new Object(); |
| private volatile boolean done = false; |
| private int exitCode = -1; |
| private boolean exitCodeSet = false; |
| |
| private static String SYSTEM_PATH_SEPARATOR = System.getProperty("path.separator"); |
| |
| |
| private TaskTracker tracker; |
| private TaskDistributedCacheManager taskDistributedCacheManager; |
| |
| protected JobConf conf; |
| JvmManager jvmManager; |
| |
| public TaskRunner(TaskTracker.TaskInProgress tip, TaskTracker tracker, |
| JobConf conf) { |
| this.tip = tip; |
| this.t = tip.getTask(); |
| this.tracker = tracker; |
| this.conf = conf; |
| this.jvmManager = tracker.getJvmManagerInstance(); |
| } |
| |
| public Task getTask() { return t; } |
| public TaskTracker.TaskInProgress getTaskInProgress() { return tip; } |
| public TaskTracker getTracker() { return tracker; } |
| |
| public JvmManager getJvmManager() { return jvmManager; } |
| |
| /** Called when this task's output is no longer needed. |
| * This method is run in the parent process after the child exits. It should |
| * not execute user code, only system code. |
| */ |
| public void close() throws IOException {} |
| |
| |
| /** |
| * Get the java command line options for the child map/reduce tasks. |
| * @param jobConf job configuration |
| * @param defaultValue default value |
| * @return the java command line options for child map/reduce tasks |
| * @deprecated Use command line options specific to map or reduce tasks set |
| * via {@link JobConf#MAPRED_MAP_TASK_JAVA_OPTS} or |
| * {@link JobConf#MAPRED_REDUCE_TASK_JAVA_OPTS} |
| */ |
| @Deprecated |
| public String getChildJavaOpts(JobConf jobConf, String defaultValue) { |
| return jobConf.get(JobConf.MAPRED_TASK_JAVA_OPTS, defaultValue); |
| } |
| |
| /** |
| * Get the maximum virtual memory of the child map/reduce tasks. |
| * @param jobConf job configuration |
| * @return the maximum virtual memory of the child task or <code>-1</code> if |
| * none is specified |
| * @deprecated Use limits specific to the map or reduce tasks set via |
| * {@link JobConf#MAPRED_MAP_TASK_ULIMIT} or |
| * {@link JobConf#MAPRED_REDUCE_TASK_ULIMIT} |
| */ |
| @Deprecated |
| public int getChildUlimit(JobConf jobConf) { |
| return jobConf.getInt(JobConf.MAPRED_TASK_ULIMIT, -1); |
| } |
| |
| /** |
| * Get the environment variables for the child map/reduce tasks. |
| * @param jobConf job configuration |
| * @return the environment variables for the child map/reduce tasks or |
| * <code>null</code> if unspecified |
| * @deprecated Use environment variables specific to the map or reduce tasks |
| * set via {@link JobConf#MAPRED_MAP_TASK_ENV} or |
| * {@link JobConf#MAPRED_REDUCE_TASK_ENV} |
| */ |
| @Deprecated |
| public String getChildEnv(JobConf jobConf) { |
| return jobConf.get(JobConf.MAPRED_TASK_ENV); |
| } |
| |
| /** |
| * Get the log {@link Level} for the child map/reduce tasks. |
| * @param jobConf |
| * @return the log-level for the child map/reduce tasks |
| */ |
| public abstract Level getLogLevel(JobConf jobConf); |
| |
| @Override |
| public final void run() { |
| String errorInfo = "Child Error"; |
| try { |
| |
| //before preparing the job localize |
| //all the archives |
| TaskAttemptID taskid = t.getTaskID(); |
| final LocalDirAllocator lDirAlloc = new LocalDirAllocator(MRConfig.LOCAL_DIR); |
| final File workDir = formWorkDir(lDirAlloc, taskid, t.isTaskCleanupTask(), conf); |
| |
| // We don't create any symlinks yet, so presence/absence of workDir |
| // actually on the file system doesn't matter. |
| tip.getUGI().doAs(new PrivilegedExceptionAction<Void>() { |
| public Void run() throws IOException { |
| taskDistributedCacheManager = |
| tracker.getTrackerDistributedCacheManager() |
| .newTaskDistributedCacheManager(conf); |
| taskDistributedCacheManager.setup(lDirAlloc, workDir, TaskTracker |
| .getPrivateDistributedCacheDir(conf.getUser()), |
| TaskTracker.getPublicDistributedCacheDir()); |
| return null; |
| } |
| }); |
| |
| // Set up the child task's configuration. After this call, no localization |
| // of files should happen in the TaskTracker's process space. Any changes to |
| // the conf object after this will NOT be reflected to the child. |
| setupChildTaskConfiguration(lDirAlloc); |
| |
| // Build classpath |
| List<String> classPaths = |
| getClassPaths(conf, workDir, taskDistributedCacheManager); |
| |
| long logSize = TaskLog.getTaskLogLength(conf); |
| |
| // Build exec child JVM args. |
| Vector<String> vargs = |
| getVMArgs(taskid, workDir, classPaths, logSize); |
| |
| tracker.addToMemoryManager(t.getTaskID(), t.isMapTask(), conf); |
| |
| // set memory limit using ulimit if feasible and necessary ... |
| List<String> setup = getVMSetupCmd(); |
| |
| // Set up the redirection of the task's stdout and stderr streams |
| File[] logFiles = prepareLogFiles(taskid, t.isTaskCleanupTask()); |
| File stdout = logFiles[0]; |
| File stderr = logFiles[1]; |
| tracker.getTaskTrackerInstrumentation().reportTaskLaunch(taskid, stdout, |
| stderr); |
| |
| Map<String, String> env = new HashMap<String, String>(); |
| errorInfo = getVMEnvironment(errorInfo, workDir, conf, env, |
| taskid, logSize); |
| |
| launchJvmAndWait(setup, vargs, stdout, stderr, logSize, workDir, env); |
| tracker.getTaskTrackerInstrumentation().reportTaskEnd(t.getTaskID()); |
| if (exitCodeSet) { |
| if (!killed && exitCode != 0) { |
| if (exitCode == 65) { |
| tracker.getTaskTrackerInstrumentation().taskFailedPing(t.getTaskID()); |
| } |
| throw new IOException("Task process exit with nonzero status of " + |
| exitCode + "."); |
| } |
| } |
| } catch (FSError e) { |
| LOG.fatal("FSError", e); |
| try { |
| tracker.internalFsError(t.getTaskID(), e.getMessage()); |
| } catch (IOException ie) { |
| LOG.fatal(t.getTaskID()+" reporting FSError", ie); |
| } |
| } catch (Throwable throwable) { |
| LOG.warn(t.getTaskID() + " : " + errorInfo, throwable); |
| Throwable causeThrowable = new Throwable(errorInfo, throwable); |
| ByteArrayOutputStream baos = new ByteArrayOutputStream(); |
| causeThrowable.printStackTrace(new PrintStream(baos)); |
| try { |
| tracker.internalReportDiagnosticInfo(t.getTaskID(), baos.toString()); |
| } catch (IOException e) { |
| LOG.warn(t.getTaskID()+" Reporting Diagnostics", e); |
| } |
| } finally { |
| try{ |
| if (taskDistributedCacheManager != null) { |
| taskDistributedCacheManager.release(); |
| } |
| }catch(IOException ie){ |
| LOG.warn("Error releasing caches : Cache files might not have been cleaned up"); |
| } |
| |
| // It is safe to call TaskTracker.TaskInProgress.reportTaskFinished with |
| // *false* since the task has either |
| // a) SUCCEEDED - which means commit has been done |
| // b) FAILED - which means we do not need to commit |
| tip.reportTaskFinished(false); |
| } |
| } |
| |
| void launchJvmAndWait(List<String> setup, Vector<String> vargs, File stdout, |
| File stderr, long logSize, File workDir, Map<String, String> env) |
| throws InterruptedException { |
| jvmManager.launchJvm(this, jvmManager.constructJvmEnv(setup, vargs, stdout, |
| stderr, logSize, workDir, env, conf)); |
| synchronized (lock) { |
| while (!done) { |
| lock.wait(); |
| } |
| } |
| } |
| |
| /** |
| * Prepare the log files for the task |
| * |
| * @param taskid |
| * @param isCleanup |
| * @return an array of files. The first file is stdout, the second is stderr. |
| * @throws IOException |
| */ |
| File[] prepareLogFiles(TaskAttemptID taskid, boolean isCleanup) |
| throws IOException { |
| File[] logFiles = new File[2]; |
| logFiles[0] = TaskLog.getTaskLogFile(taskid, isCleanup, |
| TaskLog.LogName.STDOUT); |
| logFiles[1] = TaskLog.getTaskLogFile(taskid, isCleanup, |
| TaskLog.LogName.STDERR); |
| File logDir = logFiles[0].getParentFile(); |
| boolean b = logDir.mkdirs(); |
| if (!b) { |
| LOG.warn("mkdirs failed. Ignoring"); |
| } else { |
| FileSystem localFs = FileSystem.getLocal(conf); |
| localFs.setPermission(new Path(logDir.getCanonicalPath()), |
| new FsPermission((short)0700)); |
| } |
| |
| return logFiles; |
| } |
| |
| /** |
| * Write the child's configuration to the disk and set it in configuration so |
| * that the child can pick it up from there. |
| * |
| * @param lDirAlloc |
| * @throws IOException |
| */ |
| void setupChildTaskConfiguration(LocalDirAllocator lDirAlloc) |
| throws IOException { |
| |
| Path localTaskFile = |
| lDirAlloc.getLocalPathForWrite(TaskTracker.getTaskConfFile( |
| t.getUser(), t.getJobID().toString(), t.getTaskID().toString(), t |
| .isTaskCleanupTask()), conf); |
| |
| // write the child's task configuration file to the local disk |
| writeLocalTaskFile(localTaskFile.toString(), conf); |
| |
| // Set the final job file in the task. The child needs to know the correct |
| // path to job.xml. So set this path accordingly. |
| t.setJobFile(localTaskFile.toString()); |
| } |
| |
| /** |
| * @return |
| */ |
| private List<String> getVMSetupCmd() { |
| |
| int ulimit = getChildUlimit(conf); |
| if (ulimit <= 0) { |
| return null; |
| } |
| List<String> setup = null; |
| String[] ulimitCmd = Shell.getUlimitMemoryCommand(ulimit); |
| if (ulimitCmd != null) { |
| setup = new ArrayList<String>(); |
| for (String arg : ulimitCmd) { |
| setup.add(arg); |
| } |
| } |
| return setup; |
| } |
| |
| /** |
| * Parse the given string and return an array of individual java opts. Split |
| * on whitespace and replace the special string "@taskid@" with the task ID |
| * given. |
| * |
| * @param javaOpts The string to parse |
| * @param taskid The task ID to replace the special string with |
| * @return An array of individual java opts. |
| */ |
| static String[] parseChildJavaOpts(String javaOpts, TaskAttemptID taskid) { |
| javaOpts = javaOpts.replace("@taskid@", taskid.toString()); |
| return javaOpts.trim().split("\\s+"); |
| } |
| |
| /** |
| * @param taskid |
| * @param workDir |
| * @param classPaths |
| * @param logSize |
| * @return |
| * @throws IOException |
| */ |
| private Vector<String> getVMArgs(TaskAttemptID taskid, File workDir, |
| List<String> classPaths, long logSize) |
| throws IOException { |
| Vector<String> vargs = new Vector<String>(8); |
| File jvm = // use same jvm as parent |
| new File(new File(System.getProperty("java.home"), "bin"), "java"); |
| |
| vargs.add(jvm.toString()); |
| |
| // Add child (task) java-vm options. |
| // |
| // The following symbols if present in mapred.{map|reduce}.child.java.opts |
| // value are replaced: |
| // + @taskid@ is interpolated with value of TaskID. |
| // Other occurrences of @ will not be altered. |
| // |
| // Example with multiple arguments and substitutions, showing |
| // jvm GC logging, and start of a passwordless JVM JMX agent so can |
| // connect with jconsole and the likes to watch child memory, threads |
| // and get thread dumps. |
| // |
| // <property> |
| // <name>mapred.child.java.opts</name> |
| // <value>-Xmx512M -verbose:gc -Xloggc:/tmp/@taskid@.gc \ |
| // -Dcom.sun.management.jmxremote.authenticate=false \ |
| // -Dcom.sun.management.jmxremote.ssl=false \ |
| // </value> |
| // </property> |
| // |
| // <property> |
| // <name>mapred.child.java.opts</name> |
| // <value>-Xmx1024M -verbose:gc -Xloggc:/tmp/@taskid@.gc \ |
| // -Dcom.sun.management.jmxremote.authenticate=false \ |
| // -Dcom.sun.management.jmxremote.ssl=false \ |
| // </value> |
| // </property> |
| // |
| String[] javaOptsSplit = parseChildJavaOpts(getChildJavaOpts(conf, |
| JobConf.DEFAULT_MAPRED_TASK_JAVA_OPTS), |
| taskid); |
| |
| // Add java.library.path; necessary for loading native libraries. |
| // |
| // 1. To support native-hadoop library i.e. libhadoop.so, we add the |
| // parent processes' java.library.path to the child. |
| // 2. We also add the 'cwd' of the task to it's java.library.path to help |
| // users distribute native libraries via the DistributedCache. |
| // 3. The user can also specify extra paths to be added to the |
| // java.library.path via mapred.{map|reduce}.child.java.opts. |
| // |
| String libraryPath = System.getProperty("java.library.path"); |
| if (libraryPath == null) { |
| libraryPath = workDir.getAbsolutePath(); |
| } else { |
| libraryPath += SYSTEM_PATH_SEPARATOR + workDir; |
| } |
| boolean hasUserLDPath = false; |
| for(int i=0; i<javaOptsSplit.length ;i++) { |
| if(javaOptsSplit[i].startsWith("-Djava.library.path=")) { |
| javaOptsSplit[i] += SYSTEM_PATH_SEPARATOR + libraryPath; |
| hasUserLDPath = true; |
| break; |
| } |
| } |
| if(!hasUserLDPath) { |
| vargs.add("-Djava.library.path=" + libraryPath); |
| } |
| for (int i = 0; i < javaOptsSplit.length; i++) { |
| vargs.add(javaOptsSplit[i]); |
| } |
| |
| Path childTmpDir = createChildTmpDir(workDir, conf); |
| vargs.add("-Djava.io.tmpdir=" + childTmpDir); |
| |
| // Add classpath. |
| vargs.add("-classpath"); |
| String classPath = StringUtils.join(SYSTEM_PATH_SEPARATOR, classPaths); |
| vargs.add(classPath); |
| |
| // Setup the log4j prop |
| setupLog4jProperties(vargs, taskid, logSize); |
| |
| if (conf.getProfileEnabled()) { |
| if (conf.getProfileTaskRange(t.isMapTask() |
| ).isIncluded(t.getPartition())) { |
| File prof = TaskLog.getTaskLogFile(taskid, t.isTaskCleanupTask(), |
| TaskLog.LogName.PROFILE); |
| vargs.add(String.format(conf.getProfileParams(), prof.toString())); |
| } |
| } |
| |
| // Add main class and its arguments |
| vargs.add(Child.class.getName()); // main of Child |
| // pass umbilical address |
| InetSocketAddress address = tracker.getTaskTrackerReportAddress(); |
| vargs.add(address.getAddress().getHostAddress()); |
| vargs.add(Integer.toString(address.getPort())); |
| vargs.add(taskid.toString()); // pass task identifier |
| // pass task log location |
| vargs.add(TaskLog.getAttemptDir(taskid, t.isTaskCleanupTask()).toString()); |
| return vargs; |
| } |
| |
| private void setupLog4jProperties(Vector<String> vargs, TaskAttemptID taskid, |
| long logSize) { |
| vargs.add("-Dhadoop.log.dir=" + |
| new File(System.getProperty("hadoop.log.dir")).getAbsolutePath()); |
| vargs.add("-Dhadoop.root.logger=" + getLogLevel(conf).toString() + ",TLA"); |
| vargs.add("-D" + TaskLogAppender.TASKID_PROPERTY + "=" + taskid); |
| vargs.add("-D" + TaskLogAppender.ISCLEANUP_PROPERTY + |
| "=" + t.isTaskCleanupTask()); |
| vargs.add("-D" + TaskLogAppender.LOGSIZE_PROPERTY + "=" + logSize); |
| } |
| |
| /** |
| * @param taskid |
| * @param workDir |
| * @return |
| * @throws IOException |
| */ |
| static Path createChildTmpDir(File workDir, |
| JobConf conf) |
| throws IOException { |
| |
| // add java.io.tmpdir given by mapreduce.task.tmp.dir |
| String tmp = conf.get(MRJobConfig.TASK_TEMP_DIR, "./tmp"); |
| Path tmpDir = new Path(tmp); |
| |
| // if temp directory path is not absolute, prepend it with workDir. |
| if (!tmpDir.isAbsolute()) { |
| tmpDir = new Path(workDir.toString(), tmp); |
| |
| FileSystem localFs = FileSystem.getLocal(conf); |
| if (!localFs.mkdirs(tmpDir) && localFs.getFileStatus(tmpDir).isFile()) { |
| throw new IOException("Mkdirs failed to create " + tmpDir.toString()); |
| } |
| } |
| return tmpDir; |
| } |
| |
| /** |
| */ |
| private static List<String> getClassPaths(JobConf conf, File workDir, |
| TaskDistributedCacheManager taskDistributedCacheManager) |
| throws IOException { |
| // Accumulates class paths for child. |
| List<String> classPaths = new ArrayList<String>(); |
| // start with same classpath as parent process |
| appendSystemClasspaths(classPaths); |
| |
| // include the user specified classpath |
| appendJobJarClasspaths(conf.getJar(), classPaths); |
| |
| // Distributed cache paths |
| classPaths.addAll(taskDistributedCacheManager.getClassPaths()); |
| |
| // Include the working dir too |
| classPaths.add(workDir.toString()); |
| return classPaths; |
| } |
| |
| /** |
| * sets the environment variables needed for task jvm and its children. |
| * @param errorInfo |
| * @param workDir |
| * @param env |
| * @return |
| * @throws Throwable |
| */ |
| private String getVMEnvironment(String errorInfo, File workDir, JobConf conf, |
| Map<String, String> env, TaskAttemptID taskid, long logSize) |
| throws Throwable { |
| StringBuffer ldLibraryPath = new StringBuffer(); |
| ldLibraryPath.append(workDir.toString()); |
| String oldLdLibraryPath = null; |
| oldLdLibraryPath = System.getenv("LD_LIBRARY_PATH"); |
| if (oldLdLibraryPath != null) { |
| ldLibraryPath.append(SYSTEM_PATH_SEPARATOR); |
| ldLibraryPath.append(oldLdLibraryPath); |
| } |
| env.put("LD_LIBRARY_PATH", ldLibraryPath.toString()); |
| |
| // put jobTokenFile name into env |
| String jobTokenFile = conf.get(TokenCache.JOB_TOKENS_FILENAME); |
| LOG.debug("putting jobToken file name into environment fn=" + jobTokenFile); |
| env.put(UserGroupInformation.HADOOP_TOKEN_FILE_LOCATION, jobTokenFile); |
| |
| // for the child of task jvm, set hadoop.root.logger |
| env.put("HADOOP_ROOT_LOGGER", "INFO,TLA"); |
| String hadoopClientOpts = System.getenv("HADOOP_CLIENT_OPTS"); |
| if (hadoopClientOpts == null) { |
| hadoopClientOpts = ""; |
| } else { |
| hadoopClientOpts = hadoopClientOpts + " "; |
| } |
| hadoopClientOpts = hadoopClientOpts + "-Dhadoop.tasklog.taskid=" + taskid |
| + " -Dhadoop.tasklog.iscleanup=" + t.isTaskCleanupTask() |
| + " -Dhadoop.tasklog.totalLogFileSize=" + logSize; |
| env.put("HADOOP_CLIENT_OPTS", hadoopClientOpts); |
| |
| // add the env variables passed by the user |
| String mapredChildEnv = getChildEnv(conf); |
| if (mapredChildEnv != null && mapredChildEnv.length() > 0) { |
| String childEnvs[] = mapredChildEnv.split(","); |
| for (String cEnv : childEnvs) { |
| try { |
| String[] parts = cEnv.split("="); // split on '=' |
| String value = env.get(parts[0]); |
| if (value != null) { |
| // replace $env with the child's env constructed by tt's |
| // example LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/tmp |
| value = parts[1].replace("$" + parts[0], value); |
| } else { |
| // this key is not configured by the tt for the child .. get it |
| // from the tt's env |
| // example PATH=$PATH:/tmp |
| value = System.getenv(parts[0]); |
| if (value != null) { |
| // the env key is present in the tt's env |
| value = parts[1].replace("$" + parts[0], value); |
| } else { |
| // the env key is note present anywhere .. simply set it |
| // example X=$X:/tmp or X=/tmp |
| value = parts[1].replace("$" + parts[0], ""); |
| } |
| } |
| env.put(parts[0], value); |
| } catch (Throwable t) { |
| // set the error msg |
| errorInfo = "Invalid User environment settings : " + mapredChildEnv |
| + ". Failed to parse user-passed environment param." |
| + " Expecting : env1=value1,env2=value2..."; |
| LOG.warn(errorInfo); |
| throw t; |
| } |
| } |
| } |
| return errorInfo; |
| } |
| |
| /** |
| * Write the task specific job-configuration file. |
| * |
| * @param localFs |
| * @throws IOException |
| */ |
| private static void writeLocalTaskFile(String jobFile, JobConf conf) |
| throws IOException { |
| Path localTaskFile = new Path(jobFile); |
| FileSystem localFs = FileSystem.getLocal(conf); |
| localFs.delete(localTaskFile, true); |
| OutputStream out = localFs.create(localTaskFile); |
| try { |
| conf.writeXml(out); |
| } finally { |
| out.close(); |
| } |
| } |
| |
| /** |
| * Prepare the Configs.LOCAL_DIR for the child. The child is sand-boxed now. |
| * Whenever it uses LocalDirAllocator from now on inside the child, it will |
| * only see files inside the attempt-directory. This is done in the Child's |
| * process space. |
| */ |
| static void setupChildMapredLocalDirs(Task t, JobConf conf) { |
| String[] localDirs = conf.getTrimmedStrings(MRConfig.LOCAL_DIR); |
| String jobId = t.getJobID().toString(); |
| String taskId = t.getTaskID().toString(); |
| boolean isCleanup = t.isTaskCleanupTask(); |
| String user = t.getUser(); |
| StringBuffer childMapredLocalDir = |
| new StringBuffer(localDirs[0] + Path.SEPARATOR |
| + TaskTracker.getLocalTaskDir(user, jobId, taskId, isCleanup)); |
| for (int i = 1; i < localDirs.length; i++) { |
| childMapredLocalDir.append("," + localDirs[i] + Path.SEPARATOR |
| + TaskTracker.getLocalTaskDir(user, jobId, taskId, isCleanup)); |
| } |
| LOG.debug(MRConfig.LOCAL_DIR + " for child : " + childMapredLocalDir); |
| conf.set(MRConfig.LOCAL_DIR, childMapredLocalDir.toString()); |
| } |
| |
| /** Creates the working directory pathname for a task attempt. */ |
| static File formWorkDir(LocalDirAllocator lDirAlloc, |
| TaskAttemptID task, boolean isCleanup, JobConf conf) |
| throws IOException { |
| Path workDir = |
| lDirAlloc.getLocalPathToRead(TaskTracker.getTaskWorkDir( |
| conf.getUser(), task.getJobID().toString(), task.toString(), |
| isCleanup), conf); |
| |
| return new File(workDir.toString()); |
| } |
| |
| private static void appendSystemClasspaths(List<String> classPaths) { |
| for (String c : System.getProperty("java.class.path").split( |
| SYSTEM_PATH_SEPARATOR)) { |
| classPaths.add(c); |
| } |
| } |
| |
| /** |
| * Given a "jobJar" (typically retrieved via {@link JobConf#getJar()}), |
| * appends classpath entries for it, as well as its lib/ and classes/ |
| * subdirectories. |
| * |
| * @param jobJar Job jar from configuration |
| * @param classPaths Accumulator for class paths |
| */ |
| static void appendJobJarClasspaths(String jobJar, List<String> classPaths) { |
| if (jobJar == null) { |
| return; |
| |
| } |
| File jobCacheDir = new File(new Path(jobJar).getParent().toString()); |
| |
| // if jar exists, it into workDir |
| File[] libs = new File(jobCacheDir, "lib").listFiles(); |
| if (libs != null) { |
| for (File l : libs) { |
| classPaths.add(l.toString()); |
| } |
| } |
| classPaths.add(new File(jobCacheDir, "classes").toString()); |
| classPaths.add(new File(jobCacheDir, "job.jar").toString()); |
| } |
| |
| /** |
| * Creates distributed cache symlinks and tmp directory, as appropriate. |
| * Note that when we setup the distributed |
| * cache, we didn't create the symlinks. This is done on a per task basis |
| * by the currently executing task. |
| * |
| * @param conf The job configuration. |
| * @param workDir Working directory, which is completely deleted. |
| */ |
| public static void setupWorkDir(JobConf conf, File workDir) throws IOException { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Fully deleting contents of " + workDir); |
| } |
| |
| /** deletes only the contents of workDir leaving the directory empty. We |
| * can't delete the workDir as it is the current working directory. |
| */ |
| FileUtil.fullyDeleteContents(workDir); |
| |
| if (DistributedCache.getSymlink(conf)) { |
| URI[] archives = DistributedCache.getCacheArchives(conf); |
| URI[] files = DistributedCache.getCacheFiles(conf); |
| Path[] localArchives = DistributedCache.getLocalCacheArchives(conf); |
| Path[] localFiles = DistributedCache.getLocalCacheFiles(conf); |
| if (archives != null) { |
| for (int i = 0; i < archives.length; i++) { |
| String link = archives[i].getFragment(); |
| String target = localArchives[i].toString(); |
| symlink(workDir, target, link); |
| } |
| } |
| if (files != null) { |
| for (int i = 0; i < files.length; i++) { |
| String link = files[i].getFragment(); |
| String target = localFiles[i].toString(); |
| symlink(workDir, target, link); |
| } |
| } |
| } |
| |
| // For streaming, create extra symlinks (for all the files |
| // in the job cache dir) in the current working directory. |
| // Note that this is only executed if the configuration |
| // points to a jar file. |
| if (conf.getJar() != null) { |
| File jobCacheDir = new File( |
| new Path(conf.getJar()).getParent().toString()); |
| try{ |
| TrackerDistributedCacheManager.createAllSymlink(conf, jobCacheDir, |
| workDir); |
| } catch(IOException ie){ |
| // Do not exit even if symlinks have not been created. |
| LOG.warn(StringUtils.stringifyException(ie)); |
| } |
| } |
| |
| createChildTmpDir(workDir, conf); |
| } |
| |
| /** |
| * Utility method for creating a symlink and warning on errors. |
| * |
| * If link is null, does nothing. |
| */ |
| private static void symlink(File workDir, String target, String link) |
| throws IOException { |
| if (link != null) { |
| link = workDir.toString() + Path.SEPARATOR + link; |
| File flink = new File(link); |
| if (!flink.exists()) { |
| LOG.info(String.format("Creating symlink: %s <- %s", target, link)); |
| if (0 != FileUtil.symLink(target, link)) { |
| throw new IOException(String.format( |
| "Failed to create symlink: %s <- %s", target, link)); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Kill the child process |
| */ |
| public void kill() { |
| killed = true; |
| jvmManager.taskKilled(this); |
| signalDone(); |
| } |
| public void signalDone() { |
| synchronized (lock) { |
| done = true; |
| lock.notify(); |
| } |
| } |
| public void setExitCode(int exitCode) { |
| this.exitCodeSet = true; |
| this.exitCode = exitCode; |
| } |
| } |
