216.58.210.78 - - [27/Feb/2015:17:55:11 +0000] "GET /2018/01/01/thread-local-storage/ HTTP/1.1"
200 3377 "http://blog.hacksmeta.com/" "Mozilla/5.0 (Macintosh; Intel Mac OS X
10_9_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/40.0.2214.115
Safari/537.36"

$remote_addr - $remote_user [$time_local] "$request"
$status $body_bytes_sent "$http_referer" "$http_user_agent"

cat /var/log/nginx/access.log |
 awk '{print $7}'   |
 sort               |
 uniq -c            |
 sort -r -n         |
 head -n 5

counter = {}
with open('/var/log/nginx/access.log') as f:
    for line in f:
        fields = line.split(' ')
        url = fields[7]
        counter[url] += 1
top5 = sorted(countser.items(), key=lambda x:x[1], reverse=True)[5]
for key, value in top5:
    print key, value

//旧API作业配置实例：
JobConf job=new JobConf（new Configuration()， MyJob.class） ；
job.setJobName（"myjob"） ；
job.setMapperClass（MyJob.MyMapper.class） ；
job.setReducerClass（MyJob.MyReducer.class） ；
JobClient.runJob（job） ；
//新API作业配置实例：
Configuration conf=new Configuration()；
Job job=new Job（conf， "myjob"） ；
job.setJarByClass（MyJob.class） ；
job.setMapperClass（MyJob.MyMapper.class） ；
job.setReducerClass（MyJob.MyReducer.class） ；
System.exit（job.waitForCompletion（true） ?0： 1） ；

//将wc_mapper封装成一个进程
ProcessBuilder builder=new ProcessBuilder("wc_mapper");
builder.environment().putAll(childEnv.toMap()); //设置环境变量
sim=builder.start();
//创建标准输出流
clientOut_=new DataOutputStream(new BufferedOutputStream(
sim.getOutputStream(), BUFFER_SIZE));
//创建标准输入流
clientIn_=new DataInputStream(new BufferedInputStream(
sim.getInputStream(), BUFFER_SIZE));
//创建标准错误流
clientErr_=new DataInputStream(new BufferedInputStream(sim.getErrorStream()));

$HADOOP_HOME/bin/hadoop jar xxx.jar\
-D mapred.job.name="xxx"\
-D mapred.reduce.tasks=2\
-files=blacklist.txt, whitelist.txt\
-libjars=third-party.jar\
-archives=dictionary.zip\
-input/test/input\
-output/test/output

  #core commands  
  *)
    # the core commands
    ...
    elif [ "$COMMAND" = "jar" ] ; then
      CLASS=org.apache.hadoop.util.RunJar
      if [[ -n "${YARN_OPTS}" ]] || [[ -n "${YARN_CLIENT_OPTS}" ]]; then
        echo "WARNING: Use \"yarn jar\" to launch YARN applications." 1>&2
      fi
    ...
    export CLASSPATH=$CLASSPATH
    exec "$JAVA" $JAVA_HEAP_MAX $HADOOP_OPTS $CLASS "$@"
    ;;

  /** Run a Hadoop job jar.  If the main class is not in the jar's manifest,
   * then it must be provided on the command line. */
  public static void main(String[] args) throws Throwable {
    String usage = "RunJar jarFile [mainClass] args...";

    if (args.length < 1) {
      System.err.println(usage);
      System.exit(-1);
    }

    int firstArg = 0;
    String fileName = args[firstArg++];
    File file = new File(fileName);
    String mainClassName = null;

    JarFile jarFile;
    try {
      jarFile = new JarFile(fileName);
    } catch(IOException io) {
      throw new IOException("Error opening job jar: " + fileName)
        .initCause(io);
    }
    // 解析参数，获取主类名称
    Manifest manifest = jarFile.getManifest();
    if (manifest != null) {
      mainClassName = manifest.getMainAttributes().getValue("Main-Class");
    }
    jarFile.close();

    if (mainClassName == null) {
      if (args.length < 2) {
        System.err.println(usage);
        System.exit(-1);
      }
      mainClassName = args[firstArg++];
    }
    mainClassName = mainClassName.replaceAll("/", ".");
    // 创建临时目录，用来存放解压 jar 包类文件
    File tmpDir = new File(new Configuration().get("hadoop.tmp.dir"));
    tmpDir.mkdirs();
    if (!tmpDir.isDirectory()) { 
      System.err.println("Mkdirs failed to create " + tmpDir);
      System.exit(-1);
    }
    final File workDir = File.createTempFile("hadoop-unjar", "", tmpDir);
    workDir.delete();
    workDir.mkdirs();
    if (!workDir.isDirectory()) {
      System.err.println("Mkdirs failed to create " + workDir);
      System.exit(-1);
    }

    // 设置虚拟机hook，关闭时删除临时文件
    Runtime.getRuntime().addShutdownHook(new Thread() {
        public void run() {
          try {
            FileUtil.fullyDelete(workDir);
          } catch (IOException e) {
          }
        }
      });

    // 解压
    unJar(file, workDir);

    // 设置环境变量，ClassPath
    ArrayList<URL> classPath = new ArrayList<URL>();
    classPath.add(new File(workDir+"/").toURL());
    classPath.add(file.toURL());
    classPath.add(new File(workDir, "classes/").toURL());
    File[] libs = new File(workDir, "lib").listFiles();
    if (libs != null) {
      for (int i = 0; i < libs.length; i++) {
        classPath.add(libs[i].toURL());
      }
    }

    ClassLoader loader =
      new URLClassLoader(classPath.toArray(new URL[0]));
    // 加载主类，并利用反射开始运行main方法
    Thread.currentThread().setContextClassLoader(loader);
    Class<?> mainClass = Class.forName(mainClassName, true, loader);
    Method main = mainClass.getMethod("main", new Class[] {
      Array.newInstance(String.class, 0).getClass()
    });
    String[] newArgs = Arrays.asList(args)
      .subList(firstArg, args.length).toArray(new String[0]);
    try {
      main.invoke(null, new Object[] { newArgs });
    } catch (InvocationTargetException e) {
      throw e.getTargetException();
    }
  }

  if (useNewApi) {
    runNewMapper(job, splitMetaInfo, umbilical, reporter);
  } else {
    runOldMapper(job, splitMetaInfo, umbilical, reporter);
  }
  done(umbilical, reporter);

