HADOOP-830. Improve mapreduce merge performance by buffering and merging map outputs as they arrive at reduce nodes. Contributed by Devaraj.

git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@497891 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -47,6 +47,10 @@ Trunk (unreleased changes)
 14. HADOOP-735.  Switch generated record code to use BytesWritable to
     represent fields of type 'buffer'. (Milind Bhandarkar via cutting)
 
+15. HADOOP-830.  Improve mapreduce merge performance by buffering and
+    merging multiple map outputs as they arrive at reduce nodes before
+    they're written to disk.  (Devaraj Das via cutting)
+
 
 Release 0.10.1 - 2007-01-10
 

conf/hadoop-default.xml

@@ -124,6 +124,18 @@ creations/deletions), or "all".</description>
   <description>The FileSystem for s3: uris.</description>
 </property>
 
+<property>
+  <name>fs.ramfs.impl</name>
+  <value>org.apache.hadoop.fs.InMemoryFileSystem</value>
+  <description>The FileSystem for ramfs: uris.</description>
+</property>
+
+<property>
+  <name>fs.inmemory.size.mb</name>
+  <value>75</value>
+  <description>The size of the in-memory filsystem instance in MB</description>
+</property>
+
 <property>
   <name>dfs.datanode.bindAddress</name>
   <value>0.0.0.0</value>

src/java/org/apache/hadoop/fs/FSDataOutputStream.java

@@ -177,4 +177,11 @@ public class FSDataOutputStream extends DataOutputStream {
     return ((Buffer)out).getPos();
   }
 
+  public static long getChecksumLength(long size, int bytesPerSum) {
+    //the checksum length is equal to size passed divided by bytesPerSum +
+    //bytes written in the beginning of the checksum file.  
+    return ((long)(Math.ceil((float)size/bytesPerSum)) + 1) * 4 + 
+            CHECKSUM_VERSION.length;  
+  }
+  
 }

src/java/org/apache/hadoop/fs/FileSystem.java

@@ -200,6 +200,13 @@ public abstract class FileSystem extends Configured {
       return new Path(file.getParent(), "."+file.getName()+".crc");
     }
 
