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

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufInputStream;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.ReferenceCountUtil;

import java.io.DataInputStream;
import java.io.IOException;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.net.SocketFactory;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableUtils;
import org.apache.hadoop.security.SaslRpcServer.AuthMethod;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hama.util.BSPNetUtils;

/**
 * A client for an IPC service using netty. 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.
 * 
 * @see AsyncClient
 */
public class AsyncClient {
  private static final String IPC_CLIENT_CONNECT_MAX_RETRIES_KEY = "ipc.client.connect.max.retries";
  private static final int IPC_CLIENT_CONNECT_MAX_RETRIES_DEFAULT = 10;
  private static final Log LOG = LogFactory.getLog(AsyncClient.class);
  private Hashtable<ConnectionId, Connection> connections = new Hashtable<ConnectionId, Connection>();

  private Class<? extends Writable> valueClass; // class of call values
  private int counter = 0; // counter for call ids
  private AtomicBoolean running = new AtomicBoolean(true); // if client runs
  final private Configuration conf; // configuration obj

  private SocketFactory socketFactory; // only use in order to meet the
                                       // consistency with other clients
  private int refCount = 1;

  final private static String PING_INTERVAL_NAME = "ipc.ping.interval";
  final static int DEFAULT_PING_INTERVAL = 60000; // 1 min

  /**
   * set the ping interval value in configuration
   * 
   * @param conf Configuration
   * @param pingInterval the ping interval
   */
  final public static void setPingInterval(Configuration conf, int pingInterval) {
    conf.setInt(PING_INTERVAL_NAME, pingInterval);
  }

  /**
   * Get the ping interval from configuration; If not set in the configuration,
   * return the default value.
   * 
   * @param conf Configuration
   * @return the ping interval
   */
  final static int getPingInterval(Configuration conf) {
    return conf.getInt(PING_INTERVAL_NAME, DEFAULT_PING_INTERVAL);
  }

  /**
   * The time after which a RPC will timeout. If ping is not enabled (via
   * ipc.client.ping), then the timeout value is the same as the pingInterval.
   * If ping is enabled, then there is no timeout value.
   * 
   * @param conf Configuration
   * @return the timeout period in milliseconds. -1 if no timeout value is set
   */
  final public static int getTimeout(Configuration conf) {
    if (!conf.getBoolean("ipc.client.ping", true)) {
      return getPingInterval(conf);
    }
    return -1;
  }

  /**
   * Increment this client's reference count
   * 
   */
  synchronized void incCount() {
    refCount++;
  }

  /**
   * Decrement this client's reference count
   * 
   */
  synchronized void decCount() {
    refCount--;
  }

  /**
   * Return if this client has no reference
   * 
   * @return true if this client has no reference; false otherwise
   */
  synchronized boolean isZeroReference() {
    return refCount == 0;
  }

  /**
   * Thread that reads responses and notifies callers. Each connection owns a
   * socket connected to a remote address. Calls are multiplexed through this
   * socket: responses may be delivered out of order.
   */
  private class Connection {
    private InetSocketAddress serverAddress; // server ip:port
    private ConnectionHeader header; // connection header
    private final ConnectionId remoteId; // connection id
    private AuthMethod authMethod; // authentication method

    private EventLoopGroup group;
    private Bootstrap bootstrap;
    private Channel channel;
    private int rpcTimeout;
    private int maxIdleTime; // connections will be culled if it was idle

    private final RetryPolicy connectionRetryPolicy;
    private boolean tcpNoDelay; // if T then disable Nagle's Algorithm
    private int pingInterval; // how often sends ping to the server in msecs

    // currently active calls
    private Hashtable<Integer, Call> calls = new Hashtable<Integer, Call>();
    private AtomicBoolean shouldCloseConnection = new AtomicBoolean(); // indicate
    private IOException closeException; // if the connection is closed, close
                                        // reason