// 生产者 TaskTracker
public void addToTaskQueue(LaunchTaskAction action) {
  synchronized (tasksToLaunch) {
    TaskInProgress tip = registerTask(action, this);
    tasksToLaunch.add(tip);
    tasksToLaunch.notifyAll();
  }
}
private TaskInProgress registerTask(LaunchTaskAction action, 
      TaskLauncher launcher) {
  Task t = action.getTask();
  LOG.info("LaunchTaskAction (registerTask): " + t.getTaskID() +
            " task's state:" + t.getState());
  TaskInProgress tip = new TaskInProgress(t, this.fConf, launcher);
  synchronized (this) {
    tasks.put(t.getTaskID(), tip);
    runningTasks.put(t.getTaskID(), tip);
    boolean isMap = t.isMapTask();
    if (isMap) {
      mapTotal++;
    } else {
      reduceTotal++;
    }
  }
  return tip;
}

// 消费者 TaskLauncher
public void run() {
  while (!Thread.interrupted()) {
      TaskInProgress tip;
      Task task;
      synchronized (tasksToLaunch) {
        while (tasksToLaunch.isEmpty()) {
          tasksToLaunch.wait();
        }
        //get the TIP
        tip = tasksToLaunch.remove(0);
        task = tip.getTask();
        LOG.info("Trying to launch : " + tip.getTask().getTaskID() + 
                  " which needs " + task.getNumSlotsRequired() + " slots");
      }
      //wait for free slots to run
      synchronized (numFreeSlots) {
        boolean canLaunch = true;
        while (numFreeSlots.get() < task.getNumSlotsRequired()) {
          //Make sure that there is no kill task action for this task!
          //We are not locking tip here, because it would reverse the
          //locking order!
          //Also, Lock for the tip is not required here! because :
          // 1. runState of TaskStatus is volatile
          // 2. Any notification is not missed because notification is
          // synchronized on numFreeSlots. So, while we are doing the check,
          // if the tip is half way through the kill(), we don't miss
          // notification for the following wait().
          if (!tip.canBeLaunched()) {
            //got killed externally while still in the launcher queue
            LOG.info("Not blocking slots for " + task.getTaskID()
                + " as it got killed externally. Task's state is "
                + tip.getRunState());
            canLaunch = false;
            break;
          }
          LOG.info("TaskLauncher : Waiting for " + task.getNumSlotsRequired() + 
                    " to launch " + task.getTaskID() + ", currently we have " + 
                    numFreeSlots.get() + " free slots");
          numFreeSlots.wait();
        }
        if (!canLaunch) {
          continue;
        }
        LOG.info("In TaskLauncher, current free slots : " + numFreeSlots.get()+
                  " and trying to launch "+tip.getTask().getTaskID() + 
                  " which needs " + task.getNumSlotsRequired() + " slots");
        numFreeSlots.set(numFreeSlots.get() - task.getNumSlotsRequired());
        assert (numFreeSlots.get() >= 0);
      }
      synchronized (tip) {
        //to make sure that there is no kill task action for this
        if (!tip.canBeLaunched()) {
          //got killed externally while still in the launcher queue
          LOG.info("Not launching task " + task.getTaskID() + " as it got"
            + " killed externally. Task's state is " + tip.getRunState());
          addFreeSlots(task.getNumSlotsRequired());
          continue;
        }
        tip.slotTaken = true;
      }
      //got a free slot. launch the task
      startNewTask(tip);
  }
}