+    /** Return the length of the checksum file given the size of the 
+     * actual file.
+     **/
+    public static long getChecksumFileLength(long fileSize, int bytesPerSum) {
+      return FSDataOutputStream.getChecksumLength(fileSize, bytesPerSum);
+    }
+    
     /** Return true iff file is a checksum file name.*/
     public static boolean isChecksumFile(Path file) {
       String name = file.getName();

src/java/org/apache/hadoop/fs/InMemoryFileSystem.java

@@ -0,0 +1,407 @@
+/**
+ * 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.fs;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.net.URI;
+import java.util.*;
+
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.util.Progressable;
+
+/** An implementation of the in-memory filesystem. This implementation assumes
+ * that the file lengths are known ahead of time and the total lengths of all
+ * the files is below a certain number (like 100 MB, configurable). Use the API
+ * reserveSpaceWithCheckSum(Path f, int size) (see below for a description of
+ * the API for reserving space in the FS. The uri of this filesystem starts with
+ * ramfs:// .
+ * @author ddas
+ *
+ */
+public class InMemoryFileSystem extends FileSystem {
+  private URI uri;
+  private int fsSize;
+  private volatile int totalUsed;
+  private Path staticWorkingDir;
+  private int bytesPerSum;
+  
+  //pathToFileAttribs is the final place where a file is put after it is closed
+  private Map <String, FileAttributes> pathToFileAttribs = 
+    Collections.synchronizedMap(new HashMap());
+  
+  //tempFileAttribs is a temp place which is updated while reserving memory for
+  //files we are going to create. It is read in the createRaw method and the
+  //temp key/value is discarded. If the file makes it to "close", then it
+  //ends up being in the pathToFileAttribs map.
+  private Map <String, FileAttributes> tempFileAttribs = 
+    Collections.synchronizedMap(new HashMap());
+  
+  public InMemoryFileSystem() {}
+  
+  public InMemoryFileSystem(URI uri, Configuration conf) {
+    initialize(uri, conf);
+  }
+  
+  //inherit javadoc
+  public void initialize(URI uri, Configuration conf) {
+    int size = Integer.parseInt(conf.get("fs.inmemory.size.mb", "100"));
+    this.fsSize = size * 1024 * 1024;
+    this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority());
+    this.staticWorkingDir = new Path(this.uri.getPath());
+    this.bytesPerSum = conf.getInt("io.bytes.per.checksum", 512);
+    LOG.info("Initialized InMemoryFileSystem: " + uri.toString() + 
+             " of size (in bytes): " + fsSize);
+  }
+
+  //inherit javadoc
+  public URI getUri() {
+    return uri;
+  }
+
+  /** @deprecated */
+  public String getName() {
+    return uri.toString();
+  }
+
+  /**
+   * Return 1x1 'inmemory' cell if the file exists.
+   * Return null if otherwise.
+   */
+  public String[][] getFileCacheHints(Path f, long start, long len)
+      throws IOException {
+    if (! exists(f)) {
+      return null;
+    } else {
+      return new String[][] {{"inmemory"}};
+    }
+  }
+
+  private class InMemoryInputStream extends FSInputStream {
+    private DataInputBuffer din = new DataInputBuffer();
+    private FileAttributes fAttr;
+    
+    public InMemoryInputStream(Path f) throws IOException {
+      fAttr = pathToFileAttribs.get(getPath(f));
+      if (fAttr == null) throw new FileNotFoundException("File " + f + 
+                                                         " does not exist");
+      din.reset(fAttr.data, 0, fAttr.size);
+    }
+    
+    public long getPos() throws IOException {
+      return din.getPosition();
+    }
+    
+    public void seek(long pos) throws IOException {
+      if ((int)pos > fAttr.size)
+        throw new IOException("Cannot seek after EOF");
+      din.reset(fAttr.data, (int)pos, fAttr.size - (int)pos);
+    }
+    
+    public int available() throws IOException {
+      return din.available(); 
+    }
+    public boolean markSupport() { return false; }
+
+    public int read() throws IOException {
+      return din.read();
+    }
+
+    public int read(byte[] b, int off, int len) throws IOException {
+      return din.read(b, off, len);
+    }
+    
+    public long skip(long n) throws IOException { return din.skip(n); }
+  }
+
+  public FSInputStream openRaw(Path f) throws IOException {
+    return new InMemoryInputStream(f);
+  }
+
+  private class InMemoryOutputStream extends FSOutputStream {
+    private int count;
+    private FileAttributes fAttr;
+    private Path f;
+    
+    public InMemoryOutputStream(Path f, FileAttributes fAttr) 
+    throws IOException {
+      this.fAttr = fAttr;
+      this.f = f;
+    }
+    
+    public long getPos() throws IOException {
+      return count;
+    }
+    
+    public void close() throws IOException {
+      synchronized (InMemoryFileSystem.this) {
+        pathToFileAttribs.put(getPath(f), fAttr);
+      }
+    }
+    
+    public void write(byte[] b, int off, int len) throws IOException {
+      if ((off < 0) || (off > b.length) || (len < 0) ||
+          ((off + len) > b.length) || ((off + len) < 0)) {
+        throw new IndexOutOfBoundsException();
+      } else if (len == 0) {
+        return;
+      }
+      int newcount = count + len;
+      if (newcount > fAttr.size) {
+        throw new IOException("Insufficient space");
+      }
+      System.arraycopy(b, off, fAttr.data, count, len);
+      count = newcount;
+    }
+    
+    public void write(int b) throws IOException {
+      int newcount = count + 1;
+      if (newcount > fAttr.size) {
+        throw new IOException("Insufficient space");
+      }
+      fAttr.data[count] = (byte)b;
+      count = newcount;
+    }
+  }
+  
+  public FSOutputStream createRaw(Path f, boolean overwrite, short replication,
+      long blockSize) throws IOException {
+    if (exists(f) && ! overwrite) {
+      throw new IOException("File already exists:"+f);
+    }
+    synchronized (this) {
+      FileAttributes fAttr =(FileAttributes)tempFileAttribs.remove(getPath(f));
+      if (fAttr != null)
+        return createRaw(f, fAttr);
+      return null;
+    }
+  }
+
+  public FSOutputStream createRaw(Path f, boolean overwrite, short replication,
+      long blockSize, Progressable progress) throws IOException {
+    //ignore write-progress reporter for in-mem files
+    return createRaw(f, overwrite, replication, blockSize);
+  }
+
+  public FSOutputStream createRaw(Path f, FileAttributes fAttr) 
+  throws IOException {
+    //the path is not added into the filesystem (in the pathToFileAttribs
+    //map) until close is called on the outputstream that this method is 
+    //going to return
+    //Create an output stream out of data byte array
+    return new InMemoryOutputStream(f, fAttr);
+  }
+
+  public void close() throws IOException {
+    super.close();
+    if (pathToFileAttribs != null) 
+      pathToFileAttribs.clear();
+    pathToFileAttribs = null;
+    if (tempFileAttribs != null)
+      tempFileAttribs.clear();
+    tempFileAttribs = null;
+  }
+
+  /**
+   * Replication is not supported for the inmemory file system.
+   */
+  public short getReplication(Path src) throws IOException {
+    return 1;
+  }
+
+  public boolean setReplicationRaw(Path src, short replication)
+      throws IOException {
+    return true;
+  }
+
+  public boolean renameRaw(Path src, Path dst) throws IOException {
+    synchronized (this) {
+      FileAttributes fAttr = pathToFileAttribs.remove(getPath(src));
+      if (fAttr == null) return false;
+      pathToFileAttribs.put(getPath(dst), fAttr);
+      return true;
+    }
+  }
+
+  public boolean deleteRaw(Path f) throws IOException {
+    synchronized (this) {
+      FileAttributes fAttr = pathToFileAttribs.remove(getPath(f));
+      if (fAttr != null) {
+        fAttr.data = null;
+        totalUsed -= fAttr.size;
+        return true;
+      }
+      return false;
+    }
+  }
+
+  public boolean exists(Path f) throws IOException {
+    return pathToFileAttribs.containsKey(getPath(f));
+  }
+  
+  /**
+   * Directory operations are not supported
+   */
+  public boolean isDirectory(Path f) throws IOException {
+    return false;
+  }
+
+  public long getLength(Path f) throws IOException {
+    return pathToFileAttribs.get(getPath(f)).size;
+  }
+  
+  /**
+   * Directory operations are not supported
+   */
+  public Path[] listPathsRaw(Path f) throws IOException {
+    return null;
+  }
+  public void setWorkingDirectory(Path new_dir) {}
+  public Path getWorkingDirectory() {
+    return staticWorkingDir;
+  }
+  public boolean mkdirs(Path f) throws IOException {
+    return false;
+  }
+  
+  /** lock operations are not supported */
+  public void lock(Path f, boolean shared) throws IOException {}
+  public void release(Path f) throws IOException {}
+  
+  /** copy/move operations are not supported */
+  public void copyFromLocalFile(Path src, Path dst) throws IOException {}
+  public void moveFromLocalFile(Path src, Path dst) throws IOException {}
+  public void copyToLocalFile(Path src, Path dst, boolean copyCrc)
+  throws IOException {}
+
+  public Path startLocalOutput(Path fsOutputFile, Path tmpLocalFile)
+      throws IOException {
+    return fsOutputFile;
+  }
+
+  public void completeLocalOutput(Path fsOutputFile, Path tmpLocalFile)
+      throws IOException {
+  }
+
+  public void reportChecksumFailure(Path p, FSInputStream in,
+      long inPos,
+      FSInputStream sums, long sumsPos) {
+  }
+
+  public long getBlockSize(Path f) throws IOException {
+    return getDefaultBlockSize();
+  }
+
+  public long getDefaultBlockSize() {
+    return 32 * 1024; //some random large number. can be anything actually
+  }
+
+  public short getDefaultReplication() {
+    return 1;
+  }
+  
+  /** Some APIs exclusively for InMemoryFileSystem */
+  
+  /** Register a path with its size. This will also register a checksum for 
+   * the file that the user is trying to create. This is required since none
+   * of the FileSystem APIs accept the size of the file as argument. But since
+   * it is required for us to apriori know the size of the file we are going to
+   * create, the user must call this method for each file he wants to create
+   * and reserve memory for that file. We either succeed in reserving memory
+   * for both the main file and the checksum file and return true, or return 
+   * false.
+   */
+  public boolean reserveSpaceWithCheckSum(Path f, int size) {
+    //get the size of the checksum file (we know it is going to be 'int'
+    //since this is an inmem fs with file sizes that will fit in 4 bytes)
+    int checksumSize = getChecksumFileLength(size);
+    synchronized (this) {
+      if (!canFitInMemory(size + checksumSize)) return false;
+      FileAttributes fileAttr;
+      FileAttributes checksumAttr;
+      try {
+        fileAttr = new FileAttributes(size);
+        checksumAttr = new FileAttributes(checksumSize);
+      } catch (OutOfMemoryError o) {
+        return false;
+      }
+      totalUsed += size + checksumSize;
+      tempFileAttribs.put(getPath(f), fileAttr);
+      tempFileAttribs.put(getPath(FileSystem.getChecksumFile(f)),checksumAttr); 
+      return true;
+    }
+  }
+  
+  public int getChecksumFileLength(int size) {
+    return (int)super.getChecksumFileLength(size, bytesPerSum);
+  }
+  
+  /** This API getClosedFiles could have been implemented over listPathsRaw
+   * but it is an overhead to maintain directory structures for this impl of
+   * the in-memory fs.
+   */
+  public Path[] getFiles(PathFilter filter) {
+    synchronized (this) {
+      List <String> closedFilesList = new ArrayList();
+      Set paths = pathToFileAttribs.keySet();
+      if (paths == null || paths.isEmpty()) return new Path[0];
+      Iterator iter = paths.iterator();
+      while (iter.hasNext()) {
+        String f = (String)iter.next();
+        if (filter.accept(new Path(f)))
+          closedFilesList.add(f);
+      }
+      String [] names = 
+        closedFilesList.toArray(new String[closedFilesList.size()]);
+      Path [] results = new Path[names.length];
+      for (int i = 0; i < names.length; i++) {
+        results[i] = new Path(names[i]);
+      }
+      return results;
+    }
+  }
+  
+  public int getFSSize() {
+    return fsSize;
+  }
+  
+  public float getPercentUsed() {
+    return (float)totalUsed/fsSize;
+  }
+ 
+  private boolean canFitInMemory(int size) {
+    if (size + totalUsed < fsSize)
+      return true;
+    return false;
+  }
+  
+  private String getPath(Path f) {
+    return f.toUri().getPath();
+  }
+  
+  private static class FileAttributes {
+    private byte[] data;
+    private int size;
+    
+    public FileAttributes(int size) {
+      this.size = size;
+      this.data = new byte[size];
+    }
+  }
+}

