HADOOP-5585. Clear FileSystem statistics between tasks when jvm-reuse

is enabled. (omalley)


git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/branches/branch-0.20@763110 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -857,6 +857,9 @@ Release 0.20.0 - Unreleased
     HADOOP-5068. Fix NPE in TestCapacityScheduler.  (Vinod Kumar Vavilapalli
     via szetszwo)
 
+    HADOOP-5585. Clear FileSystem statistics between tasks when jvm-reuse
+    is enabled. (omalley)
+
 Release 0.19.2 - Unreleased
 
   BUG FIXES

src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/ClusterWithCapacityScheduler.java

@@ -31,6 +31,7 @@ import junit.framework.TestCase;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.LocalFileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
@@ -117,11 +118,7 @@ public class ClusterWithCapacityScheduler extends TestCase {
       throws IOException {
     Configuration config = new Configuration(false);
 
-    LocalFileSystem fs = new LocalFileSystem();
-    fs.setConf(config);
-    // The above call doesn't set the configuration for the underlying
-    // RawFileSystem. Explicitly doing it.
-    fs.getRawFileSystem().setConf(config);
+    LocalFileSystem fs = FileSystem.getLocal(config);
 
     String myResourcePath = System.getProperty("test.build.data");
     Path schedulerConfigFilePath =
@@ -146,9 +143,7 @@ public class ClusterWithCapacityScheduler extends TestCase {
   private void cleanUpSchedulerConfigFile() throws IOException {
     Configuration config = new Configuration(false);
 
-    LocalFileSystem fs = new LocalFileSystem();
-    fs.setConf(config);
-    fs.getRawFileSystem().setConf(config);
+    LocalFileSystem fs = FileSystem.getLocal(config);
 
     String myResourcePath = System.getProperty("test.build.data");
     Path schedulerConfigFilePath =

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

@@ -17,9 +17,20 @@
  */
 package org.apache.hadoop.fs;
 
-import java.io.*;
-import java.net.*;
-import java.util.*;
+import java.io.Closeable;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.IdentityHashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.regex.Pattern;
 
@@ -68,10 +79,6 @@ public abstract class FileSystem extends Configured implements Closeable {
     statisticsTable =
       new IdentityHashMap<Class<? extends FileSystem>, Statistics>();
   
-  /** Recording statistics per FileSystem URI scheme */
-  private static final Map<String, Statistics> statsByUriScheme = 
-    new HashMap<String, Statistics>();
-
   /**
    * The statistics for this file system.
    */
@@ -117,8 +124,9 @@ public abstract class FileSystem extends Configured implements Closeable {
    *   for this FileSystem
    * @param conf the configuration
    */
-  public abstract void initialize(URI name, Configuration conf)
-    throws IOException;
+  public void initialize(URI name, Configuration conf) throws IOException {
+    statistics = getStatistics(name.getScheme(), getClass());    
+  }
 
   /** Returns a URI whose scheme and authority identify this FileSystem.*/
   public abstract URI getUri();
@@ -260,7 +268,6 @@ public abstract class FileSystem extends Configured implements Closeable {
 
   protected FileSystem() {
     super(null);
-    statistics = getStatistics(this.getClass());
   }
 
   /** Check that a Path belongs to this FileSystem. */
@@ -1364,7 +1371,6 @@ public abstract class FileSystem extends Configured implements Closeable {
     }
     FileSystem fs = (FileSystem)ReflectionUtils.newInstance(clazz, conf);
     fs.initialize(uri, conf);
-    statsByUriScheme.put(uri.getScheme(), fs.statistics);
     return fs;
   }
 
@@ -1474,9 +1480,14 @@ public abstract class FileSystem extends Configured implements Closeable {
   }
   
   public static final class Statistics {
+    private final String scheme;
     private AtomicLong bytesRead = new AtomicLong();
     private AtomicLong bytesWritten = new AtomicLong();
     
+    public Statistics(String scheme) {
+      this.scheme = scheme;
+    }
+
     /**
      * Increment the bytes read in the statistics
      * @param newBytes the additional bytes read
@@ -1513,32 +1524,65 @@ public abstract class FileSystem extends Configured implements Closeable {
       return bytesRead + " bytes read and " + bytesWritten + 
              " bytes written";
     }
+    
+    /**
+     * Reset the counts of bytes to 0.
+     */
+    public void reset() {
+      bytesWritten.set(0);
+      bytesRead.set(0);
+    }
+    
+    /**
+     * Get the uri scheme associated with this statistics object.
+     * @return the schema associated with this set of statistics
+     */
+    public String getScheme() {
+      return scheme;
+    }
   }
   
   /**
    * Get the Map of Statistics object indexed by URI Scheme.
    * @return a Map having a key as URI scheme and value as Statistics object
+   * @deprecated use {@link #getAllStatistics} instead
    */
   public static synchronized Map<String, Statistics> getStatistics() {
-    return statsByUriScheme;
+    Map<String, Statistics> result = new HashMap<String, Statistics>();
+    for(Statistics stat: statisticsTable.values()) {
+      result.put(stat.getScheme(), stat);
+    }
+    return result;
+  }
+
+  /**
+   * Return the FileSystem classes that have Statistics
+   */
+  public static synchronized List<Statistics> getAllStatistics() {
+    return new ArrayList<Statistics>(statisticsTable.values());
   }
   
   /**
    * Get the statistics for a particular file system
-   * @deprecated Consider using {@link #getStatistics()} instead.
    * @param cls the class to lookup
    * @return a statistics object
    */
   public static synchronized 
-  Statistics getStatistics(Class<? extends FileSystem> cls) {
+  Statistics getStatistics(String scheme, Class<? extends FileSystem> cls) {
     Statistics result = statisticsTable.get(cls);
     if (result == null) {
-      result = new Statistics();
+      result = new Statistics(scheme);
       statisticsTable.put(cls, result);
     }
     return result;
   }
   
+  public static synchronized void clearStatistics() {
+    for(Statistics stat: statisticsTable.values()) {
+      stat.reset();
+    }
+  }
+
   public static synchronized
   void printStatistics() throws IOException {
     for (Map.Entry<Class<? extends FileSystem>, Statistics> pair: 

src/core/org/apache/hadoop/fs/RawLocalFileSystem.java

@@ -53,7 +53,8 @@ public class RawLocalFileSystem extends FileSystem {
 
   public URI getUri() { return NAME; }
   
-  public void initialize(URI uri, Configuration conf) {
+  public void initialize(URI uri, Configuration conf) throws IOException {
+    super.initialize(uri, conf);
     setConf(conf);
   }
   

src/core/org/apache/hadoop/fs/ftp/FTPFileSystem.java

@@ -56,6 +56,7 @@ public class FTPFileSystem extends FileSystem {
 
   @Override
   public void initialize(URI uri, Configuration conf) throws IOException { // get
+    super.initialize(uri, conf);
     // get host information from uri (overrides info in conf)
     String host = uri.getHost();
     host = (host == null) ? conf.get("fs.ftp.host", null) : host;

src/core/org/apache/hadoop/fs/kfs/KosmosFileSystem.java

@@ -59,27 +59,29 @@ public class KosmosFileSystem extends FileSystem {
     }
 
     public void initialize(URI uri, Configuration conf) throws IOException {
-        try {
-	    if (kfsImpl == null) {
-                if (uri.getHost() == null) {
-                    kfsImpl = new KFSImpl(conf.get("fs.kfs.metaServerHost", ""),
-                                          conf.getInt("fs.kfs.metaServerPort", -1),
-                                          statistics);
-                } else {
-                    kfsImpl = new KFSImpl(uri.getHost(), uri.getPort(), statistics);
-                }
-	    }
-
-            this.localFs = FileSystem.getLocal(conf);
-            this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority());
-            this.workingDir = new Path("/user", System.getProperty("user.name")).makeQualified(this);
-            setConf(conf);
-            
-        } catch (Exception e) {
-            e.printStackTrace();
-            System.out.println("Unable to initialize KFS");
-            System.exit(-1);
+      super.initialize(uri, conf);
+      try {
+        if (kfsImpl == null) {
+          if (uri.getHost() == null) {
+            kfsImpl = new KFSImpl(conf.get("fs.kfs.metaServerHost", ""),
+                                  conf.getInt("fs.kfs.metaServerPort", -1),
+                                  statistics);
+          } else {
+            kfsImpl = new KFSImpl(uri.getHost(), uri.getPort(), statistics);
+          }
         }
+
+        this.localFs = FileSystem.getLocal(conf);
+        this.uri = URI.create(uri.getScheme() + "://" + uri.getAuthority());
+        this.workingDir = new Path("/user", System.getProperty("user.name")
+                                   ).makeQualified(this);
+        setConf(conf);
+
+      } catch (Exception e) {
+        e.printStackTrace();
+        System.out.println("Unable to initialize KFS");
+        System.exit(-1);
+      }
     }
 
     @Deprecated

src/core/org/apache/hadoop/fs/s3/S3FileSystem.java

@@ -70,6 +70,7 @@ public class S3FileSystem extends FileSystem {
 
   @Override
   public void initialize(URI uri, Configuration conf) throws IOException {
+    super.initialize(uri, conf);
     if (store == null) {
       store = createDefaultStore(conf);
     }

src/core/org/apache/hadoop/fs/s3native/NativeS3FileSystem.java

@@ -209,6 +209,7 @@ public class NativeS3FileSystem extends FileSystem {
   
   @Override
   public void initialize(URI uri, Configuration conf) throws IOException {
+    super.initialize(uri, conf);
     if (store == null) {
       store = createDefaultStore(conf);
     }

src/examples/org/apache/hadoop/examples/Sort.java

@@ -51,6 +51,7 @@ import org.apache.hadoop.util.ToolRunner;
  *            <i>in-dir</i> <i>out-dir</i> 
  */
 public class Sort<K,V> extends Configured implements Tool {
+  private RunningJob jobResult = null;
 
   static int printUsage() {
     System.out.println("sort [-m <maps>] [-r <reduces>] " +
@@ -172,7 +173,7 @@ public class Sort<K,V> extends Configured implements Tool {
         " with " + num_reduces + " reduces.");
     Date startTime = new Date();
     System.out.println("Job started: " + startTime);
-    JobClient.runJob(jobConf);
+    jobResult = JobClient.runJob(jobConf);
     Date end_time = new Date();
     System.out.println("Job ended: " + end_time);
     System.out.println("The job took " + 
@@ -187,4 +188,11 @@ public class Sort<K,V> extends Configured implements Tool {
     System.exit(res);
   }
 
+  /**
+   * Get the last job that was run using this instance.
+   * @return the results of the last job that was run
+   */
+  public RunningJob getResult() {
+    return jobResult;
+  }
 }

src/hdfs/org/apache/hadoop/hdfs/DFSClient.java

@@ -81,7 +81,7 @@ public class DFSClient implements FSConstants, java.io.Closeable {
   private int socketTimeout;
   private int datanodeWriteTimeout;
   final int writePacketSize;
-  private FileSystem.Statistics stats;
+  private final FileSystem.Statistics stats;
   private int maxBlockAcquireFailures;
     
  
@@ -146,7 +146,7 @@ public class DFSClient implements FSConstants, java.io.Closeable {
    * Create a new DFSClient connected to the default namenode.
    */
   public DFSClient(Configuration conf) throws IOException {
-    this(NameNode.getAddress(conf), conf);
+    this(NameNode.getAddress(conf), conf, null);
   }
 
   /** 
@@ -189,8 +189,7 @@ public class DFSClient implements FSConstants, java.io.Closeable {
 
   public DFSClient(InetSocketAddress nameNodeAddr, 
                    Configuration conf) throws IOException {
-    this(nameNodeAddr, conf, 
-         FileSystem.getStatistics(DistributedFileSystem.class));
+    this(nameNodeAddr, conf, null);
   }
 
   private void checkOpen() throws IOException {

src/hdfs/org/apache/hadoop/hdfs/DistributedFileSystem.java

@@ -24,7 +24,6 @@ import java.net.*;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.fs.*;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hdfs.protocol.ClientProtocol;
 import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
 import org.apache.hadoop.hdfs.protocol.FSConstants;
 import org.apache.hadoop.hdfs.protocol.Block;
@@ -35,7 +34,7 @@ import org.apache.hadoop.hdfs.server.common.UpgradeStatusReport;
 import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.hdfs.DFSClient.DFSOutputStream;
 import org.apache.hadoop.security.AccessControlException;
-import org.apache.hadoop.util.*;
+import org.apache.hadoop.util.Progressable;
 
 
 /****************************************************************
@@ -71,6 +70,7 @@ public class DistributedFileSystem extends FileSystem {
   public URI getUri() { return uri; }
 
   public void initialize(URI uri, Configuration conf) throws IOException {
+    super.initialize(uri, conf);
     setConf(conf);
 
     String host = uri.getHost();

src/hdfs/org/apache/hadoop/hdfs/HftpFileSystem.java

@@ -81,6 +81,7 @@ public class HftpFileSystem extends FileSystem {
 
   @Override
   public void initialize(URI name, Configuration conf) throws IOException {
+    super.initialize(name, conf);
     setConf(conf);
     try {
       this.ugi = UnixUserGroupInformation.login(conf, true);

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

@@ -130,6 +130,9 @@ class Child {
         task = myTask.getTask();
         taskid = task.getTaskID();
         isCleanup = task.isTaskCleanupTask();
+        // reset the statistics for the task
+        FileSystem.clearStatistics();
+
         //create the index file so that the log files 
         //are viewable immediately
         TaskLog.syncLogs(firstTaskid, taskid, isCleanup);

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

@@ -37,6 +37,7 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.LocalDirAllocator;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FileSystem.Statistics;
 import org.apache.hadoop.io.DataInputBuffer;
 import org.apache.hadoop.io.RawComparator;
 import org.apache.hadoop.io.Text;
@@ -629,15 +630,14 @@ abstract class Task implements Writable, Configurable {
      new HashMap<String, FileSystemStatisticUpdater>();
   
   private synchronized void updateCounters() {
-    for(Map.Entry<String, FileSystem.Statistics> entry : 
-      FileSystem.getStatistics().entrySet()) {
-      String uriScheme = entry.getKey();
+    for(Statistics stat: FileSystem.getAllStatistics()) {
+      String uriScheme = stat.getScheme();
       FileSystemStatisticUpdater updater = statisticUpdaters.get(uriScheme);
       if(updater==null) {//new FileSystem has been found in the cache
-        updater = new FileSystemStatisticUpdater(uriScheme, entry.getValue());
+        updater = new FileSystemStatisticUpdater(uriScheme, stat);
         statisticUpdaters.put(uriScheme, updater);
       }
-      updater.updateCounters();
+      updater.updateCounters();      
     }
   }
 

src/test/org/apache/hadoop/io/TestSequenceFile.java

@@ -390,9 +390,7 @@ public class TestSequenceFile extends TestCase {
 
   public void testClose() throws IOException {
     Configuration conf = new Configuration();
-    LocalFileSystem fs = new LocalFileSystem();
-    fs.setConf(conf);
-    fs.getRawFileSystem().setConf(conf);
+    LocalFileSystem fs = FileSystem.getLocal(conf);
   
     // create a sequence file 1
     Path path1 = new Path(System.getProperty("test.build.data",".")+"/test1.seq");

src/test/org/apache/hadoop/mapred/TestMiniMRDFSSort.java

@@ -94,7 +94,17 @@ public class TestMiniMRDFSSort extends TestCase {
     String[] sortArgs = {sortInput.toString(), sortOutput.toString()};
     
     // Run Sort
-    assertEquals(ToolRunner.run(job, new Sort(), sortArgs), 0);
+    Sort sort = new Sort();
+    assertEquals(ToolRunner.run(job, sort, sortArgs), 0);
+    Counters counters = sort.getResult().getCounters();
+    long mapInput = counters.findCounter(Task.Counter.MAP_INPUT_BYTES
+    ).getValue();
+    long hdfsRead = counters.findCounter(Task.FILESYSTEM_COUNTER_GROUP,
+                                         "HDFS_BYTES_READ").getValue();
+    // the hdfs read should be between 100% and 110% of the map input bytes
+    assertTrue("map input = " + mapInput + ", hdfs read = " + hdfsRead,
+               (hdfsRead < (mapInput * 1.1)) &&
+               (hdfsRead > mapInput));  
   }
   
   private static void runSortValidator(JobConf job, 
@@ -139,8 +149,6 @@ public class TestMiniMRDFSSort extends TestCase {
     job.setNumReduceTasks(0);
     RunningJob result = JobClient.runJob(job);
     long uses = result.getCounters().findCounter("jvm", "use").getValue();
-    System.out.println("maps = " + job.getNumMapTasks());
-    System.out.println(result.getCounters());
     int maps = job.getNumMapTasks();
     if (reuse) {
       assertTrue("maps = " + maps + ", uses = " + uses, maps < uses);