jvmManager.launchJvm(...)
synchronized (lock) {
  while (!done) {
    lock.wait();
  }
}

// MapTask.run()
if (useNewApi) {
  runNewMapper(job, splitMetaInfo, umbilical, reporter);
} else {
  runOldMapper(job, splitMetaInfo, umbilical, reporter);
}
done(umbilical, reporter);


// Task.done
if (commitRequired) {
  int retries = MAX_RETRIES;
  setState(TaskStatus.State.COMMIT_PENDING);
  // say the task tracker that task is commit pending
  while (true) {
    try {
      umbilical.commitPending(taskId, taskStatus, jvmContext);
      break;
    } catch (IOException ie) {
      if (--retries == 0) {
        System.exit(67);
      }
    }
  }
  commit(umbilical, reporter, committer);
}
taskDone.set(true);
reporter.stopCommunicationThread();
sendLastUpdate(umbilical);
//signal the tasktracker that we are done
sendDone(umbilical);


// Task.sendDone
int retries = MAX_RETRIES;
while (true) {
  try {
    umbilical.done(getTaskID(), jvmContext);
    return;
  } catch (IOException ie) {
    if (--retries == 0) {
      throw ie;
    }
  }
}

//TaskTracker.done
validateJVM(tip, jvmContext, taskid);
commitResponses.remove(taskid);
tip.reportDone();
//-----------------------------------------
// TaskTracker.TaskInProgress.reportDone()
this.taskStatus.setProgress(1.0f);
this.taskStatus.setFinishTime(System.currentTimeMillis());
this.done = true;
jvmManager.taskFinished(runner);
runner.signalDone();

private synchronized void reapJvm( 
    TaskRunner t, JvmEnv env) throws IOException, InterruptedException {
  if (t.getTaskInProgress().wasKilled()) {
    //the task was killed in-flight
    //no need to do the rest of the operations
    return;
  }
  boolean spawnNewJvm = false;
  JobID jobId = t.getTask().getJobID();
  //Check whether there is a free slot to start a new JVM.
  //,or, Kill a (idle) JVM and launch a new one
  //When this method is called, we *must* 
  // (1) spawn a new JVM (if we are below the max) 
  // (2) find an idle JVM (that belongs to the same job), or,
  // (3) kill an idle JVM (from a different job) 
  // (the order of return is in the order above)
  int numJvmsSpawned = jvmIdToRunner.size();
  JvmRunner runnerToKill = null;
  if (numJvmsSpawned >= maxJvms) {
    //go through the list of JVMs for all jobs.
    Iterator<Map.Entry<JVMId, JvmRunner>> jvmIter = 
      jvmIdToRunner.entrySet().iterator();

    while (jvmIter.hasNext()) {
      JvmRunner jvmRunner = jvmIter.next().getValue();
      JobID jId = jvmRunner.jvmId.getJobId();
      //look for a free JVM for this job; if one exists then just break
      if (jId.equals(jobId) && !jvmRunner.isBusy() && !jvmRunner.ranAll()){
        setRunningTaskForJvm(jvmRunner.jvmId, t); //reserve the JVM
        LOG.info("No new JVM spawned for jobId/taskid: " + 
                  jobId+"/"+t.getTask().getTaskID() +
                  ". Attempting to reuse: " + jvmRunner.jvmId);
        return;
      }
      //Cases when a JVM is killed: 
      // (1) the JVM under consideration belongs to the same job 
      //     (passed in the argument). In this case, kill only when
      //     the JVM ran all the tasks it was scheduled to run (in terms
      //     of count).
      // (2) the JVM under consideration belongs to a different job and is
      //     currently not busy
      //But in both the above cases, we see if we can assign the current
      //task to an idle JVM (hence we continue the loop even on a match)
      if ((jId.equals(jobId) && jvmRunner.ranAll()) ||
          (!jId.equals(jobId) && !jvmRunner.isBusy())) {
        runnerToKill = jvmRunner;
        spawnNewJvm = true;
      }
    }
  } else {
    spawnNewJvm = true;
  }

  if (spawnNewJvm) {
    if (runnerToKill != null) {
      killJvmRunner(runnerToKill);
    }
    spawnNewJvm(jobId, env, t);
    return;
  }
}