src/java/org/apache/hadoop/io/SequenceFile.java

@@ -21,6 +21,8 @@ package org.apache.hadoop.io;
 import java.io.*;
 import java.util.*;
 import java.net.InetAddress;
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.rmi.server.UID;
 import java.security.MessageDigest;
 import org.apache.lucene.util.PriorityQueue;
@@ -1893,7 +1895,8 @@ public class SequenceFile {
     }
 
     /**
-     * Merges the contents of files passed in Path[]
+     * Merges the contents of files passed in Path[] using a max factor value
+     * that is already set
      * @param inNames the array of path names
      * @param deleteInputs true if the input files should be deleted when 
      * unnecessary
@@ -1901,6 +1904,22 @@ public class SequenceFile {
      * @throws IOException
      */
     public RawKeyValueIterator merge(Path [] inNames, boolean deleteInputs) 
+    throws IOException {
+      return merge(inNames, deleteInputs, 
+                  (inNames.length < factor) ? inNames.length : factor);
+    }
+
+    /**
+     * Merges the contents of files passed in Path[]
+     * @param inNames the array of path names
+     * @param deleteInputs true if the input files should be deleted when 
+     * unnecessary
+     * @param factor the factor that will be used as the maximum merge fan-in
+     * @return RawKeyValueIteratorMergeQueue
+     * @throws IOException
+     */
+    public RawKeyValueIterator merge(Path [] inNames, boolean deleteInputs,
+                                     int factor) 
     throws IOException {
       //get the segments from inNames
       ArrayList <SegmentDescriptor> a = new ArrayList <SegmentDescriptor>();
@@ -1911,7 +1930,7 @@ public class SequenceFile {
         s.doSync();
         a.add(s);
       }
-      factor = (inNames.length < factor) ? inNames.length : factor;
+      this.factor = factor;
       MergeQueue mQueue = new MergeQueue(a);
       return mQueue.merge();
     }
