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();

  }