branch-0.23.1/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/WritableRpcEngine.java - hadoop - Git at Google

 /**
  * 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.ipc;

 import java.lang.reflect.Field;
 import java.lang.reflect.Proxy;
 import java.lang.reflect.Method;
 import java.lang.reflect.Array;
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.InvocationTargetException;

 import java.net.InetSocketAddress;
 import java.io.*;
 import java.util.Map;
 import java.util.HashMap;

 import javax.net.SocketFactory;

 import org.apache.commons.logging.*;

 import org.apache.hadoop.io.*;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.SecretManager;
 import org.apache.hadoop.security.token.TokenIdentifier;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.*;

 /** An RpcEngine implementation for Writable data. */
 @InterfaceStability.Evolving
 public class WritableRpcEngine implements RpcEngine {
   private static final Log LOG = LogFactory.getLog(RPC.class);

   //writableRpcVersion should be updated if there is a change
   //in format of the rpc messages.
   public static long writableRpcVersion = 1L;

   /** A method invocation, including the method name and its parameters.*/
   private static class Invocation implements Writable, Configurable {
     private String methodName;
     private Class<?>[] parameterClasses;
     private Object[] parameters;
     private Configuration conf;
     private long clientVersion;
     private int clientMethodsHash;

     //This could be different from static writableRpcVersion when received
     //at server, if client is using a different version.
     private long rpcVersion;

     public Invocation() {}

     public Invocation(Method method, Object[] parameters) {
       this.methodName = method.getName();
       this.parameterClasses = method.getParameterTypes();
       this.parameters = parameters;
       rpcVersion = writableRpcVersion;
       if (method.getDeclaringClass().equals(VersionedProtocol.class)) {
         //VersionedProtocol is exempted from version check.
         clientVersion = 0;
         clientMethodsHash = 0;
       } else {
         try {
           Field versionField = method.getDeclaringClass().getField("versionID");
           versionField.setAccessible(true);
           this.clientVersion = versionField.getLong(method.getDeclaringClass());
         } catch (NoSuchFieldException ex) {
           throw new RuntimeException(ex);
         } catch (IllegalAccessException ex) {
           throw new RuntimeException(ex);
         }
         this.clientMethodsHash = ProtocolSignature.getFingerprint(method
             .getDeclaringClass().getMethods());
       }
     }

     /** The name of the method invoked. */
     public String getMethodName() { return methodName; }

     /** The parameter classes. */
     public Class<?>[] getParameterClasses() { return parameterClasses; }

     /** The parameter instances. */
     public Object[] getParameters() { return parameters; }

     private long getProtocolVersion() {
       return clientVersion;
     }

     private int getClientMethodsHash() {
       return clientMethodsHash;
     }

     /**
      * Returns the rpc version used by the client.
      * @return rpcVersion
      */
     public long getRpcVersion() {
       return rpcVersion;
     }

     public void readFields(DataInput in) throws IOException {
       rpcVersion = in.readLong();
       methodName = UTF8.readString(in);
       clientVersion = in.readLong();
       clientMethodsHash = in.readInt();
       parameters = new Object[in.readInt()];
       parameterClasses = new Class[parameters.length];
       ObjectWritable objectWritable = new ObjectWritable();
       for (int i = 0; i < parameters.length; i++) {
         parameters[i] = ObjectWritable.readObject(in, objectWritable, this.conf);
         parameterClasses[i] = objectWritable.getDeclaredClass();
       }
     }

     public void write(DataOutput out) throws IOException {
       out.writeLong(rpcVersion);
       UTF8.writeString(out, methodName);
       out.writeLong(clientVersion);
       out.writeInt(clientMethodsHash);
       out.writeInt(parameterClasses.length);
       for (int i = 0; i < parameterClasses.length; i++) {
         ObjectWritable.writeObject(out, parameters[i], parameterClasses[i],
                                    conf, true);
       }
     }

     public String toString() {
       StringBuilder buffer = new StringBuilder();
       buffer.append(methodName);
       buffer.append("(");
       for (int i = 0; i < parameters.length; i++) {
         if (i != 0)
           buffer.append(", ");
         buffer.append(parameters[i]);
       }
       buffer.append(")");
       buffer.append(", rpc version="+rpcVersion);
       buffer.append(", client version="+clientVersion);
       buffer.append(", methodsFingerPrint="+clientMethodsHash);
       return buffer.toString();
     }

     public void setConf(Configuration conf) {
       this.conf = conf;
     }

     public Configuration getConf() {
       return this.conf;
     }

   }

   private static ClientCache CLIENTS=new ClientCache();

   private static class Invoker implements InvocationHandler {
     private Client.ConnectionId remoteId;
     private Client client;
     private boolean isClosed = false;

     public Invoker(Class<?> protocol,
                    InetSocketAddress address, UserGroupInformation ticket,
                    Configuration conf, SocketFactory factory,
                    int rpcTimeout) throws IOException {
       this.remoteId = Client.ConnectionId.getConnectionId(address, protocol,
           ticket, rpcTimeout, conf);
       this.client = CLIENTS.getClient(conf, factory);
     }

     public Object invoke(Object proxy, Method method, Object[] args)
       throws Throwable {
       long startTime = 0;
       if (LOG.isDebugEnabled()) {
         startTime = System.currentTimeMillis();
       }

       ObjectWritable value = (ObjectWritable)
         client.call(new Invocation(method, args), remoteId);
       if (LOG.isDebugEnabled()) {
         long callTime = System.currentTimeMillis() - startTime;
         LOG.debug("Call: " + method.getName() + " " + callTime);
       }
       return value.get();
     }

     /* close the IPC client that's responsible for this invoker's RPCs */
     synchronized private void close() {
       if (!isClosed) {
         isClosed = true;
         CLIENTS.stopClient(client);
       }
     }
   }

   // for unit testing only
   @InterfaceAudience.Private
   @InterfaceStability.Unstable
   static Client getClient(Configuration conf) {
     return CLIENTS.getClient(conf);
   }

   /** Construct a client-side proxy object that implements the named protocol,
    * talking to a server at the named address.
    * @param <T>*/
   @Override
   @SuppressWarnings("unchecked")
   public <T> ProtocolProxy<T> getProxy(Class<T> protocol, long clientVersion,
                          InetSocketAddress addr, UserGroupInformation ticket,
                          Configuration conf, SocketFactory factory,
                          int rpcTimeout)
     throws IOException {

     T proxy = (T) Proxy.newProxyInstance(protocol.getClassLoader(),
         new Class[] { protocol }, new Invoker(protocol, addr, ticket, conf,
             factory, rpcTimeout));
     return new ProtocolProxy<T>(protocol, proxy, true);
   }

   /**
    * Stop this proxy and release its invoker's resource
    * @param proxy the proxy to be stopped
    */
   public void stopProxy(Object proxy) {
     ((Invoker)Proxy.getInvocationHandler(proxy)).close();
   }


   /** Expert: Make multiple, parallel calls to a set of servers. */
   public Object[] call(Method method, Object[][] params,
                        InetSocketAddress[] addrs,
                        UserGroupInformation ticket, Configuration conf)
     throws IOException, InterruptedException {

     Invocation[] invocations = new Invocation[params.length];
     for (int i = 0; i < params.length; i++)
       invocations[i] = new Invocation(method, params[i]);
     Client client = CLIENTS.getClient(conf);
     try {
     Writable[] wrappedValues =
       client.call(invocations, addrs, method.getDeclaringClass(), ticket, conf);

     if (method.getReturnType() == Void.TYPE) {
       return null;
     }

     Object[] values =
       (Object[])Array.newInstance(method.getReturnType(), wrappedValues.length);
     for (int i = 0; i < values.length; i++)
       if (wrappedValues[i] != null)
         values[i] = ((ObjectWritable)wrappedValues[i]).get();

     return values;
     } finally {
       CLIENTS.stopClient(client);
     }
   }

   /** Construct a server for a protocol implementation instance listening on a
    * port and address. */
   public Server getServer(Class<?> protocol,
                           Object instance, String bindAddress, int port,
                           int numHandlers, int numReaders, int queueSizePerHandler,
                           boolean verbose, Configuration conf,
                       SecretManager<? extends TokenIdentifier> secretManager)
     throws IOException {
     return new Server(instance, conf, bindAddress, port, numHandlers,
         numReaders, queueSizePerHandler, verbose, secretManager);
   }

   /** An RPC Server. */
   public static class Server extends RPC.Server {
     private Object instance;
     private boolean verbose;

     /** Construct an RPC server.
      * @param instance the instance whose methods will be called
      * @param conf the configuration to use
      * @param bindAddress the address to bind on to listen for connection
      * @param port the port to listen for connections on
      */
     public Server(Object instance, Configuration conf, String bindAddress, int port)
       throws IOException {
       this(instance, conf,  bindAddress, port, 1, -1, -1, false, null);
     }

     private static String classNameBase(String className) {
       String[] names = className.split("\\.", -1);
       if (names == null || names.length == 0) {
         return className;
       }
       return names[names.length-1];
     }

     /** Construct an RPC server.
      * @param instance the instance whose methods will be called
      * @param conf the configuration to use
      * @param bindAddress the address to bind on to listen for connection
      * @param port the port to listen for connections on
      * @param numHandlers the number of method handler threads to run
      * @param verbose whether each call should be logged
      */
     public Server(Object instance, Configuration conf, String bindAddress,  int port,
                   int numHandlers, int numReaders, int queueSizePerHandler, boolean verbose,
                   SecretManager<? extends TokenIdentifier> secretManager)
         throws IOException {
       super(bindAddress, port, Invocation.class, numHandlers, numReaders,
           queueSizePerHandler, conf,
           classNameBase(instance.getClass().getName()), secretManager);
       this.instance = instance;
       this.verbose = verbose;
     }

     public Writable call(Class<?> protocol, Writable param, long receivedTime)
     throws IOException {
       try {
         Invocation call = (Invocation)param;
         if (verbose) log("Call: " + call);

         Method method = protocol.getMethod(call.getMethodName(),
                                            call.getParameterClasses());
         method.setAccessible(true);

         // Verify rpc version
         if (call.getRpcVersion() != writableRpcVersion) {
           // Client is using a different version of WritableRpc
           throw new IOException(
               "WritableRpc version mismatch, client side version="
                   + call.getRpcVersion() + ", server side version="
                   + writableRpcVersion);
         }

         //Verify protocol version.
         //Bypass the version check for VersionedProtocol
         if (!method.getDeclaringClass().equals(VersionedProtocol.class)) {
           long clientVersion = call.getProtocolVersion();
           ProtocolSignature serverInfo = ((VersionedProtocol) instance)
               .getProtocolSignature(protocol.getCanonicalName(), call
                   .getProtocolVersion(), call.getClientMethodsHash());
           long serverVersion = serverInfo.getVersion();
           if (serverVersion != clientVersion) {
             LOG.warn("Version mismatch: client version=" + clientVersion
                 + ", server version=" + serverVersion);
             throw new RPC.VersionMismatch(protocol.getName(), clientVersion,
                 serverVersion);
           }
         }

         long startTime = System.currentTimeMillis();
         Object value = method.invoke(instance, call.getParameters());
         int processingTime = (int) (System.currentTimeMillis() - startTime);
         int qTime = (int) (startTime-receivedTime);
         if (LOG.isDebugEnabled()) {
           LOG.debug("Served: " + call.getMethodName() +
                     " queueTime= " + qTime +
                     " procesingTime= " + processingTime);
         }
         rpcMetrics.addRpcQueueTime(qTime);
         rpcMetrics.addRpcProcessingTime(processingTime);
         rpcDetailedMetrics.addProcessingTime(call.getMethodName(),
                                              processingTime);
         if (verbose) log("Return: "+value);

         return new ObjectWritable(method.getReturnType(), value);

       } catch (InvocationTargetException e) {
         Throwable target = e.getTargetException();
         if (target instanceof IOException) {
           throw (IOException)target;
         } else {
           IOException ioe = new IOException(target.toString());
           ioe.setStackTrace(target.getStackTrace());
           throw ioe;
         }
       } catch (Throwable e) {
         if (!(e instanceof IOException)) {
           LOG.error("Unexpected throwable object ", e);
         }
         IOException ioe = new IOException(e.toString());
         ioe.setStackTrace(e.getStackTrace());
         throw ioe;
       }
     }
   }

   private static void log(String value) {
     if (value!= null && value.length() > 55)
       value = value.substring(0, 55)+"...";
     LOG.info(value);
   }
 }
	/**
	* 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.ipc;

	import java.lang.reflect.Field;
	import java.lang.reflect.Proxy;
	import java.lang.reflect.Method;
	import java.lang.reflect.Array;
	import java.lang.reflect.InvocationHandler;
	import java.lang.reflect.InvocationTargetException;

	import java.net.InetSocketAddress;
	import java.io.*;
	import java.util.Map;
	import java.util.HashMap;

	import javax.net.SocketFactory;

	import org.apache.commons.logging.*;

	import org.apache.hadoop.io.*;
	import org.apache.hadoop.security.UserGroupInformation;
	import org.apache.hadoop.security.token.SecretManager;
	import org.apache.hadoop.security.token.TokenIdentifier;
	import org.apache.hadoop.classification.InterfaceAudience;
	import org.apache.hadoop.classification.InterfaceStability;
	import org.apache.hadoop.conf.*;

	/** An RpcEngine implementation for Writable data. */
	@InterfaceStability.Evolving
	public class WritableRpcEngine implements RpcEngine {
	private static final Log LOG = LogFactory.getLog(RPC.class);

	//writableRpcVersion should be updated if there is a change
	//in format of the rpc messages.
	public static long writableRpcVersion = 1L;

	/** A method invocation, including the method name and its parameters.*/
	private static class Invocation implements Writable, Configurable {
	private String methodName;
	private Class<?>[] parameterClasses;
	private Object[] parameters;
	private Configuration conf;
	private long clientVersion;
	private int clientMethodsHash;

	//This could be different from static writableRpcVersion when received
	//at server, if client is using a different version.
	private long rpcVersion;

	public Invocation() {}

	public Invocation(Method method, Object[] parameters) {
	this.methodName = method.getName();
	this.parameterClasses = method.getParameterTypes();
	this.parameters = parameters;
	rpcVersion = writableRpcVersion;
	if (method.getDeclaringClass().equals(VersionedProtocol.class)) {
	//VersionedProtocol is exempted from version check.
	clientVersion = 0;
	clientMethodsHash = 0;
	} else {
	try {
	Field versionField = method.getDeclaringClass().getField("versionID");
	versionField.setAccessible(true);
	this.clientVersion = versionField.getLong(method.getDeclaringClass());
	} catch (NoSuchFieldException ex) {
	throw new RuntimeException(ex);
	} catch (IllegalAccessException ex) {
	throw new RuntimeException(ex);
	}
	this.clientMethodsHash = ProtocolSignature.getFingerprint(method
	.getDeclaringClass().getMethods());
	}
	}

	/** The name of the method invoked. */
	public String getMethodName() { return methodName; }

	/** The parameter classes. */
	public Class<?>[] getParameterClasses() { return parameterClasses; }

	/** The parameter instances. */
	public Object[] getParameters() { return parameters; }

	private long getProtocolVersion() {
	return clientVersion;
	}

	private int getClientMethodsHash() {
	return clientMethodsHash;
	}

	/**
	* Returns the rpc version used by the client.
	* @return rpcVersion
	*/
	public long getRpcVersion() {
	return rpcVersion;
	}

	public void readFields(DataInput in) throws IOException {
	rpcVersion = in.readLong();
	methodName = UTF8.readString(in);
	clientVersion = in.readLong();
	clientMethodsHash = in.readInt();
	parameters = new Object[in.readInt()];
	parameterClasses = new Class[parameters.length];
	ObjectWritable objectWritable = new ObjectWritable();
	for (int i = 0; i < parameters.length; i++) {
	parameters[i] = ObjectWritable.readObject(in, objectWritable, this.conf);
	parameterClasses[i] = objectWritable.getDeclaredClass();
	}
	}

	public void write(DataOutput out) throws IOException {
	out.writeLong(rpcVersion);
	UTF8.writeString(out, methodName);
	out.writeLong(clientVersion);
	out.writeInt(clientMethodsHash);
	out.writeInt(parameterClasses.length);
	for (int i = 0; i < parameterClasses.length; i++) {
	ObjectWritable.writeObject(out, parameters[i], parameterClasses[i],
	conf, true);
	}
	}

	public String toString() {
	StringBuilder buffer = new StringBuilder();
	buffer.append(methodName);
	buffer.append("(");
	for (int i = 0; i < parameters.length; i++) {
	if (i != 0)
	buffer.append(", ");
	buffer.append(parameters[i]);
	}
	buffer.append(")");
	buffer.append(", rpc version="+rpcVersion);
	buffer.append(", client version="+clientVersion);
	buffer.append(", methodsFingerPrint="+clientMethodsHash);
	return buffer.toString();
	}

	public void setConf(Configuration conf) {
	this.conf = conf;
	}

	public Configuration getConf() {
	return this.conf;
	}

	}

	private static ClientCache CLIENTS=new ClientCache();

	private static class Invoker implements InvocationHandler {
	private Client.ConnectionId remoteId;
	private Client client;
	private boolean isClosed = false;

	public Invoker(Class<?> protocol,
	InetSocketAddress address, UserGroupInformation ticket,
	Configuration conf, SocketFactory factory,
	int rpcTimeout) throws IOException {
	this.remoteId = Client.ConnectionId.getConnectionId(address, protocol,
	ticket, rpcTimeout, conf);
	this.client = CLIENTS.getClient(conf, factory);
	}

	public Object invoke(Object proxy, Method method, Object[] args)
	throws Throwable {
	long startTime = 0;
	if (LOG.isDebugEnabled()) {
	startTime = System.currentTimeMillis();
	}

	ObjectWritable value = (ObjectWritable)
	client.call(new Invocation(method, args), remoteId);
	if (LOG.isDebugEnabled()) {
	long callTime = System.currentTimeMillis() - startTime;
	LOG.debug("Call: " + method.getName() + " " + callTime);
	}
	return value.get();
	}

	/* close the IPC client that's responsible for this invoker's RPCs */
	synchronized private void close() {
	if (!isClosed) {
	isClosed = true;
	CLIENTS.stopClient(client);
	}
	}
	}

	// for unit testing only
	@InterfaceAudience.Private
	@InterfaceStability.Unstable
	static Client getClient(Configuration conf) {
	return CLIENTS.getClient(conf);
	}

	/** Construct a client-side proxy object that implements the named protocol,
	* talking to a server at the named address.
	* @param <T>*/
	@Override
	@SuppressWarnings("unchecked")
	public <T> ProtocolProxy<T> getProxy(Class<T> protocol, long clientVersion,
	InetSocketAddress addr, UserGroupInformation ticket,
	Configuration conf, SocketFactory factory,
	int rpcTimeout)
	throws IOException {

	T proxy = (T) Proxy.newProxyInstance(protocol.getClassLoader(),
	new Class[] { protocol }, new Invoker(protocol, addr, ticket, conf,
	factory, rpcTimeout));
	return new ProtocolProxy<T>(protocol, proxy, true);
	}

	/**
	* Stop this proxy and release its invoker's resource
	* @param proxy the proxy to be stopped
	*/
	public void stopProxy(Object proxy) {
	((Invoker)Proxy.getInvocationHandler(proxy)).close();
	}


	/** Expert: Make multiple, parallel calls to a set of servers. */
	public Object[] call(Method method, Object[][] params,
	InetSocketAddress[] addrs,
	UserGroupInformation ticket, Configuration conf)
	throws IOException, InterruptedException {

	Invocation[] invocations = new Invocation[params.length];
	for (int i = 0; i < params.length; i++)
	invocations[i] = new Invocation(method, params[i]);
	Client client = CLIENTS.getClient(conf);
	try {
	Writable[] wrappedValues =
	client.call(invocations, addrs, method.getDeclaringClass(), ticket, conf);

	if (method.getReturnType() == Void.TYPE) {
	return null;
	}

	Object[] values =
	(Object[])Array.newInstance(method.getReturnType(), wrappedValues.length);
	for (int i = 0; i < values.length; i++)
	if (wrappedValues[i] != null)
	values[i] = ((ObjectWritable)wrappedValues[i]).get();

	return values;
	} finally {
	CLIENTS.stopClient(client);
	}
	}

	/** Construct a server for a protocol implementation instance listening on a
	* port and address. */
	public Server getServer(Class<?> protocol,
	Object instance, String bindAddress, int port,
	int numHandlers, int numReaders, int queueSizePerHandler,
	boolean verbose, Configuration conf,
	SecretManager<? extends TokenIdentifier> secretManager)
	throws IOException {
	return new Server(instance, conf, bindAddress, port, numHandlers,
	numReaders, queueSizePerHandler, verbose, secretManager);
	}

	/** An RPC Server. */
	public static class Server extends RPC.Server {
	private Object instance;
	private boolean verbose;

	/** Construct an RPC server.
	* @param instance the instance whose methods will be called
	* @param conf the configuration to use
	* @param bindAddress the address to bind on to listen for connection
	* @param port the port to listen for connections on
	*/
	public Server(Object instance, Configuration conf, String bindAddress, int port)
	throws IOException {
	this(instance, conf, bindAddress, port, 1, -1, -1, false, null);
	}

	private static String classNameBase(String className) {
	String[] names = className.split("\\.", -1);
	if (names == null \|\| names.length == 0) {
	return className;
	}
	return names[names.length-1];
	}

	/** Construct an RPC server.
	* @param instance the instance whose methods will be called
	* @param conf the configuration to use
	* @param bindAddress the address to bind on to listen for connection
	* @param port the port to listen for connections on
	* @param numHandlers the number of method handler threads to run
	* @param verbose whether each call should be logged
	*/
	public Server(Object instance, Configuration conf, String bindAddress, int port,
	int numHandlers, int numReaders, int queueSizePerHandler, boolean verbose,
	SecretManager<? extends TokenIdentifier> secretManager)
	throws IOException {
	super(bindAddress, port, Invocation.class, numHandlers, numReaders,
	queueSizePerHandler, conf,
	classNameBase(instance.getClass().getName()), secretManager);
	this.instance = instance;
	this.verbose = verbose;
	}

	public Writable call(Class<?> protocol, Writable param, long receivedTime)
	throws IOException {
	try {
	Invocation call = (Invocation)param;
	if (verbose) log("Call: " + call);

	Method method = protocol.getMethod(call.getMethodName(),
	call.getParameterClasses());
	method.setAccessible(true);

	// Verify rpc version
	if (call.getRpcVersion() != writableRpcVersion) {
	// Client is using a different version of WritableRpc
	throw new IOException(
	"WritableRpc version mismatch, client side version="
	+ call.getRpcVersion() + ", server side version="
	+ writableRpcVersion);
	}

	//Verify protocol version.
	//Bypass the version check for VersionedProtocol
	if (!method.getDeclaringClass().equals(VersionedProtocol.class)) {
	long clientVersion = call.getProtocolVersion();
	ProtocolSignature serverInfo = ((VersionedProtocol) instance)
	.getProtocolSignature(protocol.getCanonicalName(), call
	.getProtocolVersion(), call.getClientMethodsHash());
	long serverVersion = serverInfo.getVersion();
	if (serverVersion != clientVersion) {
	LOG.warn("Version mismatch: client version=" + clientVersion
	+ ", server version=" + serverVersion);
	throw new RPC.VersionMismatch(protocol.getName(), clientVersion,
	serverVersion);
	}
	}

	long startTime = System.currentTimeMillis();
	Object value = method.invoke(instance, call.getParameters());
	int processingTime = (int) (System.currentTimeMillis() - startTime);
	int qTime = (int) (startTime-receivedTime);
	if (LOG.isDebugEnabled()) {
	LOG.debug("Served: " + call.getMethodName() +
	" queueTime= " + qTime +
	" procesingTime= " + processingTime);
	}
	rpcMetrics.addRpcQueueTime(qTime);
	rpcMetrics.addRpcProcessingTime(processingTime);
	rpcDetailedMetrics.addProcessingTime(call.getMethodName(),
	processingTime);
	if (verbose) log("Return: "+value);

	return new ObjectWritable(method.getReturnType(), value);

	} catch (InvocationTargetException e) {
	Throwable target = e.getTargetException();
	if (target instanceof IOException) {
	throw (IOException)target;
	} else {
	IOException ioe = new IOException(target.toString());
	ioe.setStackTrace(target.getStackTrace());
	throw ioe;
	}
	} catch (Throwable e) {
	if (!(e instanceof IOException)) {
	LOG.error("Unexpected throwable object ", e);
	}
	IOException ioe = new IOException(e.toString());
	ioe.setStackTrace(e.getStackTrace());
	throw ioe;
	}
	}
	}

	private static void log(String value) {
	if (value!= null && value.length() > 55)
	value = value.substring(0, 55)+"...";
	LOG.info(value);
	}
	}