    /**
     * Setup Connection Configuration
     * 
     * @param remoteId remote connection Id
     * @throws IOException
     */
    public Connection(ConnectionId remoteId) throws IOException {
      group = new NioEventLoopGroup();
      bootstrap = new Bootstrap();
      this.remoteId = remoteId;
      this.serverAddress = remoteId.getAddress();
      if (serverAddress.isUnresolved()) {
        throw new UnknownHostException("unknown host: "
            + remoteId.getAddress().getHostName());
      }
      this.maxIdleTime = remoteId.getMaxIdleTime();
      this.connectionRetryPolicy = remoteId.connectionRetryPolicy;
      this.tcpNoDelay = remoteId.getTcpNoDelay();
      this.pingInterval = remoteId.getPingInterval();
      if (LOG.isDebugEnabled()) {
        LOG.debug("The ping interval is" + this.pingInterval + "ms.");
      }
      this.rpcTimeout = remoteId.getRpcTimeout();
      Class<?> protocol = remoteId.getProtocol();

      authMethod = AuthMethod.SIMPLE;
      header = new ConnectionHeader(protocol == null ? null
          : protocol.getName(), null, authMethod);
    }

    /**
     * Add a call to this connection's call queue and notify a listener;
     * synchronized. Returns false if called during shutdown.
     * 
     * @param call to add
     * @return true if the call was added.
     */
    private synchronized boolean addCall(Call call) {
      if (shouldCloseConnection.get())
        return false;
      calls.put(call.id, call);
      notify();
      return true;
    }

    /**
     * Update the server address if the address corresponding to the host name
     * has changed.
     */
    private synchronized boolean updateAddress() throws IOException {
      // Do a fresh lookup with the old host name.
      InetSocketAddress currentAddr = BSPNetUtils.makeSocketAddr(
          serverAddress.getHostName(), serverAddress.getPort());

      if (!serverAddress.equals(currentAddr)) {
        LOG.warn("Address change detected. Old: " + serverAddress.toString()
            + " New: " + currentAddr.toString());
        serverAddress = currentAddr;
        return true;
      }
      return false;
    }

    /**
     * Connect to the server and set up the I/O streams. It then sends a header
     * to the server.
     */
    private void setupIOstreams() throws InterruptedException {
      if (channel != null && channel.isActive()) {
        return;
      }
      try {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Connecting to " + serverAddress);
        }

        setupConnection();
        writeHeader();
      } catch (Throwable t) {
        if (t instanceof IOException) {
          markClosed((IOException) t);
        } else {
          markClosed(new IOException("Couldn't set up IO streams", t));
        }
        close();
      }
    }

