private JobSubmissionProtocol createRPCProxy(InetSocketAddress addr,
      Configuration conf) throws IOException {
    return (JobSubmissionProtocol) RPC.getProxy(JobSubmissionProtocol.class,
        JobSubmissionProtocol.versionID, addr, getUGI(conf), conf,
        NetUtils.getSocketFactory(conf, JobSubmissionProtocol.class));
  }
  public static VersionedProtocol getProxy(Class<?> protocol,
      long clientVersion, InetSocketAddress addr, UserGroupInformation ticket,
      Configuration conf, SocketFactory factory) throws IOException {    

    VersionedProtocol proxy =
        (VersionedProtocol) Proxy.newProxyInstance(
            protocol.getClassLoader(), new Class[] { protocol },
            new Invoker(addr, ticket, conf, factory));
    long serverVersion = proxy.getProtocolVersion(protocol.getName(), 
                                                  clientVersion);
    if (serverVersion == clientVersion) {
      return proxy;
    } else {
      throw new VersionMismatch(protocol.getName(), clientVersion, 
                                serverVersion);
    }
}

public Object invoke(Object proxy, Method method, Object[] args)
      throws Throwable {
         ObjectWritable value = (ObjectWritable)
        client.call(new Invocation(method, args), address, 
                    method.getDeclaringClass(), ticket);
      return value.get();}

    public Writable call(Writable param, InetSocketAddress addr,
            Class<?> protocol, UserGroupInformation ticket)
            {
        Call call = new Call(param);
        Connection connection = getConnection(addr, protocol, ticket, call);
        connection.sendParam(call); // send the parameter
        return call.value;
    }

public static Server getServer(final Object instance, final String bindAddress, final int port,
                                     final int numHandlers,
                                     final boolean verbose, Configuration conf) 
        throws IOException {
        return new Server(instance, conf, bindAddress, port, numHandlers, verbose);
    }

protected Server(String bindAddress, int port, 
                  Class<? extends Writable> paramClass, int handlerCount, 
                  Configuration conf, String serverName) 
    throws IOException {
    this.bindAddress = bindAddress;
    this.conf = conf;
    this.port = port;
    this.paramClass = paramClass;
    this.handlerCount = handlerCount;
    this.socketSendBufferSize = 0;
    this.maxQueueSize = handlerCount * MAX_QUEUE_SIZE_PER_HANDLER;
    this.callQueue  = new LinkedBlockingQueue<Call>(maxQueueSize); 
    this.maxIdleTime = 2*conf.getInt("ipc.client.connection.maxidletime", 1000);
    this.maxConnectionsToNuke = conf.getInt("ipc.client.kill.max", 10);
    this.thresholdIdleConnections = conf.getInt("ipc.client.idlethreshold", 4000);
    
    // Start the listener here and let it bind to the port
    listener = new Listener();
    this.port = listener.getAddress().getPort();    
    this.rpcMetrics = new RpcMetrics(serverName,
                          Integer.toString(this.port), this);
    this.tcpNoDelay = conf.getBoolean("ipc.server.tcpnodelay", false);


    // Create the responder here
    responder = new Responder();
  }

 private BlockingQueue<Call> callQueue; // queued calls
  private List<Connection> connectionList =
    Collections.synchronizedList(new LinkedList<Connection>());
  private Listener listener = null;
  private Responder responder = null;
  private Handler[] handlers = null;

public synchronized void start() throws IOException {
    responder.start();
    listener.start();
    handlers = new Handler[handlerCount];
    
    for (int i = 0; i < handlerCount; i++) {
      handlers[i] = new Handler(i);
      handlers[i].start();
    }
}

public Listener() throws IOException {
          address = new InetSocketAddress(bindAddress, port);
          // Create a new server socket and set to non blocking mode
          acceptChannel = ServerSocketChannel.open();
          acceptChannel.configureBlocking(false);

          // Bind the server socket to the local host and port
          bind(acceptChannel.socket(), address, backlogLength);
          port = acceptChannel.socket().getLocalPort(); //Could be an ephemeral port
          // create a selector;
          selector= Selector.open();

          // Register accepts on the server socket with the selector.
          acceptChannel.register(selector, SelectionKey.OP_ACCEPT);
}

public void run() {
          SERVER.set(Server.this);
          while (running) {
            SelectionKey key = null;
                     selector.select();
              Iterator<SelectionKey> iter = selector.selectedKeys().iterator();
              while (iter.hasNext()) {
                key = iter.next();
                iter.remove();
                try {
                  if (key.isValid()) {
                    if (key.isAcceptable())
                      doAccept(key);
                    else if (key.isReadable())
                      doRead(key);
                  }
                  
                }

void doAccept(SelectionKey key)
    {
    Connection c = null;
    ServerSocketChannel server = (ServerSocketChannel) key.channel();
    // accept up to 10 connections
    for (int i=0; i<10; i++) {
      SocketChannel channel = server.accept();
      SelectionKey readKey = channel.register(selector, SelectionKey.OP_READ);
      c = new Connection(readKey, channel, System.currentTimeMillis());
      readKey.attach(c);
      synchronized (connectionList) {
        connectionList.add(numConnections, c);
        numConnections++;
      }
      }

  void doRead(SelectionKey key) 
  {  
    Connection c = (Connection)key.attachment();
    c.setLastContact(System.currentTimeMillis());
    count = c.readAndProcess();  
  }