Brownsys pivottrace 210 beta

hadoop-common-project/hadoop-auth/pom.xml

@@ -52,12 +52,17 @@
       <groupId>org.mockito</groupId>
       <artifactId>mockito-all</artifactId>
       <scope>test</scope>
-    </dependency>
+    </dependency>     
     <dependency>
-      <groupId>org.mortbay.jetty</groupId>
-      <artifactId>jetty</artifactId>
-      <scope>test</scope>
-    </dependency>
+       <groupId>org.mortbay.jetty</groupId>
+       <artifactId>jetty-util</artifactId>
+       <scope>test</scope>
+     </dependency>
+     <dependency>
+       <groupId>org.mortbay.jetty</groupId>
+       <artifactId>jetty</artifactId>
+       <scope>test</scope>
+     </dependency>
     <dependency>
       <groupId>javax.servlet</groupId>
       <artifactId>servlet-api</artifactId>

hadoop-common-project/hadoop-common/src/main/bin/hadoop-config.cmd

@@ -289,4 +289,10 @@ if not "%HADOOP_MAPRED_HOME%\%MAPRED_DIR%" == "%HADOOP_YARN_HOME%\%YARN_DIR%" (
   set CLASSPATH=!CLASSPATH!;%HADOOP_MAPRED_HOME%\%MAPRED_DIR%\*
 )
 
+@rem XResourceTracing - add the XBootclasspath jars
+@rem set HADOOP_OPTS=%HADOOP_OPTS% -Xbootclasspath/p:%HADOOP_COMMON_HOME%\%HADOOP_COMMON_LIB_JARS_DIR%\xtrace-resource-tracing-1.0.jar
+@rem set HADOOP_OPTS=%HADOOP_OPTS% -Xbootclasspath/p:%HADOOP_COMMON_HOME%\%HADOOP_COMMON_LIB_JARS_DIR%\xtrace-2.1-20120824.jar
+@rem set HADOOP_OPTS=%HADOOP_OPTS% -Xbootclasspath/p:%HADOOP_COMMON_HOME%\%HADOOP_COMMON_LIB_JARS_DIR%\log4j-1.2.17.jar
+@rem set HADOOP_OPTS=%HADOOP_OPTS% -Xbootclasspath/p:%HADOOP_COMMON_HOME%\%HADOOP_COMMON_LIB_JARS_DIR%\aspectjrt-1.7.3.jar
+
 :eof

hadoop-common-project/hadoop-common/src/main/bin/hadoop-config.sh

@@ -293,3 +293,7 @@ if [ "$HADOOP_CLASSPATH" != "" ]; then
   fi
 fi
 
+#HADOOP_OPTS="$HADOOP_OPTS -Xbootclasspath/p:${HADOOP_COMMON_LIB_JARS_DIR}/xtrace-resource-tracing-1.0.jar"
+#HADOOP_OPTS="$HADOOP_OPTS -Xbootclasspath/p:${HADOOP_COMMON_LIB_JARS_DIR}/xtrace-2.1-20120824.jar"
+#HADOOP_OPTS="$HADOOP_OPTS -Xbootclasspath/p:${HADOOP_COMMON_LIB_JARS_DIR}/log4j-1.2.17.jar"
+#HADOOP_OPTS="$HADOOP_OPTS -Xbootclasspath/p:${HADOOP_COMMON_LIB_JARS_DIR}/aspectjrt-1.7.3.jar"

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSOutputSummer.java

@@ -103,6 +103,7 @@ public synchronized void write(byte b[], int off, int len)
 
     for (int n=0;n<len;n+=write1(b, off+n, len-n)) {
     }