// Child.main
while (true) {
  ......
  JvmTask myTask = umbilical.getTask(context);
  ......
  final Task taskFinal = task;
  childUGI.doAs(new PrivilegedExceptionAction<Object>() {
    public Object run() throws Exception {
      try {
        // use job-specified working directory
        FileSystem.get(job).setWorkingDirectory(job.getWorkingDirectory());
        taskFinal.run(job, umbilical);        // run the task
      } finally {
      }
      return null;
    }
  });
}

// ReduceTask.run
boolean isLocal = "local".equals(job.get("mapred.job.tracker", "local"));
if (!isLocal) {
  reduceCopier = new ReduceCopier(umbilical, job, reporter);
  if (!reduceCopier.fetchOutputs()) {
    if(reduceCopier.mergeThrowable instanceof FSError) {
      throw (FSError)reduceCopier.mergeThrowable;
    }
    throw new IOException("Task: " + getTaskID() + 
        " - The reduce copier failed", reduceCopier.mergeThrowable);
  }
}
copyPhase.complete();                         // copy is already complete
setPhase(TaskStatus.Phase.SORT);
statusUpdate(umbilical);

final FileSystem rfs = FileSystem.getLocal(job).getRaw();
RawKeyValueIterator rIter = isLocal
  ? Merger.merge(job, rfs, job.getMapOutputKeyClass(),
      job.getMapOutputValueClass(), codec, getMapFiles(rfs, true),
      !conf.getKeepFailedTaskFiles(), job.getInt("io.sort.factor", 100),
      new Path(getTaskID().toString()), job.getOutputKeyComparator(),
      reporter, spilledRecordsCounter, null)
  : reduceCopier.createKVIterator(job, rfs, reporter);

// free up the data structures
mapOutputFilesOnDisk.clear();

sortPhase.complete();                         // sort is complete
setPhase(TaskStatus.Phase.REDUCE); 
statusUpdate(umbilical);
Class keyClass = job.getMapOutputKeyClass();
Class valueClass = job.getMapOutputValueClass();
RawComparator comparator = job.getOutputValueGroupingComparator();

