hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cutt...@apache.org
Subject svn commit: r1177399 - in /hadoop/common/trunk/hadoop-common-project/hadoop-common: ./ src/main/java/org/apache/hadoop/ipc/ src/test/java/org/apache/hadoop/ipc/
Date Thu, 29 Sep 2011 19:03:11 GMT
Author: cutting
Date: Thu Sep 29 19:03:10 2011
New Revision: 1177399

URL: http://svn.apache.org/viewvc?rev=1177399&view=rev
Log:
HADOOP-7693. Enhance AvroRpcEngine to support the new #addProtocol introduced in HADOOP-724.

Modified:
    hadoop/common/trunk/hadoop-common-project/hadoop-common/CHANGES.txt
    hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/AvroRpcEngine.java
    hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RPC.java
    hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java
    hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAvroRpc.java

Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/CHANGES.txt?rev=1177399&r1=1177398&r2=1177399&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/CHANGES.txt (original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/CHANGES.txt Thu Sep 29 19:03:10
2011
@@ -413,6 +413,9 @@ Release 0.23.0 - Unreleased
     HADOOP-7575. Enhanced LocalDirAllocator to support fully-qualified
     paths. (Jonathan Eagles via vinodkv)
 
+    HADOOP-7693. Enhance AvroRpcEngine to support the new #addProtocol
+    interface introduced in HADOOP-7524.  (cutting)
+
   OPTIMIZATIONS
   
     HADOOP-7333. Performance improvement in PureJavaCrc32. (Eric Caspole

Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/AvroRpcEngine.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/AvroRpcEngine.java?rev=1177399&r1=1177398&r2=1177399&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/AvroRpcEngine.java
(original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/AvroRpcEngine.java
Thu Sep 29 19:03:10 2011
@@ -29,6 +29,8 @@ import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
+import java.util.HashMap;
 
 import javax.net.SocketFactory;
 
@@ -54,7 +56,7 @@ import org.apache.hadoop.security.token.
 public class AvroRpcEngine implements RpcEngine {
   private static final Log LOG = LogFactory.getLog(RPC.class);
 
-  private static int VERSION = 0;
+  private static int VERSION = 1;
 
   // the implementation we tunnel through
   private static final RpcEngine ENGINE = new WritableRpcEngine();
@@ -62,9 +64,10 @@ public class AvroRpcEngine implements Rp
   /** Tunnel an Avro RPC request and response through Hadoop's RPC. */
   private static interface TunnelProtocol extends VersionedProtocol {
     //WritableRpcEngine expects a versionID in every protocol.
-    public static final long versionID = 0L;
+    public static final long versionID = VERSION;
     /** All Avro methods and responses go through this. */
-    BufferListWritable call(BufferListWritable request) throws IOException;
+    BufferListWritable call(String protocol, BufferListWritable request)
+      throws IOException;
   }
 
   /** A Writable that holds a List<ByteBuffer>, The Avro RPC Transceiver's
@@ -103,23 +106,25 @@ public class AvroRpcEngine implements Rp
   private static class ClientTransceiver extends Transceiver {
     private TunnelProtocol tunnel;
     private InetSocketAddress remote;
+    private String protocol;
   
     public ClientTransceiver(InetSocketAddress addr,
                              UserGroupInformation ticket,
                              Configuration conf, SocketFactory factory,
-                             int rpcTimeout)
+                             int rpcTimeout, String protocol)
       throws IOException {
       this.tunnel = ENGINE.getProxy(TunnelProtocol.class, VERSION,
                                         addr, ticket, conf, factory,
                                         rpcTimeout).getProxy();
       this.remote = addr;
+      this.protocol = protocol;
     }
 
     public String getRemoteName() { return remote.toString(); }
 
     public List<ByteBuffer> transceive(List<ByteBuffer> request)
       throws IOException {
-      return tunnel.call(new BufferListWritable(request)).buffers;
+      return tunnel.call(protocol, new BufferListWritable(request)).buffers;
     }
 
     public List<ByteBuffer> readBuffers() throws IOException {
@@ -159,7 +164,8 @@ public class AvroRpcEngine implements Rp
                    UserGroupInformation ticket, Configuration conf,
                    SocketFactory factory,
                    int rpcTimeout) throws IOException {
-      this.tx = new ClientTransceiver(addr, ticket, conf, factory, rpcTimeout);
+      this.tx = new ClientTransceiver(addr, ticket, conf, factory, rpcTimeout,
+                                      protocol.getName());
       this.requestor = createRequestor(protocol, tx);
     }
     @Override public Object invoke(Object proxy, Method method, Object[] args) 
@@ -182,9 +188,11 @@ public class AvroRpcEngine implements Rp
 
   /** An Avro RPC Responder that can process requests passed via Hadoop RPC. */
   private class TunnelResponder implements TunnelProtocol {
-    private Responder responder;
-    public TunnelResponder(Class<?> iface, Object impl) {
-      responder = createResponder(iface, impl);
+    private Map<String, Responder> responders =
+      new HashMap<String, Responder>();
+
+    public void addProtocol(Class<?> iface, Object impl) {
+      responders.put(iface.getName(), createResponder(iface, impl));
     }
 
     @Override
@@ -197,13 +205,18 @@ public class AvroRpcEngine implements Rp
     public ProtocolSignature getProtocolSignature(
         String protocol, long version, int clientMethodsHashCode)
       throws IOException {
-      return new ProtocolSignature(VERSION, null);
+      return ProtocolSignature.getProtocolSignature
+        (clientMethodsHashCode, VERSION, TunnelProtocol.class);
     }
 
-    public BufferListWritable call(final BufferListWritable request)
+    public BufferListWritable call(String protocol, BufferListWritable request)
       throws IOException {
+      Responder responder = responders.get(protocol);
+      if (responder == null)
+        throw new IOException("No responder for: "+protocol);
       return new BufferListWritable(responder.respond(request.buffers));
     }
+
   }
 
   public Object[] call(Method method, Object[][] params,
@@ -212,6 +225,32 @@ public class AvroRpcEngine implements Rp
     throw new UnsupportedOperationException();
   }
 
+  private class Server extends WritableRpcEngine.Server {
+    private TunnelResponder responder = new TunnelResponder();
+
+    public Server(Class<?> iface, Object impl, String bindAddress,
+                  int port, int numHandlers, int numReaders,
+                  int queueSizePerHandler, boolean verbose,
+                  Configuration conf, 
+                  SecretManager<? extends TokenIdentifier> secretManager
+                  ) throws IOException {
+      super((Class)null, new Object(), conf,
+            bindAddress, port, numHandlers, numReaders,
+            queueSizePerHandler, verbose, secretManager);
+      super.addProtocol(TunnelProtocol.class, responder);
+      responder.addProtocol(iface, impl);
+    }
+
+
+    @Override
+    public <PROTO, IMPL extends PROTO> Server
+      addProtocol(Class<PROTO> protocolClass, IMPL protocolImpl)
+        throws IOException {
+      responder.addProtocol(protocolClass, protocolImpl);
+      return this;
+    }
+  }
+
   /** Construct a server for a protocol implementation instance listening on a
    * port and address. */
   public RPC.Server getServer(Class<?> iface, Object impl, String bindAddress,
@@ -220,10 +259,9 @@ public class AvroRpcEngine implements Rp
                               Configuration conf, 
                        SecretManager<? extends TokenIdentifier> secretManager
                               ) throws IOException {
-    return ENGINE.getServer(TunnelProtocol.class,
-                            new TunnelResponder(iface, impl),
-                            bindAddress, port, numHandlers, numReaders,
-                            queueSizePerHandler, verbose, conf, secretManager);
+    return new Server
+      (iface, impl, bindAddress, port, numHandlers, numReaders,
+       queueSizePerHandler, verbose, conf, secretManager);
   }
 
 }

Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RPC.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RPC.java?rev=1177399&r1=1177398&r2=1177399&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RPC.java
(original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RPC.java
Thu Sep 29 19:03:10 2011
@@ -605,7 +605,7 @@ public class RPC {
      * @param protocolImpl - the impl of the protocol that will be called
      * @return the server (for convenience)
      */
-    public <PROTO extends VersionedProtocol, IMPL extends PROTO>
+    public <PROTO, IMPL extends PROTO>
       Server addProtocol(Class<PROTO> protocolClass, IMPL protocolImpl
     ) throws IOException {
       throw new IOException("addProtocol Not Implemented");

Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java?rev=1177399&r1=1177398&r2=1177399&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java
(original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java
Thu Sep 29 19:03:10 2011
@@ -555,7 +555,7 @@ public class WritableRpcEngine implement
 
  
     @Override
-    public <PROTO extends VersionedProtocol, IMPL extends PROTO> Server
+    public <PROTO, IMPL extends PROTO> Server
       addProtocol(
         Class<PROTO> protocolClass, IMPL protocolImpl) throws IOException {
       registerProtocolAndImpl(protocolClass, protocolImpl);

Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAvroRpc.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAvroRpc.java?rev=1177399&r1=1177398&r2=1177399&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAvroRpc.java
(original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestAvroRpc.java
Thu Sep 29 19:03:10 2011
@@ -43,6 +43,7 @@ import org.apache.hadoop.security.Securi
 import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
 
 /** Unit tests for AvroRpc. */
 public class TestAvroRpc extends TestCase {
@@ -56,6 +57,9 @@ public class TestAvroRpc extends TestCas
 
   public TestAvroRpc(String name) { super(name); }
 	
+  public static interface EmptyProtocol {}
+  public static class EmptyImpl implements EmptyProtocol {}
+
   public static class TestImpl implements AvroTestProtocol {
 
     public void ping() {}
@@ -93,10 +97,12 @@ public class TestAvroRpc extends TestCas
       sm = new TestTokenSecretManager();
     }
     UserGroupInformation.setConfiguration(conf);
+    RPC.setProtocolEngine(conf, EmptyProtocol.class, AvroRpcEngine.class);
     RPC.setProtocolEngine(conf, AvroTestProtocol.class, AvroRpcEngine.class);
-    Server server = RPC.getServer(AvroTestProtocol.class,
-                                  new TestImpl(), ADDRESS, 0, 5, true, 
-                                  conf, sm);
+    RPC.Server server = RPC.getServer(EmptyProtocol.class, new EmptyImpl(),
+                                      ADDRESS, 0, 5, true, conf, sm);
+    server.addProtocol(AvroTestProtocol.class, new TestImpl());
+
     try {
       server.start();
       InetSocketAddress addr = NetUtils.getConnectAddress(server);



Mime
View raw message