@@ -1948,26 +1967,48 @@ public class SequenceFile {
     /**
      * Clones the attributes (like compression of the input file and creates a 
      * corresponding Writer
-     * @param fileSys the FileSystem object
+     * @param ignoredFileSys the (ignored) FileSystem object
      * @param inputFile the path of the input file whose attributes should be 
      * cloned 
      * @param outputFile the path of the output file 
      * @param prog the Progressable to report status during the file write
      * @return Writer
      * @throws IOException
+     * @deprecated call  #cloneFileAttributes(Path,Path,Progressable) instead
      */
-    public Writer cloneFileAttributes(FileSystem fileSys, Path inputFile, 
-                  Path outputFile, Progressable prog) throws IOException {
-      Reader reader = new Reader(fileSys, inputFile, 4096, conf);
+    public Writer cloneFileAttributes(FileSystem ignoredFileSys,
+                  Path inputFile, Path outputFile, Progressable prog) 
+    throws IOException {
+      return cloneFileAttributes(inputFile, outputFile, prog);
+    }
+
+    /**
+     * Clones the attributes (like compression of the input file and creates a 
+     * corresponding Writer
+     * @param inputFile the path of the input file whose attributes should be 
+     * cloned
+     * @param outputFile the path of the output file 
+     * @param prog the Progressable to report status during the file write
+     * @return Writer
+     * @throws IOException
+     */
+    public Writer cloneFileAttributes(Path inputFile, Path outputFile, 
+                  Progressable prog) throws IOException {
+      FileSystem srcFileSys = inputFile.getFileSystem(conf);
+      Reader reader = new Reader(srcFileSys, inputFile, 4096, conf);
       boolean compress = reader.isCompressed();
       boolean blockCompress = reader.isBlockCompressed();
       CompressionCodec codec = reader.getCompressionCodec();
       reader.close();
+      
+      FileSystem dstFileSys = outputFile.getFileSystem(conf);
       FSDataOutputStream out;
       if (prog != null)
-        out = fs.create(outputFile, true, memory/(factor+1), prog);
+        out = dstFileSys.create(outputFile, true, 
+            conf.getInt("io.file.buffer.size", 4096), prog);
       else
-        out = fs.create(outputFile, true, memory/(factor+1));
+        out = dstFileSys.create(outputFile, true, 
+            conf.getInt("io.file.buffer.size", 4096));
       Writer writer = createWriter(conf, out, keyClass, valClass, compress, 
                           blockCompress, codec);
       return writer;
@@ -2158,7 +2199,8 @@ public class SequenceFile {
             Path outputFile = conf.getLocalPath("mapred.local.dir", 
                                   (outFile.suffix("." + passNo)).toString());
             Writer writer = cloneFileAttributes(fs, 
-                            mStream[0].segmentPathName, outputFile, null);
+                            fs.makeQualified(mStream[0].segmentPathName), 
+                            fs.makeQualified(outputFile), null);
             writer.sync = null; //disable sync for temp files
             writeFile(this, writer);
             writer.close();
@@ -2276,7 +2318,7 @@ public class SequenceFile {
       public boolean nextRawKey() throws IOException {
         if (in == null) {
         Reader reader = new Reader(fs, segmentPathName, 
-                memory/(factor+1), segmentOffset, 
+            conf.getInt("io.file.buffer.size", 4096), segmentOffset, 
                 segmentLength, conf);
         
         //sometimes we ignore syncs especially for temp merge files

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

@@ -29,6 +29,19 @@ interface MRConstants {
     public static final long HEARTBEAT_INTERVAL = 10 * 1000;
     public static final long TASKTRACKER_EXPIRY_INTERVAL = 10 * 60 * 1000;
 
+    //for the inmemory filesystem (to do in-memory merge)
+    /**
+     * Constant denoting when a merge of in memory files will be triggered 
+     */
+    public static final float MAX_INMEM_FILESYS_USE = 0.5f;
+    /**
+     * Constant denoting the max size (in terms of the fraction of the total 
+     * size of the filesys) of a map output file that we will try
+     * to keep in mem. Ideally, this should be a factor of MAX_INMEM_FILESYS_USE
+     */
+    public static final float MAX_INMEM_FILESIZE_FRACTION =
+      MAX_INMEM_FILESYS_USE/2;
+    
     //
     // Result codes
     //

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

@@ -22,6 +22,8 @@ import java.io.IOException;
 
 import java.io.*;
 import java.net.*;
+
+import org.apache.hadoop.fs.InMemoryFileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.io.*;
@@ -29,7 +31,7 @@ import org.apache.hadoop.util.Progressable;
 
 /** The location of a map output file, as passed to a reduce task via the
  * {@link InterTrackerProtocol}. */ 
-class MapOutputLocation implements Writable {
+class MapOutputLocation implements Writable, MRConstants {
 
     static {                                      // register a ctor
       WritableFactories.setFactory
@@ -162,4 +164,96 @@ class MapOutputLocation implements Writable {
     }
     return totalBytes;
   }
+  
+  /**
+   * Get the map output into a local file (either in the inmemory fs or on the 
+   * local fs) from the remote server.
+   * We use the file system so that we generate checksum files on the data.
+   * @param inMemFileSys the inmemory filesystem to write the file to
+   * @param localFileSys the local filesystem to write the file to
+   * @param localFilename the filename to write the data into
+   * @param reduce the reduce id to get for
+   * @param timeout number of ms for connection and read timeout
+   * @return the path of the file that got created
+   * @throws IOException when something goes wrong
+   */
+  public Path getFile(InMemoryFileSystem inMemFileSys,
+                      FileSystem localFileSys,
+                      Path localFilename, 
+                      int reduce,
+                      int timeout) throws IOException, InterruptedException {
+    boolean good = false;
+    long totalBytes = 0;
+    FileSystem fileSys = localFileSys;
+    Thread currentThread = Thread.currentThread();
+    URL path = new URL(toString() + "&reduce=" + reduce);
+    try {
+      URLConnection connection = path.openConnection();
+      if (timeout > 0) {
+        connection.setConnectTimeout(timeout);
+        connection.setReadTimeout(timeout);
+      }
+      InputStream input = connection.getInputStream();
+      OutputStream output = null;
+      
+      //We will put a file in memory if it meets certain criteria:
+      //1. The size of the file should be less than 25% of the total inmem fs
+      //2. There is space available in the inmem fs
+      
+      int length = connection.getContentLength();
+      int inMemFSSize = inMemFileSys.getFSSize();
+      int checksumLength = inMemFileSys.getChecksumFileLength(length);
+        
+      boolean createInMem = 
+        (((float)(length + checksumLength) / inMemFSSize <= 
+        MAX_INMEM_FILESIZE_FRACTION) && 
+        inMemFileSys.reserveSpaceWithCheckSum(localFilename, length));
+      
+      if (createInMem)
+        fileSys = inMemFileSys;
+      else
+        fileSys = localFileSys;
+
+      output = fileSys.create(localFilename);
+      try {  
+        try {
+          byte[] buffer = new byte[64 * 1024];
+          if (currentThread.isInterrupted()) {
+            throw new InterruptedException();
+          }
+          int len = input.read(buffer);
+          while (len > 0) {
+            totalBytes += len;
+            output.write(buffer, 0 ,len);
+            if (currentThread.isInterrupted()) {
+              throw new InterruptedException();
+            }
+            len = input.read(buffer);
+          }
+        } finally {
+          output.close();
+        }
+      } finally {
+        input.close();
+      }
+      good = ((int) totalBytes) == connection.getContentLength();
+      if (!good) {
+        throw new IOException("Incomplete map output received for " + path +
+                              " (" + totalBytes + " instead of " + 
+                              connection.getContentLength() + ")");
+      }
+    } finally {
+      if (!good) {
+        try {
+          fileSys.delete(localFilename);
+          totalBytes = 0;
+        } catch (Throwable th) {
+          // IGNORED because we are cleaning up
+        }
+        return null;
+      }
+    }
+    return fileSys.makeQualified(localFilename);
+  }
+
 }

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

@@ -233,11 +233,18 @@ class ReduceTask extends Task {
     
 
     // open a file to collect map output
-    Path[] mapFiles = new Path[numMaps];
+    // since we don't know how many map outputs got merged in memory, we have
+    // to check whether a given map output exists, and if it does, add it in
+    // the list of files to merge, otherwise not.
+    List <Path> mapFilesList = new ArrayList();
     for(int i=0; i < numMaps; i++) {
-      mapFiles[i] = mapOutputFile.getInputFile(i, getTaskId());
+      Path f = mapOutputFile.getInputFile(i, getTaskId());
+      if (lfs.exists(f))
+        mapFilesList.add(f);
     }
-
+    Path[] mapFiles = new Path[mapFilesList.size()];
+    mapFiles = mapFilesList.toArray(mapFiles);
+    
     // spawn a thread to give sort progress heartbeats
     Thread sortProgress = new Thread() {
         public void run() {

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

@@ -18,16 +18,22 @@
 package org.apache.hadoop.mapred;
 
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.InMemoryFileSystem;
+import org.apache.hadoop.fs.LocalFileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.SequenceFile.Sorter.RawKeyValueIterator;
 import org.apache.hadoop.util.*;
 
 import java.io.*;
 import java.util.*;
+import java.net.*;
 import java.text.DecimalFormat;
 import org.apache.hadoop.util.Progressable;
 
 /** Runs a reduce task. */
-class ReduceTaskRunner extends TaskRunner {
+class ReduceTaskRunner extends TaskRunner implements MRConstants {
   /** Number of ms before timing out a copy */
   private static final int STALLED_COPY_TIMEOUT = 3 * 60 * 1000;
   
@@ -79,11 +85,31 @@ class ReduceTaskRunner extends TaskRunner {
    */
   private long lastPollTime;
   
+  /**
+   * A reference to the in memory file system for writing the map outputs to.
+   */
+  private InMemoryFileSystem inMemFileSys;
+  
   /**
    * A reference to the local file system for writing the map outputs to.
    */
   private FileSystem localFileSys;
 
+  /**
+   * An instance of the sorter used for doing merge
+   */
+  private SequenceFile.Sorter sorter;
+  
+  /**
+   * A reference to the throwable object (if merge throws an exception)
+   */
+  private volatile Throwable mergeThrowable;
+  
+  /** 
+   * A flag to indicate that merge is in progress
+   */
+  private volatile boolean mergeInProgress = false;
+
   /**
    * The threads for fetching the files.
    */
@@ -120,9 +146,28 @@ class ReduceTaskRunner extends TaskRunner {
     public MapOutputLocation getLocation() { return loc; }
   }
 
-  private class PingTimer implements Progressable {
+  private class PingTimer extends Thread implements Progressable {
     Task task = getTask();
     TaskTracker tracker = getTracker();
+
+    public void run() {
+      LOG.info(task.getTaskId() + " Started thread: " + getName());
+      while (true) {
+        try {
+          progress();
+          Thread.sleep(Task.PROGRESS_INTERVAL);
+        }
+        catch (InterruptedException i) {
+          return;
+        }
+        catch (Throwable e) {
+          LOG.info(task.getTaskId() + " Thread Exception in " +
+                   "reporting sort progress\n" +
+                   StringUtils.stringifyException(e));
+          continue;
+        }
+      }
+    }
     
     public void progress() {
       task.reportProgress(tracker);
@@ -134,7 +179,6 @@ class ReduceTaskRunner extends TaskRunner {
   /** Copies map outputs as they become available */
   private class MapOutputCopier extends Thread {
 
-    private PingTimer pingTimer = new PingTimer();
     private MapOutputLocation currentLocation = null;
     private int id = nextMapOutputCopierId++;
     
@@ -196,7 +240,7 @@ class ReduceTaskRunner extends TaskRunner {
 
           try {
             start(loc);
-            size = copyOutput(loc, pingTimer);
+            size = copyOutput(loc);
           } catch (IOException e) {
             LOG.warn(reduceTask.getTaskId() + " copy failed: " +
                         loc.getMapTaskId() + " from " + loc.getHost());
@@ -215,13 +259,11 @@ class ReduceTaskRunner extends TaskRunner {
 
     /** Copies a a map output from a remote host, using raw RPC. 
      * @param currentLocation the map output location to be copied
-     * @param pingee a status object to ping as we make progress
-     * @return the size of the copied file
+     * @return the path (fully qualified) of the copied file
      * @throws IOException if there is an error copying the file
      * @throws InterruptedException if the copier should give up
      */
-    private long copyOutput(MapOutputLocation loc, 
-                            Progressable pingee
+    private long copyOutput(MapOutputLocation loc
                             ) throws IOException, InterruptedException {
 
       String reduceId = reduceTask.getTaskId();
@@ -233,21 +275,42 @@ class ReduceTaskRunner extends TaskRunner {
       // a working filename that will be unique to this attempt
       Path tmpFilename = new Path(finalFilename + "-" + id);
       // this copies the map output file
-      long bytes = loc.getFile(localFileSys, tmpFilename,
-                               reduceTask.getPartition(), pingee,
+      tmpFilename = loc.getFile(inMemFileSys, localFileSys, tmpFilename,
+                               reduceTask.getPartition(),
                                STALLED_COPY_TIMEOUT);
+      if (tmpFilename == null)
+        throw new IOException("File " + finalFilename + "-" + id + 
+                              " not created");
+      long bytes = -1;
       // lock the ReduceTaskRunner while we do the rename
       synchronized (ReduceTaskRunner.this) {
-        // if we can't rename the file, something is broken
-        if (!(new File(tmpFilename.toString()).
-                 renameTo(new File(finalFilename.toString())))) {
-          localFileSys.delete(tmpFilename);
+        // if we can't rename the file, something is broken (and IOException
+        // will be thrown). This file could have been created in the inmemory
+        // fs or the localfs. So need to get the filesystem owning the path. 
+        FileSystem fs = tmpFilename.getFileSystem(conf);
+        if (!fs.rename(tmpFilename, finalFilename)) {
+          fs.delete(tmpFilename);
           throw new IOException("failure to rename map output " + tmpFilename);
         }
+        bytes = fs.getLength(finalFilename);
+        LOG.info(reduceId + " done copying " + loc.getMapTaskId() +
+                 " output from " + loc.getHost() + ".");
+        //Create a thread to do merges. Synchronize access/update to 
+        //mergeInProgress
+        if (!mergeInProgress && inMemFileSys.getPercentUsed() >= 
+                                                       MAX_INMEM_FILESYS_USE) {
+          LOG.info(reduceId + " InMemoryFileSystem " + 
+                   inMemFileSys.getUri().toString() +
+                   " is " + inMemFileSys.getPercentUsed() + 
+                   " full. Triggering merge");
+          InMemFSMergeThread m = new InMemFSMergeThread(inMemFileSys,
+                                     (LocalFileSystem)localFileSys, sorter);
+          m.setName("Thread for merging in memory files");
+          m.setDaemon(true);
+          mergeInProgress = true;
+          m.start();
+        }
       }
-      LOG.info(reduceTask.getTaskId() + " done copying " + loc.getMapTaskId() +
-               " output from " + loc.getHost() + ".");
-      
       return bytes;
     }
 
@@ -258,7 +321,6 @@ class ReduceTaskRunner extends TaskRunner {
     super(task, tracker, conf);
     this.mapOutputFile = new MapOutputFile();
     this.mapOutputFile.setConf(conf);
-    localFileSys = FileSystem.getLocal(conf);
 
     this.reduceTask = (ReduceTask)getTask();
     this.scheduledCopies = new ArrayList(100);
@@ -266,6 +328,18 @@ class ReduceTaskRunner extends TaskRunner {
     this.numCopiers = conf.getInt("mapred.reduce.parallel.copies", 5);
     this.maxBackoff = conf.getInt("mapred.reduce.copy.backoff", 300);
 
+    //we want to distinguish inmem fs instances for different reduces. Hence,
+    //append a unique string in the uri for the inmem fs name
+    URI uri = URI.create("ramfs://mapoutput" + reduceTask.hashCode());
+    inMemFileSys = (InMemoryFileSystem)FileSystem.get(uri, conf);
+    LOG.info(reduceTask.getTaskId() + " Created an InMemoryFileSystem, uri: " +
+             uri);
+    localFileSys = FileSystem.getLocal(conf);
+    //create an instance of the sorter
+    sorter =
+      new SequenceFile.Sorter(inMemFileSys, conf.getOutputKeyComparator(), 
+          conf.getMapOutputValueClass(), conf);
+    
     // hosts -> next contact time
     this.penaltyBox = new Hashtable();
     
@@ -311,9 +385,14 @@ class ReduceTaskRunner extends TaskRunner {
     // start the clock for bandwidth measurement
     long startTime = System.currentTimeMillis();
     long currentTime = startTime;
-    
+    PingTimer pingTimer = new PingTimer();
+    pingTimer.setName("Map output copy reporter for task " + 
+                      reduceTask.getTaskId());
+    pingTimer.setDaemon(true);
+    pingTimer.start();
+    try {
     // loop until we get all required outputs or are killed
-    while (!killed && numCopied < numOutputs) {
+    while (!killed && numCopied < numOutputs && mergeThrowable == null) {
 
       LOG.info(reduceTask.getTaskId() + " Need " + (numOutputs-numCopied) +
                " map output(s)");
@@ -382,12 +461,11 @@ class ReduceTaskRunner extends TaskRunner {
       // new, just wait for a bit
       try {
         if (numInFlight == 0 && numScheduled == 0) {
-          getTask().reportProgress(getTracker());
           Thread.sleep(5000);
         }
       } catch (InterruptedException e) { } // IGNORE
 
-      while (!killed && numInFlight > 0) {
+      while (!killed && numInFlight > 0 && mergeThrowable == null) {
         LOG.debug(reduceTask.getTaskId() + " numInFlight = " + numInFlight);
         CopyResult cr = getCopyResult();
         
@@ -404,7 +482,6 @@ class ReduceTaskRunner extends TaskRunner {
             copyPhase.setStatus("copy (" + numCopied + " of " + numOutputs + 
                                 " at " +
                                 mbpsFormat.format(transferRate) +  " MB/s)");          
-            getTask().reportProgress(getTracker());
           } else {
             // this copy failed, put it back onto neededOutputs
             neededOutputs.add(new Integer(cr.getMapId()));
@@ -454,7 +531,64 @@ class ReduceTaskRunner extends TaskRunner {
       }
     }
     
-    return numCopied == numOutputs && !killed;
+    if (mergeThrowable != null) {
+      //set the task state to FAILED
+      TaskTracker tracker = ReduceTaskRunner.this.getTracker();
+      TaskTracker.TaskInProgress tip = 
+        tracker.runningTasks.get(reduceTask.getTaskId());
+      tip.runstate = TaskStatus.State.FAILED;
+      try {
+        tip.cleanup();
+      } catch (Throwable ie2) {
+        // Ignore it, we are just trying to cleanup.
+      }
+      inMemFileSys.close();
+    }
+    
+    //Do a merge of in-memory files (if there are any)
+    if (!killed && mergeThrowable == null) {
+      try {
+        //wait for an ongoing merge (if it is in flight) to complete
+        while (mergeInProgress) {
+          Thread.sleep(200);
+        }
+        LOG.info(reduceTask.getTaskId() + 
+                 " Copying of all map outputs complete. " + 
+                 "Initiating the last merge on the remaining files in " + 
+                 inMemFileSys.getUri());
+        //initiate merge
+        Path[] inMemClosedFiles = inMemFileSys.getFiles(MAP_OUTPUT_FILTER);
+        if (inMemClosedFiles.length == 0) {
+          LOG.info(reduceTask.getTaskId() + "Nothing to merge from " + 
+              inMemFileSys.getUri());
+          return numCopied == numOutputs;
+        }
+        RawKeyValueIterator rIter = sorter.merge(inMemClosedFiles, true, 
+                                                 inMemClosedFiles.length);
+        //name this output file same as the name of the first file that is 
+        //there in the current list of inmem files (this is guaranteed to be
+        //absent on the disk currently. So we don't overwrite a prev. 
+        //created spill)
+        SequenceFile.Writer writer = sorter.cloneFileAttributes(
+            inMemFileSys.makeQualified(inMemClosedFiles[0]), 
+            localFileSys.makeQualified(inMemClosedFiles[0]), null);
+        sorter.writeFile(rIter, writer);
+        writer.close();
+        LOG.info(reduceTask.getTaskId() +
+                 " Merge of the " +inMemClosedFiles.length +
+                 " files in InMemoryFileSystem complete." +
+                 " Local file is " + inMemClosedFiles[0]);
+      } catch (Throwable t) {
+        LOG.warn("Merge of the inmemory files threw an exception: " + 
+            StringUtils.stringifyException(t));
+        inMemFileSys.close();
+        return false;
+      }
+    }
+    return mergeThrowable == null && numCopied == numOutputs && !killed;
+    } finally {
+      pingTimer.interrupt();
+    }
   }
   
   
@@ -529,4 +663,60 @@ class ReduceTaskRunner extends TaskRunner {
     }
   }
 
+  private class InMemFSMergeThread extends Thread {
+    private InMemoryFileSystem inMemFileSys;
+    private LocalFileSystem localFileSys;
+    private SequenceFile.Sorter sorter;
+    
+    public InMemFSMergeThread(InMemoryFileSystem inMemFileSys, 
+        LocalFileSystem localFileSys, SequenceFile.Sorter sorter) {
+      this.inMemFileSys = inMemFileSys;
+      this.localFileSys = localFileSys;
+      this.sorter = sorter;
+    }
+    public void run() {
+      LOG.info(reduceTask.getTaskId() + " Thread started: " + getName());
+      try {
+        //initiate merge
+        Path[] inMemClosedFiles = inMemFileSys.getFiles(MAP_OUTPUT_FILTER);
+        //Note that the above Path[] could be of length 0 if all copies are 
+        //in flight. So we make sure that we have some 'closed' map
+        //output files to merge to get the benefit of in-memory merge
+        if (inMemClosedFiles.length >= 
+          (int)(MAX_INMEM_FILESYS_USE/MAX_INMEM_FILESIZE_FRACTION)) {
+          RawKeyValueIterator rIter = sorter.merge(inMemClosedFiles, true, 
+              inMemClosedFiles.length);
+          //name this output file same as the name of the first file that is 
+          //there in the current list of inmem files (this is guaranteed to be
+          //absent on the disk currently. So we don't overwrite a prev. 
+          //created spill)
+          SequenceFile.Writer writer = sorter.cloneFileAttributes(
+              inMemFileSys.makeQualified(inMemClosedFiles[0]), 
+              localFileSys.makeQualified(inMemClosedFiles[0]), null);
+          sorter.writeFile(rIter, writer);
+          writer.close();
+          LOG.info(reduceTask.getTaskId() + 
+                   " Merge of the " +inMemClosedFiles.length +
+                   " files in InMemoryFileSystem complete." +
+                   " Local file is " + inMemClosedFiles[0]);
+        }
+        else {
+          LOG.info(reduceTask.getTaskId() + " Nothing to merge from " + 
+              inMemFileSys.getUri());
+        }
+      } catch (Throwable t) {
+        LOG.warn("Merge of the inmemory files threw an exception: " + 
+            StringUtils.stringifyException(t));
+        ReduceTaskRunner.this.mergeThrowable = t;
+      }
+      finally {
+        mergeInProgress = false;
+      }
+    }
+  }
+  final private static PathFilter MAP_OUTPUT_FILTER = new PathFilter() {
+    public boolean accept(Path file) {
+      return file.toString().endsWith(".out");
+    }     
+  };
 }

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

@@ -885,7 +885,7 @@ public class TaskTracker
     class TaskInProgress {
         Task task;
         float progress;
-        TaskStatus.State runstate;
+        volatile TaskStatus.State runstate;
         long lastProgressReport;
         StringBuffer diagnosticInfo = new StringBuffer();
         TaskRunner runner;