param, to the IPC server running at
* address, returning the value. Throws exceptions if there are
* network problems or if the remote code threw an exception.
- * @deprecated Use {@link #call(Writable, ConnectionId)} instead
+ * @deprecated Use {@link #call(RpcKind, Writable, ConnectionId)} instead
*/
@Deprecated
- public Writable call(Writable param, InetSocketAddress address)
+ public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress address)
throws InterruptedException, IOException {
- return call(param, address, null);
+ return call(rpcKind, param, address, null);
}
/** Make a call, passing param, to the IPC server running at
@@ -1009,15 +1025,15 @@ public Writable call(Writable param, InetSocketAddress address)
* the value.
* Throws exceptions if there are network problems or if the remote code
* threw an exception.
- * @deprecated Use {@link #call(Writable, ConnectionId)} instead
+ * @deprecated Use {@link #call(RpcKind, Writable, ConnectionId)} instead
*/
@Deprecated
- public Writable call(Writable param, InetSocketAddress addr,
+ public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress addr,
UserGroupInformation ticket)
throws InterruptedException, IOException {
ConnectionId remoteId = ConnectionId.getConnectionId(addr, null, ticket, 0,
conf);
- return call(param, remoteId);
+ return call(rpcKind, param, remoteId);
}
/** Make a call, passing param, to the IPC server running at
@@ -1026,18 +1042,33 @@ public Writable call(Writable param, InetSocketAddress addr,
* timeout, returning the value.
* Throws exceptions if there are network problems or if the remote code
* threw an exception.
- * @deprecated Use {@link #call(Writable, ConnectionId)} instead
+ * @deprecated Use {@link #call(RpcKind, Writable, ConnectionId)} instead
*/
@Deprecated
- public Writable call(Writable param, InetSocketAddress addr,
+ public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress addr,
Class protocol, UserGroupInformation ticket,
int rpcTimeout)
throws InterruptedException, IOException {
ConnectionId remoteId = ConnectionId.getConnectionId(addr, protocol,
ticket, rpcTimeout, conf);
- return call(param, remoteId);
+ return call(rpcKind, param, remoteId);
}
+
+ /**
+ * Same as {@link #call(RpcKind, Writable, InetSocketAddress,
+ * Class, UserGroupInformation, int, Configuration)}
+ * except that rpcKind is writable.
+ */
+ public Writable call(Writable param, InetSocketAddress addr,
+ Class protocol, UserGroupInformation ticket,
+ int rpcTimeout, Configuration conf)
+ throws InterruptedException, IOException {
+ ConnectionId remoteId = ConnectionId.getConnectionId(addr, protocol,
+ ticket, rpcTimeout, conf);
+ return call(RpcKind.RPC_WRITABLE, param, remoteId);
+ }
+
/**
* Make a call, passing param, to the IPC server running at
* address which is servicing the protocol protocol,
@@ -1046,22 +1077,31 @@ public Writable call(Writable param, InetSocketAddress addr,
* value. Throws exceptions if there are network problems or if the remote
* code threw an exception.
*/
- public Writable call(Writable param, InetSocketAddress addr,
+ public Writable call(RpcKind rpcKind, Writable param, InetSocketAddress addr,
Class protocol, UserGroupInformation ticket,
int rpcTimeout, Configuration conf)
throws InterruptedException, IOException {
ConnectionId remoteId = ConnectionId.getConnectionId(addr, protocol,
ticket, rpcTimeout, conf);
- return call(param, remoteId);
+ return call(rpcKind, param, remoteId);
+ }
+
+ /**
+ * Same as {link {@link #call(RpcKind, Writable, ConnectionId)}
+ * except the rpcKind is RPC_WRITABLE
+ */
+ public Writable call(Writable param, ConnectionId remoteId)
+ throws InterruptedException, IOException {
+ return call(RpcKind.RPC_WRITABLE, param, remoteId);
}
/** Make a call, passing param, to the IPC server defined by
* remoteId, returning the value.
* Throws exceptions if there are network problems or if the remote code
* threw an exception. */
- public Writable call(Writable param, ConnectionId remoteId)
+ public Writable call(RpcKind rpcKind, Writable param, ConnectionId remoteId)
throws InterruptedException, IOException {
- Call call = new Call(param);
+ Call call = new Call(rpcKind, param);
Connection connection = getConnection(remoteId, call);
connection.sendParam(call); // send the parameter
boolean interrupted = false;
@@ -1093,7 +1133,7 @@ public Writable call(Writable param, ConnectionId remoteId)
call.error);
}
} else {
- return call.value;
+ return call.rpcResponse;
}
}
}
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
index aec56a9d57a..dad94227ca2 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/ProtobufRpcEngine.java
@@ -37,6 +37,7 @@
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcKind;
import org.apache.hadoop.ipc.protobuf.HadoopRpcProtos.HadoopRpcExceptionProto;
import org.apache.hadoop.ipc.protobuf.HadoopRpcProtos.HadoopRpcRequestProto;
import org.apache.hadoop.ipc.protobuf.HadoopRpcProtos.HadoopRpcResponseProto;
@@ -139,7 +140,7 @@ public Object invoke(Object proxy, Method method, Object[] args)
HadoopRpcRequestProto rpcRequest = constructRpcRequest(method, args);
RpcResponseWritable val = null;
try {
- val = (RpcResponseWritable) client.call(
+ val = (RpcResponseWritable) client.call(RpcKind.RPC_PROTOCOL_BUFFER,
new RpcRequestWritable(rpcRequest), remoteId);
} catch (Exception e) {
RpcClientException ce = new RpcClientException("Client exception", e);
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RpcPayloadHeader.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RpcPayloadHeader.java
new file mode 100644
index 00000000000..1b62f0caa7f
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RpcPayloadHeader.java
@@ -0,0 +1,118 @@
+package org.apache.hadoop.ipc;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.io.Writable;
+
+/**
+ * This is the rpc payload header. It is sent with every rpc call
+ *
+ * The format of RPC call is as follows:
+ * +---------------------------------------------------+
+ * | Rpc length in bytes (header + payload length) |
+ * +---------------------------------------------------+
+ * | Rpc Header | Rpc Payload |
+ * +---------------------------------------------------+
+ *
+ * The format of Rpc Header is:
+ * +----------------------------------+
+ * | RpcKind (1 bytes) |
+ * +----------------------------------+
+ * | RpcPayloadOperation (1 bytes) |
+ * +----------------------------------+
+ * | Call ID (4 bytes) |
+ * +----------------------------------+
+ *
+ * {@link RpcKind} determines the type of serialization used for Rpc Payload.
+ *
+ *
+ * Note this header does NOT have its own version number,
+ * it used the version number from the connection header.
+ */
+public class RpcPayloadHeader implements Writable {
+ public enum RpcPayloadOperation {
+ RPC_FINAL_PAYLOAD ((short)1),
+ RPC_CONTINUATION_PAYLOAD ((short)2), // not implemented yet
+ RPC_CLOSE_CONNECTION ((short)3); // close the rpc connection
+
+ private final short code;
+ private static final short FIRST_INDEX = RPC_FINAL_PAYLOAD.code;
+ RpcPayloadOperation(short val) {
+ this.code = val;
+ }
+
+ public void write(DataOutput out) throws IOException {
+ out.writeByte(code);
+ }
+
+ static RpcPayloadOperation readFields(DataInput in) throws IOException {
+ short inValue = in.readByte();
+ return RpcPayloadOperation.values()[inValue - FIRST_INDEX];
+ }
+ }
+
+ public enum RpcKind {
+ RPC_BUILTIN ((short ) 1), // Used for built in calls
+ RPC_WRITABLE ((short ) 2),
+ RPC_PROTOCOL_BUFFER ((short)3),
+ RPC_AVRO ((short)4);
+
+ private final short value;
+ private static final short FIRST_INDEX = RPC_BUILTIN.value;
+ RpcKind(short val) {
+ this.value = val;
+ }
+
+ public void write(DataOutput out) throws IOException {
+ out.writeByte(value);
+ }
+
+ static RpcKind readFields(DataInput in) throws IOException {
+ short inValue = in.readByte();
+ return RpcKind.values()[inValue - FIRST_INDEX];
+ }
+ }
+
+ private RpcKind kind;
+ private RpcPayloadOperation operation;
+ private int callId;
+
+ public RpcPayloadHeader() {
+ kind = RpcKind.RPC_WRITABLE;
+ operation = RpcPayloadOperation.RPC_CLOSE_CONNECTION;
+ }
+
+ public RpcPayloadHeader(RpcKind kind, RpcPayloadOperation op, int callId) {
+ this.kind = kind;
+ this.operation = op;
+ this.callId = callId;
+ }
+
+ int getCallId() {
+ return callId;
+ }
+
+ RpcKind getkind() {
+ return kind;
+ }
+
+ RpcPayloadOperation getOperation() {
+ return operation;
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ kind.write(out);
+ operation.write(out);
+ out.writeInt(callId);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ kind = RpcKind.readFields(in);
+ operation = RpcPayloadOperation.readFields(in);
+ this.callId = in.readInt();
+ }
+}
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
index 78c0bc76a10..c04833f232d 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
@@ -61,11 +61,13 @@
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.hadoop.ipc.RpcPayloadHeader.*;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.ipc.RPC.VersionMismatch;
+import org.apache.hadoop.ipc.RpcPayloadHeader.RpcPayloadOperation;
import org.apache.hadoop.ipc.metrics.RpcDetailedMetrics;
import org.apache.hadoop.ipc.metrics.RpcMetrics;
import org.apache.hadoop.net.NetUtils;
@@ -124,7 +126,8 @@ public abstract class Server {
// 4 : Introduced SASL security layer
// 5 : Introduced use of {@link ArrayPrimitiveWritable$Internal}
// in ObjectWritable to efficiently transmit arrays of primitives
- public static final byte CURRENT_VERSION = 5;
+ // 6 : Made RPC payload header explicit
+ public static final byte CURRENT_VERSION = 6;
/**
* Initial and max size of response buffer
@@ -277,28 +280,33 @@ public ServiceAuthorizationManager getServiceAuthorizationManager() {
/** A call queued for handling. */
private static class Call {
- private int id; // the client's call id
- private Writable param; // the parameter passed
- private Connection connection; // connection to client
- private long timestamp; // the time received when response is null
- // the time served when response is not null
- private ByteBuffer response; // the response for this call
+ private final int callId; // the client's call id
+ private final Writable rpcRequest; // Serialized Rpc request from client
+ private final Connection connection; // connection to client
+ private long timestamp; // time received when response is null
+ // time served when response is not null
+ private ByteBuffer rpcResponse; // the response for this call
+ private final RpcKind rpcKind;
- public Call(int id, Writable param, Connection connection) {
- this.id = id;
- this.param = param;
+ public Call(int id, Writable param, Connection connection) {
+ this( id, param, connection, RpcKind.RPC_BUILTIN );
+ }
+ public Call(int id, Writable param, Connection connection, RpcKind kind) {
+ this.callId = id;
+ this.rpcRequest = param;
this.connection = connection;
this.timestamp = System.currentTimeMillis();
- this.response = null;
+ this.rpcResponse = null;
+ this.rpcKind = kind;
}
@Override
public String toString() {
- return param.toString() + " from " + connection.toString();
+ return rpcRequest.toString() + " from " + connection.toString();
}
public void setResponse(ByteBuffer response) {
- this.response = response;
+ this.rpcResponse = response;
}
}
@@ -795,17 +803,17 @@ private boolean processResponse(LinkedList