+
   }
 
   /**

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FsShell.java

@@ -34,10 +34,15 @@
 import org.apache.hadoop.util.Tool;
 import org.apache.hadoop.util.ToolRunner;
 
+import com.google.common.base.Joiner;
+
+import edu.brown.cs.systems.xtrace.XTrace;
+
 /** Provide command line access to a FileSystem. */
 @InterfaceAudience.Private
 public class FsShell extends Configured implements Tool {
 
+  static final XTrace.Logger XTRACE = XTrace.getLogger(FsShell.class);
   static final Log LOG = LogFactory.getLog(FsShell.class);
 
   private FileSystem fs;
@@ -247,22 +252,28 @@ public int run(String argv[]) throws Exception {
     } else {
       String cmd = argv[0];
       Command instance = null;
+      XTrace.startTask(true);
+      XTrace.setTenantClass(0);
+      XTRACE.tag("Executing command", Joiner.on(" ").join(argv));
       try {
         instance = commandFactory.getInstance(cmd);
         if (instance == null) {
           throw new UnknownCommandException();
         }
         exitCode = instance.run(Arrays.copyOfRange(argv, 1, argv.length));
+        XTRACE.log("Finished executing command");
       } catch (IllegalArgumentException e) {
         displayError(cmd, e.getLocalizedMessage());
         if (instance != null) {
           printInstanceUsage(System.err, instance);
         }
+        XTRACE.log("Command failed due to illegal argument");
       } catch (Exception e) {
         // instance.run catches IOE, so something is REALLY wrong if here
         LOG.debug("Error", e);
         displayError(cmd, "Fatal internal error");
         e.printStackTrace(System.err);
+        XTRACE.log("Fatal internal error", "Message", e.getMessage());
       }
     }
     return exitCode;

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/RawLocalFileSystem.java

@@ -49,6 +49,7 @@
 @InterfaceAudience.Public
 @InterfaceStability.Stable
 public class RawLocalFileSystem extends FileSystem {
+
   static final URI NAME = URI.create("file:///");
   private Path workingDir;
 
@@ -123,7 +124,7 @@ class LocalFSFileInputStream extends FSInputStream implements HasFileDescriptor
     private long position;
 
     public LocalFSFileInputStream(Path f) throws IOException {
-      this.fis = new TrackingFileInputStream(pathToFile(f));
+      this.fis = new FileInputStream(pathToFile(f));
     }
 
     @Override

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/nativeio/NativeIO.java

@@ -26,6 +26,8 @@
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
@@ -34,8 +36,7 @@
 import org.apache.hadoop.util.NativeCodeLoader;
 import org.apache.hadoop.util.Shell;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+import edu.brown.cs.systems.xtrace.XTrace;
 
 /**
  * JNI wrappers for various native IO-related calls not available in Java.
@@ -90,6 +91,7 @@ public static class POSIX {
        write.  */
     public static final int SYNC_FILE_RANGE_WAIT_AFTER = 4;
 
+    private static final XTrace.Logger xtrace = XTrace.getLogger(NativeIO.class);
     private static final Log LOG = LogFactory.getLog(NativeIO.class);
 
     private static boolean nativeLoaded = false;
@@ -184,6 +186,7 @@ public static void posixFadviseIfPossible(
       if (nativeLoaded && fadvisePossible) {
         try {
           posix_fadvise(fd, offset, len, flags);
+          xtrace.log("posix_fadvise", "offset", offset, "len", len, "flags", flags);
         } catch (UnsupportedOperationException uoe) {
           fadvisePossible = false;
         } catch (UnsatisfiedLinkError ule) {
@@ -205,6 +208,7 @@ public static void syncFileRangeIfPossible(
       if (nativeLoaded && syncFileRangePossible) {
         try {
           sync_file_range(fd, offset, nbytes, flags);
+          xtrace.log("sync_file_range", "offset", offset, "nbytes", nbytes, "flags", flags);
         } catch (UnsupportedOperationException uoe) {
           syncFileRangePossible = false;
         } catch (UnsatisfiedLinkError ule) {

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Client.java

@@ -18,7 +18,7 @@
 
 package org.apache.hadoop.ipc;
 
-import static org.apache.hadoop.ipc.RpcConstants.*;
+import static org.apache.hadoop.ipc.RpcConstants.PING_CALL_ID;
 
 import java.io.BufferedInputStream;
 import java.io.BufferedOutputStream;
@@ -92,8 +92,13 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.google.protobuf.ByteString;
 import com.google.protobuf.CodedOutputStream;
 
+import edu.brown.cs.systems.resourcetracing.resources.Network;
+import edu.brown.cs.systems.xtrace.Context;
+import edu.brown.cs.systems.xtrace.XTrace;
+
 /** A client for an IPC service.  IPC calls take a single {@link Writable} as a
  * parameter, and return a {@link Writable} as their value.  A service runs on
  * a port and is defined by a parameter class and a value class.
@@ -102,6 +107,7 @@
  */
 public class Client {
 
+  public static final XTrace.Logger xtrace = XTrace.getLogger(Client.class);
   public static final Log LOG = LogFactory.getLog(Client.class);
 
   /** A counter for generating call IDs. */
@@ -238,6 +244,9 @@ void checkResponse(RpcResponseHeaderProto header) throws IOException {
         }
       }
     }
+    if (header.hasXtrace()) {
+      XTrace.set(header.getXtrace().toByteArray());
+    }
   }
 
   Call createCall(RPC.RpcKind rpcKind, Writable rpcRequest) {
@@ -255,6 +264,7 @@ static class Call {
     IOException error;          // exception, null if success
     final RPC.RpcKind rpcKind;      // Rpc EngineKind
     boolean done;               // true when call is done
+    Context xtrace;      // X-Trace context for the return
 
     private Call(RPC.RpcKind rpcKind, Writable param) {
       this.rpcKind = rpcKind;
@@ -631,6 +641,8 @@ private synchronized void setupIOstreams() {
       if (socket != null || shouldCloseConnection.get()) {
         return;
       } 
+      xtrace.log("Connecting to server");
+      Context start_context = XTrace.get();
       try {
         if (LOG.isDebugEnabled()) {
           LOG.debug("Connecting to "+server);
@@ -702,9 +714,13 @@ public AuthMethod run()
           // start the receiver thread after the socket connection has been set
           // up
           start();
+          XTrace.join(start_context);
+          xtrace.log("Connected to server");
           return;
         }
       } catch (Throwable t) {
+        XTrace.join(start_context);
+        xtrace.log("Failed to connect to server: "+t.getClass().getName(), "Message", t.getMessage());
         if (t instanceof IOException) {
           markClosed((IOException)t);
         } else {
@@ -882,10 +898,13 @@ private synchronized void sendPing() throws IOException {
 
     @Override
     public void run() {
+      // XTrace: this is a long-lived thread that is lazily started by the first client.
+      // Clear any thread contexts that may have been set after the thread begins
+      xtrace.log("RPC Response reader thread started");
       if (LOG.isDebugEnabled())
         LOG.debug(getName() + ": starting, having connections " 
             + connections.size());
-
+      XTrace.stop();
       try {
         while (waitForWork()) {//wait here for work - read or close connection
           receiveRpcResponse();
@@ -946,6 +965,7 @@ public void run() {
 
                 if (LOG.isDebugEnabled())
                   LOG.debug(getName() + " sending #" + call.id);
+                xtrace.log("Client senderFuture sending call");
 
                 byte[] data = d.getData();
                 int totalLength = d.getLength();
@@ -991,24 +1011,36 @@ private void receiveRpcResponse() {
       }
       touch();
 
+      XTrace.stop();
       try {
-        int totalLen = in.readInt();
-        RpcResponseHeaderProto header = 
-            RpcResponseHeaderProto.parseDelimitedFrom(in);
+        Network.ignore(true);
+        int totalLen;
+        RpcResponseHeaderProto header;
+        try {
+          totalLen = in.readInt();
+          header = RpcResponseHeaderProto.parseDelimitedFrom(in);
+        } finally {
+          Network.ignore(false);
+        }
         checkResponse(header);
 
         int headerLen = header.getSerializedSize();
         headerLen += CodedOutputStream.computeRawVarint32Size(headerLen);
+
+        Network.Read.alreadyStarted(in);
+        Network.Read.alreadyFinished(in, headerLen);
 
         int callId = header.getCallId();
         if (LOG.isDebugEnabled())
           LOG.debug(getName() + " got value #" + callId);
 
-        Call call = calls.get(callId);
+        Call call = calls.get(callId);        
         RpcStatusProto status = header.getStatus();
+
         if (status == RpcStatusProto.SUCCESS) {
           Writable value = ReflectionUtils.newInstance(valueClass, conf);
           value.readFields(in);                 // read value
+          call.xtrace = XTrace.get();
           call.setRpcResponse(value);
           calls.remove(callId);
 
@@ -1053,6 +1085,8 @@ private void receiveRpcResponse() {
         }
       } catch (IOException e) {
         markClosed(e);
+      } finally {
+        XTrace.stop();
       }
     }
 
@@ -1316,6 +1350,8 @@ public Writable call(RPC.RpcKind rpcKind, Writable rpcRequest,
       ConnectionId remoteId, int serviceClass) throws IOException {
     final Call call = createCall(rpcKind, rpcRequest);
     Connection connection = getConnection(remoteId, call, serviceClass);
+    xtrace.log("Sending RPC request", "Call ID", call.id);
+    call.xtrace = XTrace.get();
     try {
       connection.sendRpcRequest(call);                 // send the rpc request
     } catch (RejectedExecutionException e) {
@@ -1325,7 +1361,6 @@ public Writable call(RPC.RpcKind rpcKind, Writable rpcRequest,
       LOG.warn("interrupted waiting to send rpc request to server", e);
       throw new IOException(e);
     }
-
     boolean interrupted = false;
     synchronized (call) {
       while (!call.done) {
@@ -1342,19 +1377,24 @@ public Writable call(RPC.RpcKind rpcKind, Writable rpcRequest,
         Thread.currentThread().interrupt();
       }
 
+      XTrace.join(call.xtrace);
       if (call.error != null) {
         if (call.error instanceof RemoteException) {
           call.error.fillInStackTrace();
+          xtrace.log("RPC response received remote exception", "Call ID", call.id, "Message", call.error.getMessage());
           throw call.error;
         } else { // local exception
           InetSocketAddress address = connection.getRemoteAddress();
-          throw NetUtils.wrapException(address.getHostName(),
+          IOException e = NetUtils.wrapException(address.getHostName(),
                   address.getPort(),
                   NetUtils.getHostname(),
                   0,
                   call.error);
+          xtrace.log("Local exception handling RPC response", "Call ID", call.id, "Message", e.getMessage());
+          throw e;
         }
       } else {
+        xtrace.log("Received RPC response", "Call ID", call.id);
         return call.getRpcResponse();
       }
     }