    /**
     * Configure the client and connect to server
     */
    private void setupConnection() throws Exception {
      while (true) {
        short ioFailures = 0;
        try {
          // rpcTimeout overwrites pingInterval
          if (rpcTimeout > 0) {
            pingInterval = rpcTimeout;
          }

          // Configure the client.
          // NioEventLoopGroup is a multithreaded event loop that handles I/O
          // operation
          group = new NioEventLoopGroup();
          // Bootstrap is a helper class that sets up a client
          bootstrap = new Bootstrap();
          bootstrap.group(group).channel(NioSocketChannel.class)
              .option(ChannelOption.TCP_NODELAY, this.tcpNoDelay)
              .option(ChannelOption.SO_KEEPALIVE, true)
              .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, pingInterval)
              .option(ChannelOption.SO_SNDBUF, 30 * 1024 * 1024)
              .handler(new LoggingHandler(LogLevel.INFO))
              .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                  ChannelPipeline p = ch.pipeline();
                  p.addLast(new IdleStateHandler(0, 0, maxIdleTime));
                  // Register message processing handler
                  p.addLast(new NioClientInboundHandler());
                }
              });

          // Bind and start to accept incoming connections.
          ChannelFuture channelFuture = bootstrap.connect(
              serverAddress.getAddress(), serverAddress.getPort()).sync();
          // Get io channel
          channel = channelFuture.channel();
          LOG.info("AsyncClient startup");
          break;
        } catch (Exception ie) {
          /*
           * Check for an address change and update the local reference. Reset
           * the failure counter if the address was changed
           */

          if (updateAddress()) {
            ioFailures = 0;
          }
          handleConnectionFailure(ioFailures++, ie);
        }
      }
    }

    /**
     * Write the header protocol header for each connection Out is not
     * synchronized because only the first thread does this.
     * 
     * @param channel
     */
    private void writeHeader() {
      DataOutputBuffer rpcBuff = null;
      DataOutputBuffer headerBuf = null;
      try {
        ByteBuf buf = channel.alloc().buffer();
        rpcBuff = new DataOutputBuffer();
        authMethod.write(rpcBuff);

        headerBuf = new DataOutputBuffer();
        header.write(headerBuf);
        byte[] data = headerBuf.getData();
        int dataLength = headerBuf.getLength();
        // write rpcheader
        buf.writeInt(AsyncServer.HEADER_LENGTH + dataLength);
        buf.writeBytes(AsyncServer.HEADER.array());
        buf.writeByte(AsyncServer.CURRENT_VERSION);
        buf.writeByte(rpcBuff.getData()[0]);
        // write header
        buf.writeInt(dataLength);
        buf.writeBytes(data, 0, dataLength);

        channel.writeAndFlush(buf);
      } catch (Exception e) {
        LOG.error("Couldn't send header" + e);
      } finally {
        IOUtils.closeStream(rpcBuff);
        IOUtils.closeStream(headerBuf);
      }
    }

    /**
     * close the current connection gracefully.
     */
    private void closeConnection() {
      try {
        if (!this.group.isTerminated()) {
          this.group.shutdownGracefully();
          LOG.info("client gracefully shutdown");
        }
      } catch (Exception e) {
        LOG.warn("Not able to close a client", e);
      }
    }

    /**
     * This class process received response message from server.
     */
    private class NioClientInboundHandler extends ChannelInboundHandlerAdapter {

      /**
       * Receive a response. This method is called with the received response
       * message, whenever new data is received from a server.
       * 
       * @param ctx
       * @param cause
       */
      @Override
      public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf byteBuf = (ByteBuf) msg;
        ByteBufInputStream byteBufInputStream = new ByteBufInputStream(byteBuf);
        DataInputStream in = new DataInputStream(byteBufInputStream);
        while (true) {
          try {
            if (in.available() <= 0)
              break;
            // try to read an id
            int id = in.readInt();

            if (LOG.isDebugEnabled())
              LOG.debug(serverAddress.getHostName() + " got value #" + id);

            Call call = calls.get(id);

            // read call status
            int state = in.readInt();
            if (state == Status.SUCCESS.state) {
              Writable value = ReflectionUtils.newInstance(valueClass, conf);
              value.readFields(in); // read value
              call.setValue(value);
              calls.remove(id);
            } else if (state == Status.ERROR.state) {
              String className = WritableUtils.readString(in);
              byte[] errorBytes = new byte[in.available()];
              in.readFully(errorBytes);
              call.setException(new RemoteException(className, new String(
                  errorBytes)));
              calls.remove(id);
            } else if (state == Status.FATAL.state) {
              // Close the connection
              markClosed(new RemoteException(WritableUtils.readString(in),
                  WritableUtils.readString(in)));
            } else {
              byte[] garbageBytes = new byte[in.available()];
              in.readFully(garbageBytes);
            }
          } catch (IOException e) {
            markClosed(e);
          }
        }
        IOUtils.closeStream(in);
        IOUtils.closeStream(byteBufInputStream);
        ReferenceCountUtil.release(msg);
      }

      /**
       * Ths event handler method is called with a Throwable due to an I/O
       * error. Then, exception is logged and its associated channel is closed
       * here
       * 
       * @param ctx
       * @param cause
       */
      @Override
      public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        LOG.error("Occured I/O Error : " + cause.getMessage());
        ctx.close();
      }

      /**
       * this method is triggered after a long reading/writing/idle time, it is
       * marked as to be closed, or the client is marked as not running.
       * 
       * @param ctx
       * @param evt
       */
      @Override
      public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof IdleStateEvent) {
          IdleStateEvent e = (IdleStateEvent) evt;
          if (e.state() != IdleState.ALL_IDLE) {
            if (!calls.isEmpty() && !shouldCloseConnection.get()
                && running.get()) {
              return;
            } else if (shouldCloseConnection.get()) {
              markClosed(null);
            } else if (calls.isEmpty()) { // idle connection closed or stopped
              markClosed(null);
            } else { // get stopped but there are still pending requests
              markClosed((IOException) new IOException()
                  .initCause(new InterruptedException()));
            }
            closeConnection();
          }
        }

      }
    }

    /**
     * Handle connection failures If the current number of retries is equal to
     * the max number of retries, stop retrying and throw the exception;
     * Otherwise backoff 1 second and try connecting again. This Method is only
     * called from inside setupIOstreams(), which is synchronized. Hence the
     * sleep is synchronized; the locks will be retained.
     * 
     * @param curRetries current number of retries
     * @param maxRetries max number of retries allowed
     * @param ioe failure reason
     * @throws IOException if max number of retries is reached
     */
    @SuppressWarnings("unused")
    private void handleConnectionFailure(int curRetries, int maxRetries,
        IOException ioe) throws IOException {

      closeConnection();

      // throw the exception if the maximum number of retries is reached
      if (curRetries >= maxRetries) {
        throw ioe;
      }

      // otherwise back off and retry
      try {
        Thread.sleep(1000);
      } catch (InterruptedException ignored) {
      }

      LOG.info("Retrying connect to server: " + serverAddress
          + ". Already tried " + curRetries + " time(s); maxRetries="
          + maxRetries);
    }

    /*
     * Handle connection failures If the current number of retries, stop
     * retrying and throw the exception; Otherwise backoff 1 second and try
     * connecting again. This Method is only called from inside
     * setupIOstreams(), which is synchronized. Hence the sleep is synchronized;
     * the locks will be retained.
     * @param curRetries current number of retries
     * @param ioe failure reason
     * @throws Exception if max number of retries is reached
     */
    private void handleConnectionFailure(int curRetries, Exception ioe)
        throws Exception {
      closeConnection();

      final boolean retry;
      try {
        retry = connectionRetryPolicy.shouldRetry(ioe, curRetries);
      } catch (Exception e) {
        throw e instanceof IOException ? (IOException) e : new IOException(e);
      }
      if (!retry) {
        throw ioe;
      }

      LOG.info("Retrying connect to server: " + serverAddress
          + ". Already tried " + curRetries + " time(s); retry policy is "
          + connectionRetryPolicy);
    }

    /**
     * Return the remote address of server
     * 
     * @return remote server address
     */
    public InetSocketAddress getRemoteAddress() {
      return serverAddress;
    }

    /**
     * Initiates a call by sending the parameter to the remote server.
     * 
     * @param sendCall
     */
    public void sendParam(Call sendCall) {
      if (LOG.isDebugEnabled())
        LOG.debug(this.getClass().getName() + " sending #" + sendCall.id);
      DataOutputBuffer buff = null;
      try {
        buff = new DataOutputBuffer();
        buff.writeInt(sendCall.id);
        sendCall.param.write(buff);
        byte[] data = buff.getData();
        int dataLength = buff.getLength();
        ByteBuf buf = channel.alloc().buffer();

        buf.writeInt(dataLength);
        buf.writeBytes(data, 0, dataLength);
        ChannelFuture channelFuture = channel.writeAndFlush(buf);
        if (channelFuture.cause() != null) {
          throw channelFuture.cause();
        }
      } catch (IOException ioe) {
        markClosed(ioe);
      } catch (Throwable t) {
        markClosed(new IOException(t));
      } finally {
        // the buffer is just an in-memory buffer, but it is still
        // polite to close early
        IOUtils.closeStream(buff);
      }
    }

    /**
     * Mark the connection to be closed
     * 
     * @param ioe
     **/
    private synchronized void markClosed(IOException ioe) {
      if (shouldCloseConnection.compareAndSet(false, true)) {
        closeException = ioe;
        notifyAll();
      }
    }

    /** Close the connection. */
    private synchronized void close() {
      if (!shouldCloseConnection.get()) {
        LOG.error("The connection is not in the closed state");
        return;
      }

      // release the resources
      // first thing to do;take the connection out of the connection list
      synchronized (connections) {
        if (connections.get(remoteId) == this) {
          Connection connection = connections.remove(remoteId);
          connection.closeConnection();
        }
      }

      // clean up all calls
      if (closeException == null) {
        if (!calls.isEmpty()) {
          LOG.warn("A connection is closed for no cause and calls are not empty");

          // clean up calls anyway
          closeException = new IOException("Unexpected closed connection");
          cleanupCalls();
        }
      } else {
        // log the info
        if (LOG.isDebugEnabled()) {
          LOG.debug("closing ipc connection to " + serverAddress + ": "
              + closeException.getMessage(), closeException);
        }

        // cleanup calls
        cleanupCalls();
      }
      if (LOG.isDebugEnabled())
        LOG.debug(serverAddress.getHostName() + ": closed");
    }

    /** Cleanup all calls and mark them as done */
    private void cleanupCalls() {
      Iterator<Entry<Integer, Call>> itor = calls.entrySet().iterator();
      while (itor.hasNext()) {
        Call c = itor.next().getValue();
        c.setException(closeException); // local exception
        itor.remove();
      }
    }
  }

  /** A call waiting for a value. */
  private class Call {
    int id; // call id
    Writable param; // parameter
    Writable value; // value, null if error
    IOException error; // exception, null if value
    boolean done; // true when call is done

    protected Call(Writable param) {
      this.param = param;
      synchronized (AsyncClient.this) {
        this.id = counter++;
      }
    }

    /**
     * Indicate when the call is complete and the value or error are available.
     * Notifies by default.
     */
    protected synchronized void callComplete() {
      this.done = true;
      notify(); // notify caller
    }

    /**
     * Set the exception when there is an error. Notify the caller the call is
     * done.
     * 
     * @param error exception thrown by the call; either local or remote
     */
    public synchronized void setException(IOException error) {
      this.error = error;
      this.callComplete();
    }

    /**
     * Set the return value when there is no error. Notify the caller the call
     * is done.
     * 
     * @param value return value of the call.
     */
    public synchronized void setValue(Writable value) {
      this.value = value;
      callComplete();
    }
  }

  /** Call implementation used for parallel calls. */
  private class ParallelCall extends Call {
    private ParallelResults results;
    private int index;

    public ParallelCall(Writable param, ParallelResults results, int index) {
      super(param);
      this.results = results;
      this.index = index;
    }

    @Override
    /** Deliver result to result collector. */
    protected void callComplete() {
      results.callComplete(this);
    }
  }

  /** Result collector for parallel calls. */
  private static class ParallelResults {
    private Writable[] values;
    private int size;
    private int count;

    public ParallelResults(int size) {
      this.values = new Writable[size];
      this.size = size;
    }

    /**
     * Collect a result.
     * 
     * @param call
     */
    public synchronized void callComplete(ParallelCall call) {
      values[call.index] = call.value; // store the value
      count++; // count it
      if (count == size) // if all values are in
        notify(); // then notify waiting caller
    }
  }

  /**
   * Construct an IPC client whose values are of the given {@link Writable}
   * class.
   * 
   * @param valueClass
   * @param conf
   * @param factory
   */
  public AsyncClient(Class<? extends Writable> valueClass, Configuration conf,
      SocketFactory factory) {
    this.valueClass = valueClass;
    this.conf = conf;
    // SocketFactory only use in order to meet the consistency with other
    // clients
    this.socketFactory = factory;
  }

  /**
   * Construct an IPC client with the default SocketFactory
   * 
   * @param valueClass
   * @param conf
   */
  public AsyncClient(Class<? extends Writable> valueClass, Configuration conf) {
    // SocketFactory only use in order to meet the consistency with other
    // clients
    this(valueClass, conf, BSPNetUtils.getDefaultSocketFactory(conf));
  }

  /**
   * Return the socket factory of this client
   * 
   * @return this client's socket factory
   */
  SocketFactory getSocketFactory() {
    // SocketFactory only use in order to meet the consistency with other
    // clients
    return socketFactory;
  }

  /**
   * Stop all threads related to this client. No further calls may be made using
   * this client.
   */
  public void stop() {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Stopping client");
    }

    if (!running.compareAndSet(true, false)) {
      return;
    }

    // wake up all connections
    synchronized (connections) {
      for (Connection conn : connections.values()) {
        conn.closeConnection();
      }
    }
  }

  /**
   * Make a call, passing <code>param</code>, to the IPC server running at
   * <code>address</code> which is servicing the <code>protocol</code> protocol,
   * with the <code>ticket</code> credentials, <code>rpcTimeout</code> as
   * timeout and <code>conf</code> as configuration for this connection,
   * returning the value. Throws exceptions if there are network problems or if
   * the remote code threw an exception.
   * 
   * @param param
   * @param addr
   * @param protocol
   * @param ticket
   * @param rpcTimeout
   * @param conf
   * @return Response Writable value
   * @throws InterruptedException
   * @throws IOException
   */
  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(param, remoteId);
  }

  /**
   * Make a call, passing <code>param</code>, to the IPC server defined by
   * <code>remoteId</code>, returning the value. Throws exceptions if there are
   * network problems or if the remote code threw an exception.
   * 
   * @param param
   * @param remoteId
   * @return Response Writable value
   * @throws InterruptedException
   * @throws IOException
   */
  public Writable call(Writable param, ConnectionId remoteId)
      throws InterruptedException, IOException {
    Call call = new Call(param);

    Connection connection = getConnection(remoteId, call);

    connection.sendParam(call); // send the parameter
    boolean interrupted = false;

    synchronized (call) {
      int callFailCount = 0;
      while (!call.done) {
        try {
          call.wait(1000); // wait for the result
          // prevent client hang from response error
          if (callFailCount++ == IPC_CLIENT_CONNECT_MAX_RETRIES_DEFAULT)
            break;
        } catch (InterruptedException ie) {
          interrupted = true;
        }
      }

      if (interrupted) {
        // set the interrupt flag now that we are done waiting
        Thread.currentThread().interrupt();

      }

      if (call.error != null) {
        if (call.error instanceof RemoteException) {
          call.error.fillInStackTrace();
          throw call.error;
        } else { // local exception
          // use the connection because it will reflect an ip change,
          // unlike
          // the remoteId
          throw wrapException(connection.getRemoteAddress(), call.error);
        }
      } else {
        return call.value;
      }
    }
  }

  /**
   * Take an IOException and the address we were trying to connect to and return
   * an IOException with the input exception as the cause. The new exception
   * provides the stack trace of the place where the exception is thrown and
   * some extra diagnostics information. If the exception is ConnectException or
   * SocketTimeoutException, return a new one of the same type; Otherwise return
   * an IOException.
   * 
   * @param addr target address
   * @param exception the relevant exception
   * @return an exception to throw
   */
  private IOException wrapException(InetSocketAddress addr,
      IOException exception) {
    if (exception instanceof ConnectException) {
      // connection refused; include the host:port in the error
      return (ConnectException) new ConnectException("Call to " + addr
          + " failed on connection exception: " + exception)
          .initCause(exception);
    } else if (exception instanceof SocketTimeoutException) {
      return (SocketTimeoutException) new SocketTimeoutException("Call to "
          + addr + " failed on socket timeout exception: " + exception)
          .initCause(exception);
    } else {
      return (IOException) new IOException("Call to " + addr
          + " failed on local exception: " + exception).initCause(exception);

    }
  }

  /**
   * Makes a set of calls in parallel. Each parameter is sent to the
   * corresponding address. When all values are available, or have timed out or
   * errored, the collected results are returned in an array. The array contains
   * nulls for calls that timed out or errored.
   * 
   * @param params
   * @param addresses
   * @param protocol
   * @param ticket
   * @param conf
   * @return Response Writable value array
   * @throws IOException
   * @throws InterruptedException
   */
  public Writable[] call(Writable[] params, InetSocketAddress[] addresses,
      Class<?> protocol, UserGroupInformation ticket, Configuration conf)
      throws IOException, InterruptedException {
    if (addresses.length == 0)
      return new Writable[0];

    ParallelResults results = new ParallelResults(params.length);
    ConnectionId remoteId[] = new ConnectionId[addresses.length];
    synchronized (results) {
      for (int i = 0; i < params.length; i++) {
        ParallelCall call = new ParallelCall(params[i], results, i);
        try {
          remoteId[i] = ConnectionId.getConnectionId(addresses[i], protocol,
              ticket, 0, conf);
          Connection connection = getConnection(remoteId[i], call);
          connection.sendParam(call); // send each parameter
        } catch (IOException e) {
          // log errors
          LOG.info("Calling " + addresses[i] + " caught: " + e.getMessage(), e);
          results.size--; // wait for one fewer result
        }
      }

      while (results.count != results.size) {
        try {
          results.wait(); // wait for all results
        } catch (InterruptedException e) {
        }
      }

      return results.values;
    }
  }

  // for unit testing only
  Set<ConnectionId> getConnectionIds() {
    synchronized (connections) {
      return connections.keySet();
    }
  }

  /**
   * Get a connection from the pool, or create a new one and add it to the pool.
   * Connections to a given ConnectionId are reused.
   * 
   * @param remoteId
   * @param call
   * @return connection
   * @throws IOException
   * @throws InterruptedException
   */
  private synchronized Connection getConnection(ConnectionId remoteId, Call call)
      throws IOException, InterruptedException {
    if (!running.get()) {
      // the client is stopped
      throw new IOException("The client is stopped");
    }
    Connection connection;
    /*
     * we could avoid this allocation for each RPC by having a connectionsId
     * object and with set() method. We need to manage the refs for keys in
     * HashMap properly. For now its ok
     */
    do {
      connection = connections.get(remoteId);
      if (connection == null) {
        connection = new Connection(remoteId);
        connections.put(remoteId, connection);
      } else if (!connection.channel.isWritable()
          || !connection.channel.isActive()) {
        connection = new Connection(remoteId);
        connections.remove(remoteId);
        connections.put(remoteId, connection);
      }
    } while (!connection.addCall(call));
    // we don't invoke the method below inside "synchronized (connections)"
    // block above. The reason for that is if the server happens to be slow,
    // it will take longer to establish a connection and that will slow the
    // entire system down.

    connection.setupIOstreams();
    return connection;
  }

  /**
   * This class holds the address and the user ticket. The client connections to
   * servers are uniquely identified by <remoteAddress, protocol, ticket>
   */
  static class ConnectionId {
    InetSocketAddress address;
    UserGroupInformation ticket;
    Class<?> protocol;
    private static final int PRIME = 16777619;
    private int rpcTimeout;
    private String serverPrincipal;
    private int maxIdleTime; // connections will be culled if it was idle for
                             // maxIdleTime msecs
    private final RetryPolicy connectionRetryPolicy;
    private boolean tcpNoDelay; // if T then disable Nagle's Algorithm
    private int pingInterval; // how often sends ping to the server in msecs

    ConnectionId(InetSocketAddress address, Class<?> protocol,
        UserGroupInformation ticket, int rpcTimeout, String serverPrincipal,
        int maxIdleTime, RetryPolicy connectionRetryPolicy, boolean tcpNoDelay,
        int pingInterval) {
      this.protocol = protocol;
      this.address = address;
      this.ticket = ticket;
      this.rpcTimeout = rpcTimeout;
      this.serverPrincipal = serverPrincipal;
      this.maxIdleTime = maxIdleTime;
      this.connectionRetryPolicy = connectionRetryPolicy;
      this.tcpNoDelay = tcpNoDelay;
      this.pingInterval = pingInterval;
    }

    InetSocketAddress getAddress() {
      return address;
    }

    Class<?> getProtocol() {
      return protocol;
    }

    private int getRpcTimeout() {
      return rpcTimeout;
    }

    String getServerPrincipal() {
      return serverPrincipal;
    }

    int getMaxIdleTime() {
      return maxIdleTime;
    }

    boolean getTcpNoDelay() {
      return tcpNoDelay;
    }

    int getPingInterval() {
      return pingInterval;
    }

    static ConnectionId getConnectionId(InetSocketAddress addr,
        Class<?> protocol, UserGroupInformation ticket, Configuration conf)
        throws IOException {
      return getConnectionId(addr, protocol, ticket, 0, conf);
    }

    static ConnectionId getConnectionId(InetSocketAddress addr,
        Class<?> protocol, UserGroupInformation ticket, int rpcTimeout,
        Configuration conf) throws IOException {
      return getConnectionId(addr, protocol, ticket, rpcTimeout, null, conf);
    }

    static ConnectionId getConnectionId(InetSocketAddress addr,
        Class<?> protocol, UserGroupInformation ticket, int rpcTimeout,
        RetryPolicy connectionRetryPolicy, Configuration conf)
        throws IOException {

      if (connectionRetryPolicy == null) {
        final int max = conf.getInt(IPC_CLIENT_CONNECT_MAX_RETRIES_KEY,
            IPC_CLIENT_CONNECT_MAX_RETRIES_DEFAULT);
        connectionRetryPolicy = RetryPolicies
            .retryUpToMaximumCountWithFixedSleep(max, 1, TimeUnit.SECONDS);
      }

      return new ConnectionId(addr, protocol, ticket, rpcTimeout,
          null,
          conf.getInt("ipc.client.connection.maxidletime", 10000), // 10s
          connectionRetryPolicy,
          conf.getBoolean("ipc.client.tcpnodelay", true),
          AsyncClient.getPingInterval(conf));
    }

    static boolean isEqual(Object a, Object b) {
      return a == null ? b == null : a.equals(b);
    }

    @Override
    public boolean equals(Object obj) {
      if (obj == this) {
        return true;
      }
      if (obj instanceof ConnectionId) {
        ConnectionId that = (ConnectionId) obj;
        return isEqual(this.address, that.address)
            && this.maxIdleTime == that.maxIdleTime
            && isEqual(this.connectionRetryPolicy, that.connectionRetryPolicy)
            && this.pingInterval == that.pingInterval
            && isEqual(this.protocol, that.protocol)
            && this.rpcTimeout == that.rpcTimeout
            && isEqual(this.serverPrincipal, that.serverPrincipal)
            && this.tcpNoDelay == that.tcpNoDelay
            && isEqual(this.ticket, that.ticket);
      }
      return false;
    }

    @Override
    public int hashCode() {
      int result = connectionRetryPolicy.hashCode();
      result = PRIME * result + ((address == null) ? 0 : address.hashCode());
      result = PRIME * result + maxIdleTime;
      result = PRIME * result + pingInterval;
      result = PRIME * result + ((protocol == null) ? 0 : protocol.hashCode());
      result = PRIME * rpcTimeout;
      result = PRIME * result
          + ((serverPrincipal == null) ? 0 : serverPrincipal.hashCode());
      result = PRIME * result + (tcpNoDelay ? 1231 : 1237);
      result = PRIME * result + ((ticket == null) ? 0 : ticket.hashCode());
      return result;
    }
  }
}