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

/**
 * Copyright 2005 The Apache Software Foundation
 *
 * Licensed 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 org.apache.hadoop.io.*;
import org.apache.hadoop.ipc.*;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.util.*;

import java.io.*;
import java.net.*;
import java.util.*;
import java.util.logging.*;

/** Runs a reduce task. */
class ReduceTaskRunner extends TaskRunner {
  private static final Logger LOG =
    LogFormatter.getLogger("org.apache.hadoop.mapred.ReduceTaskRunner");
  private MapOutputFile mapOutputFile;

  public ReduceTaskRunner(Task task, TaskTracker tracker, Configuration conf) {
    super(task, tracker, conf);
    this.mapOutputFile = new MapOutputFile();
    this.mapOutputFile.setConf(conf);
  }

  /** Assemble all of the map output files. */
  public boolean prepare() throws IOException {
    ReduceTask task = ((ReduceTask)getTask());
    this.mapOutputFile.removeAll(task.getTaskId());    // cleanup from failures
    String[][] mapTaskIds = task.getMapTaskIds();
    final Progress copyPhase = getTask().getProgress().phase();

    // we need input from every map task
    Vector needed = new Vector();
    for (int i = 0; i < mapTaskIds.length; i++) {
      needed.add(mapTaskIds[i]);
      copyPhase.addPhase();                       // add sub-phase per file
    }

    InterTrackerProtocol jobClient = getTracker().getJobClient();
    while (needed.size() > 0) {
      getTask().reportProgress(getTracker());

      // query for a just a random subset of needed segments so that we don't
      // overwhelm jobtracker.  ideally perhaps we could send a more compact
      // representation of all needed, i.e., a bit-vector
      Collections.shuffle(needed);
      int checkSize = Math.min(10, needed.size());
      String[][] neededStrings = new String[checkSize][];
      for (int i = 0; i < checkSize; i++) {
          neededStrings[i] = (String[]) needed.elementAt(i);
      }
      MapOutputLocation[] locs =
        jobClient.locateMapOutputs(task.getTaskId(), neededStrings);

      if (locs.length == 0) {
        try {
          if (killed) {
            return false;
          }
          Thread.sleep(10000);
        } catch (InterruptedException e) {
        }
        continue;
      }

      LOG.info(task.getTaskId()+" Got "+locs.length+" map output locations.");

      // try each of these locations
      for (int i = 0; i < locs.length; i++) {
        MapOutputLocation loc = locs[i];
        InetSocketAddress addr =
          new InetSocketAddress(loc.getHost(), loc.getPort());
        MapOutputProtocol client =
          (MapOutputProtocol)RPC.getProxy(MapOutputProtocol.class, addr, this.conf);

        this.mapOutputFile.setProgressReporter(new MapOutputFile.ProgressReporter() {
            public void progress(float progress) {
              copyPhase.phase().set(progress);
              try {
                getTask().reportProgress(getTracker());
              } catch (IOException e) {
                throw new RuntimeException(e);
              }
            }
          });

        getTask().reportProgress(getTracker());
        try {
          copyPhase.phase().setStatus(loc.toString());
          
          client.getFile(loc.getMapTaskId(), task.getTaskId(),
                         new IntWritable(task.getPartition()));

          // Success: remove from 'needed'
          boolean foundit = false;
          for (Iterator it = needed.iterator(); it.hasNext() && !foundit; ) {
              String idsForSingleMap[] = (String[]) it.next();
              for (int j = 0; j < idsForSingleMap.length; j++) {
                  if (idsForSingleMap[j].equals(loc.getMapTaskId())) {
                      it.remove();
                      foundit = true;
                      break;
                  }
              }
          }
          copyPhase.startNextPhase();
          
        } catch (IOException e) {                 // failed: try again later
          LOG.log(Level.WARNING,
                  task.getTaskId()+" copy failed: "
                  +loc.getMapTaskId()+" from "+addr,
                  e);
        } finally {
          this.mapOutputFile.setProgressReporter(null);
        }
      }
    }
    getTask().reportProgress(getTracker());
    return true;
  }

  /** Delete all of the temporary map output files. */
  public void close() throws IOException {
    getTask().getProgress().setStatus("closed");
    this.mapOutputFile.removeAll(getTask().getTaskId());
  }

}