// MapEventsFetcherThread run loop
while (running) {
  try {
    List <FetchStatus> fList = null;
    synchronized (runningJobs) {
      while (((fList = reducesInShuffle()).size()) == 0) {
        runningJobs.wait();
      }
    }
    // now fetch all the map task events for all the reduce tasks
    // possibly belonging to different jobs
    boolean fetchAgain = false; //flag signifying whether we want to fetch
                                //immediately again.
    for (FetchStatus f : fList) {
      long currentTime = System.currentTimeMillis();
      //the method below will return true when we have not 
      //fetched all available events yet
      if (f.fetchMapCompletionEvents(currentTime)) {
         fetchAgain = true;
      }
      if (!running) {
        break;
      }
    }
    synchronized (waitingOn) {
      if (!fetchAgain) {
        waitingOn.wait(heartbeatInterval);
      }
    }
}

// FetchStatus.fetchMapCompletionEvents
public boolean fetchMapCompletionEvents(long currTime) throws IOException {
  .....
  synchronized (fromEventId) {
    currFromEventId = fromEventId.get();
    List <TaskCompletionEvent> recentMapEvents = 
      queryJobTracker(fromEventId, jobId, jobClient);
    synchronized (allMapEvents) {
      allMapEvents.addAll(recentMapEvents);
    }
    lastFetchTime = currTime;
    if (fromEventId.get() - currFromEventId >= probe_sample_size) {
      //return true when we have fetched the full payload, indicating
      //that we should fetch again immediately (there might be more to
      //fetch
      fetchAgain = true;
      return true;
    }
  }
  fetchAgain = false;
  ......
}


// TaskTracker.queryJobTracker

private List<TaskCompletionEvent> queryJobTracker(IntWritable fromEventId,
                                                  JobID jobId,
                                                  InterTrackerProtocol jobClient)
  throws IOException {

  TaskCompletionEvent t[] = jobClient.getTaskCompletionEvents(
                                                              jobId,
                                                              fromEventId.get(),
                                                              probe_sample_size);
  //we are interested in map task completion events only. So store
  //only those
  List <TaskCompletionEvent> recentMapEvents = 
    new ArrayList<TaskCompletionEvent>();
  for (int i = 0; i < t.length; i++) {
    if (t[i].isMap) {
      recentMapEvents.add(t[i]);
    }
  }
  fromEventId.set(fromEventId.get() + t.length);
  return recentMapEvents;
}

// 生产者
MapTaskCompletionEventsUpdate update = 
          umbilical.getMapCompletionEvents(reduceTask.getJobID(), 
                                           fromEventId.get(), 
                                           MAX_EVENTS_TO_FETCH,
                                           reduceTask.getTaskID(), jvmContext);
TaskCompletionEvent events[] = update.getMapTaskCompletionEvents();
......
for (TaskCompletionEvent event : events) {
  switch (event.getTaskStatus()) {
    case SUCCEEDED:
    {
      URI u = URI.create(event.getTaskTrackerHttp());
      String host = u.getHost();
      TaskAttemptID taskId = event.getTaskAttemptId();
      URL mapOutputLocation = new URL(event.getTaskTrackerHttp() + 
                              "/mapOutput?job=" + taskId.getJobID() +
                              "&map=" + taskId + 
                              "&reduce=" + getPartition());
      List<MapOutputLocation> loc = mapLocations.get(host);
      if (loc == null) {
        loc = Collections.synchronizedList
          (new LinkedList<MapOutputLocation>());
        mapLocations.put(host, loc);
        }
      loc.add(new MapOutputLocation(taskId, host, mapOutputLocation));
      numNewMaps ++;
    }
......
  }
}

// TaskTracker.getMapCompletionEvents --> FetchStatus.getMapEvents
public TaskCompletionEvent[] getMapEvents(int fromId, int max) {
  synchronized (allMapEvents) {
    if (allMapEvents.size() > fromId) {
      int actualMax = Math.min(max, (allMapEvents.size() - fromId));
      List <TaskCompletionEvent> eventSublist = 
        allMapEvents.subList(fromId, actualMax + fromId);
      mapEvents = eventSublist.toArray(mapEvents);
    } else {
      // Notify Fetcher thread. 
      notifyFetcher = true;
    }
  }
  if (notifyFetcher) {
  synchronized (waitingOn) {
    waitingOn.notify();
  }
}

// 消费者 
while (true) {
  ......
  MapOutputLocation loc = null;
  long size = -1;
  synchronized (scheduledCopies) {
    while (scheduledCopies.isEmpty()) {
      scheduledCopies.wait();
    }
    loc = scheduledCopies.remove(0);
  }
  ......
  shuffleClientMetrics.threadBusy();
  start(loc);
  size = copyOutput(loc);
  shuffleClientMetrics.successFetch();
}

private long copyOutput(MapOutputLocation loc
                              ) throws IOException, InterruptedException {
  ......
  // Copy the map output
  MapOutput mapOutput = getMapOutput(loc, tmpMapOutput,
                                      reduceId.getTaskID().getId());

    // Process map-output
    if (mapOutput.inMemory) {
      // Save it in the synchronized list of map-outputs
      mapOutputsFilesInMemory.add(mapOutput);
    } else {
    ......
      synchronized (mapOutputFilesOnDisk) {        
        addToMapOutputFilesOnDisk(localFileSys.getFileStatus(filename));
      }
    }
}


private MapOutput getMapOutput(MapOutputLocation mapOutputLoc, 
                                     Path filename, int reduce){
  ......
  boolean shuffleInMemory = ramManager.canFitInMemory(decompressedLength); 
  // Shuffle
  MapOutput mapOutput = null;
  if (shuffleInMemory) {
    mapOutput = shuffleInMemory(mapOutputLoc, connection, input,
                                (int)decompressedLength,
                                (int)compressedLength);
  } else {
    mapOutput = shuffleToDisk(mapOutputLoc, input, filename, 
        compressedLength);
  }
}
private MapOutput shuffleInMemory(MapOutputLocation mapOutputLoc,
                                        URLConnection connection, 
                                        InputStream input,
                                        int mapOutputLength,
                                        int compressedLength)
      throws IOException, InterruptedException {
  // Reserve ram for the map-output
  boolean createdNow = ramManager.reserve(mapOutputLength, input);
......
  // Inform the ram-manager
  ramManager.closeInMemoryFile(mapOutputLength);
  ramManager.unreserve(mapOutputLength);
}

// TaskReporter 线程
 public void run() {
  final int MAX_RETRIES = 3;
  int remainingRetries = MAX_RETRIES;
  // get current flag value and reset it as well
  boolean sendProgress = resetProgressFlag();
  while (!taskDone.get()) {
    synchronized(lock) {
      done = false;
    }
......
    if (sendProgress) {
      // we need to send progress update
      updateCounters();
      taskStatus.statusUpdate(taskProgress.get(),
                              taskProgress.toString(), 
                              counters);
      taskFound = umbilical.statusUpdate(taskId, taskStatus, jvmContext);
      taskStatus.clearStatus();
    }
    else {
      // send ping 
      taskFound = umbilical.ping(taskId, jvmContext);
    }
....
  }
  //Notify that we are done with the work
  resetDoneFlag();
}

public class Progress {
  private String status = "";
  private float progress;
  private int currentPhase;
  private ArrayList<Progress> phases = new ArrayList<Progress>();
  private Progress parent;
  private float progressPerPhase;

  /** Creates a new root node. */
  public Progress() {}

  /** Adds a named node to the tree. */
  public Progress addPhase(String status) {
    Progress phase = addPhase();
    phase.setStatus(status);
    return phase;
  }

  /** Adds a node to the tree. */
  public synchronized Progress addPhase() {
    Progress phase = new Progress();
    phases.add(phase);
    phase.parent = this;
    progressPerPhase = 1.0f / (float)phases.size();
    return phase;
  }

  /** Called during execution to move to the next phase at this level in the
   * tree. */
  public synchronized void startNextPhase() {
    currentPhase++;
  }

  /** Returns the current sub-node executing. */
  public synchronized Progress phase() {
    return phases.get(currentPhase);
  }

  /** Completes this node, moving the parent node to its next child. */
  public void complete() {
    // we have to traverse up to our parent, so be careful about locking.
    Progress myParent;
    synchronized(this) {
      progress = 1.0f;
      myParent = parent;
    }
    if (myParent != null) {
      // this will synchronize on the parent, so we make sure we release
      // our lock before getting the parent's, since we're traversing 
      // against the normal traversal direction used by get() or toString().
      // We don't need transactional semantics, so we're OK doing this. 
      myParent.startNextPhase();
    }
  }

  /** Called during execution on a leaf node to set its progress. */
  public synchronized void set(float progress) {
    this.progress = progress;
  }

  /** Returns the overall progress of the root. */
  // this method probably does not need to be synchronized as getINternal() is synchronized 
  // and the node's parent never changes. Still, it doesn't hurt. 
  public synchronized float get() {
    Progress node = this;
    while (node.parent != null) {                 // find the root
      node = parent;
    }
    return node.getInternal();
  }

  /** Computes progress in this node. */
  private synchronized float getInternal() {
    int phaseCount = phases.size();
    if (phaseCount != 0) {
      float subProgress =
        currentPhase < phaseCount ? phase().getInternal() : 0.0f;
      return progressPerPhase*(currentPhase + subProgress);
    } else {
      return progress;
    }
  }

  public synchronized void setStatus(String status) {
    this.status = status;
  }

  public String toString() {
    StringBuffer result = new StringBuffer();
    toString(result);
    return result.toString();
  }

  private synchronized void toString(StringBuffer buffer) {
    buffer.append(status);
    if (phases.size() != 0 && currentPhase < phases.size()) {
      buffer.append(" > ");
      phase().toString(buffer);
    }
  }

}

     /** Computes progress in this node. */
  private synchronized float getInternal() {
    int phaseCount = phases.size();
    if (phaseCount != 0) {
      float subProgress =
        currentPhase < phaseCount ? phase().getInternal() : 0.0f;
      return progressPerPhase*(currentPhase + subProgress);
    } else {
      return progress;
    }
  }