Http client pooling improvment

# Motivation

These changes drastically improve the performance of the HttpClient when keep-alive is turned on. Keep alive is supposed to improve performance where there are many repeated requests to the same host, and especially when the connection setup time or latency is high, as in the case of TLS (SSL) or slower/lossy network connections. However, in the current implementation the performance with keep alive turned on is abysmal, usually much lower than keep-alive turned off, which is sort of ironic given it's supposed to improve speed.  

The biggest issue I found is that when keep-alive is turned on, many requests are all piled onto the same connection, having to wait for each other to complete fully before the next request could complete, *even when the Max Pool Size was far from being reached*. In many cases, performance was much better when keep alive simply was not used. With the changes in this pull request, keep-alive on has at least the same performance as keep-alive off and significantly better as the number of short requests to the same host or latency goes up. 

A summary of the fix is that instead of picking the first connection in the pool (which usually already has several pending requests on it), we look for a completely unoccupied connection to use. If none is found, and the max pool limit has not been reached, we add a new connection to the pool. If the number of connections is already at the limit, we prefer a connection that has the least number of outstanding requests.  This way, not all the requests pile up on a single (or very few) connection(s) and get backed up waiting for previous requests to complete. 

Instead of hacking up the ConnectionPool class for this fix, I added a selectConnection() which defaults to the original behavior. Then with a subclass specific to ClientConnection and DefaultHttpClient, override this method to provide the behavior we want. 

I have included a test case below you can use to test various scenarios, and the results I got using several different latency scenarios. I encourage you to review the tests and give it a try yourself. 

# Other Notes
### Confusing Result
I added a method on the DefaultHttpClientConnectionPool called setDisableUsingOccupiedConnections(). The idea was that if you disallow connections from being shared at all, requests would go into the wait list and be serviced by the first connection to finish serving it's current request. Surprisingly (to me at least) setting this option produces atrocious results, and I couldn't really figure out why. I left this method in there for you to look at if you like, but I don't expect it to be part of the final code you accept. If it's obvious to you why this produces such bad results, I'd be curious to know, but the results without this option are quite good so I don't know if we need to try to optimize any more at the moment.

### Boss Threads
One interesting thing I observed and would like guidance on, is that the number of boss threads seems to have no effect what-so-ever on the performance tests I was doing. Setting it to 1 or 100 produced the exact same numbers. This seems counter intuitive, but maybe I'm not understanding what these threads do. Any ideas?

### Simulating Latency on a Mac
On a Mac you can simulate various types of network conditions with Network LInk Conditioner (see http://stackoverflow.com/questions/9659382/installing-apples-network-link-conditioner-tool) or you can use the very flexible but less user friendly ipfw command. 



# Test Results

## Test 1
boss_threads =5, max_pool_size = 25, requests per client = 500, request_process_ms=100;
### Results Before Modifications: 
#### Localhost ~0ms Latency
Keep Alive Off, Warm AVG:1257 TOTAL TIME: 2403 REQUEST/SEC:208
Keep Alive On, Warm AVG:2043 TOTAL TIME: 4076 REQUEST/SEC:122
#### Simulated Latency 2ms
Keep Alive Off, Warm AVG:2545 TOTAL TIME: 4356 REQUEST/SEC:114
Keep Alive On, Warm AVG:2074 TOTAL TIME: 4102 REQUEST/SEC:121
#### Simulated Latency 90ms
Keep Alive Off, Warm AVG:9496 TOTAL TIME: 18135 REQUEST/SEC:27
Keep Alive On, Warm AVG:2538 TOTAL TIME: 4666 REQUEST/SEC:107


### After fixes…
#### Localhost ~0ms Latency
Keep Alive Off AVG:11830 TOTAL TIME: 24111 REQUEST/SEC:208
Keep Alive On AVG:11810 TOTAL TIME: 24111 REQUEST/SEC:208
#### Simulated Latency 2ms
Keep Alive Off, Warm AVG:2636 TOTAL TIME: 4459 REQUEST/SEC:112
Keep Alive On, Warm AVG:2008 TOTAL TIME: 3986 REQUEST/SEC:125
#### Simulated Latency 90ms
Keep Alive Off, Warm AVG:9486 TOTAL TIME: 18105 REQUEST/SEC:27
Keep Alive On, Warm AVG:2442 TOTAL TIME: 4719 REQUEST/SEC:105

## Test 2
boss_threads =5, max_pool_size = 25, requests per client = 500, request_process_ms=200;
### Results Before Modifications: 
#### Localhost ~0ms Latency
Keep Alive Off, Warm AVG:2301 TOTAL TIME: 4397 REQUEST/SEC:113
Keep Alive On, Warm AVG:3763 TOTAL TIME: 7476 REQUEST/SEC:66
#### Simulated Latency 2ms
Keep Alive Off, Warm AVG:3310 TOTAL TIME: 5991 REQUEST/SEC:83
Keep Alive On, Warm AVG:3779 TOTAL TIME: 7485 REQUEST/SEC:66
#### Simulated Latency 90ms
Keep Alive Off, Warm AVG:12118 TOTAL TIME: 22032 REQUEST/SEC:22
Keep Alive On, Warm AVG:4079 TOTAL TIME: 7684 REQUEST/SEC:65

### After fixes…
#### Localhost ~0ms Latency
Keep Alive Off, Warm AVG:2299 TOTAL TIME: 4394 REQUEST/SEC:113
Keep Alive On, Warm AVG:2288 TOTAL TIME: 4399 REQUEST/SEC:113
#### Simulated Latency 2ms
Keep Alive Off, Warm AVG:3385 TOTAL TIME: 6042 REQUEST/SEC:82
Keep Alive On, Warm AVG:3643 TOTAL TIME: 7285 REQUEST/SEC:68
#### Simulated Latency 90ms
Keep Alive Off, Warm AVG:10529 TOTAL TIME: 20134 REQUEST/SEC:24
Keep Alive On, Warm AVG:4074 TOTAL TIME: 7683 REQUEST/SEC:65


## Test 3
boss_threads =5, max_pool_size = 25, requests per client = 500, request_process_ms=500;;
### Results Before Modifications: 
#### Localhost ~0ms Latency
Keep Alive Off, Warm AVG:5448 TOTAL TIME: 10394 REQUEST/SEC:48
Keep Alive On, Warm AVG:8913 TOTAL TIME: 17676 REQUEST/SEC:28


### After fixes…
#### Localhost ~0ms Latency
Keep Alive Off, Warm AVG:5447 TOTAL TIME: 10393 REQUEST/SEC:48
Keep Alive On, Warm AVG:5437 TOTAL TIME: 10400 REQUEST/SEC:48

# Test Case:

````java
package xxx;

import org.vertx.java.core.Handler;
import org.vertx.java.core.Vertx;
import org.vertx.java.core.http.HttpClient;
import org.vertx.java.core.http.HttpClientResponse;
import org.vertx.java.core.http.HttpServer;
import org.vertx.java.core.http.HttpServerRequest;
import org.vertx.java.core.http.impl.DefaultHttpClient;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;


/**
 * Tests the HttpClient performance. 
 *
 */
final class VertxPerformance {


    private static final int PORT = 8090;
    private static final String HOST = "localhost";
    private static final int BOSS_THREADS = 5;
    private static final int MAX_POOL_SIZE = 25;
    private static final boolean SSL = false;

    private final long MAX_WORK_TIME_MS = 100;
    private final long REQUESTS_PER_CLIENT = 500;

    public void run() throws Exception {
        final Vertx vertx = Vertx.newVertx();


        if(HOST.equals("localhost"))  {
            final HttpServer server = vertx.createHttpServer();
            server.requestHandler(new Handler<HttpServerRequest>() {
                @Override
                public void handle(final HttpServerRequest request) {
                    // Simulate a request that takes some time to do work.
                    vertx.setTimer((long) (MAX_WORK_TIME_MS), new Handler<Long>() {
                        @Override
                        public void handle(final Long event) {
                            request.response.end("This is a test result", "UTF-8");
                        }
                    });
                }
            }).listen(PORT);
        }


        final HttpClient client1 = vertx.createHttpClient().setHost(HOST).setPort(PORT).
                setBossThreads(BOSS_THREADS).setMaxPoolSize(MAX_POOL_SIZE).
                setSSL(SSL).setTrustAll(true).
                setKeepAlive(false);
        client1.exceptionHandler(new Handler<Exception>() {
            @Override
            public void handle(final Exception event) {
                System.err.println("Error on client1:" + event.getMessage());
            }
        });

        final HttpClient client2 = vertx.createHttpClient().setHost(HOST).setPort(PORT).
                setBossThreads(BOSS_THREADS).setMaxPoolSize(MAX_POOL_SIZE).
                setSSL(SSL).setTrustAll(true).
                setKeepAlive(true);
        // If this is disabled, performance plummets, not sure why, I would have expected the opposite.
//       ((DefaultHttpClient)client2).setDisableUsingOccupiedConnections(true);
        client2.exceptionHandler(new Handler<Exception>() {
            @Override
            public void handle(final Exception event) {
                System.err.println("Error on client2:" + event.getMessage());
            }
        });

        final class ResultCollector {
            private final String name;
            private final List<Long> result = new ArrayList<>();
            private final long createdTime = System.currentTimeMillis();
            private final long expectedIterations;
            private final CountDownLatch cd;


            ResultCollector(final String name, final CountDownLatch cd) {
                this.name = name;
                this.cd  = cd;
                this.expectedIterations = REQUESTS_PER_CLIENT;
            }

            ResultCollector(final String name, final CountDownLatch cd, final long expectedIterations) {
                this.name = name;
                this.cd  = cd;
                this.expectedIterations = expectedIterations;
            }

            int getCount() {
                return result.size();
            }

            Handler<HttpClientResponse> createHandler() {
                final long startTime = System.currentTimeMillis();
                return new Handler<HttpClientResponse>() {
                    @Override
                    public void handle(final HttpClientResponse event) {
                        result.add(System.currentTimeMillis() - startTime);
                        if (result.size() >= expectedIterations) report();
                    }
                };
            }

            private void report() {
                final long totalTime = System.currentTimeMillis() - createdTime;
                long total = 0;
                for (int i = 0; i < result.size(); i++) {
                    total += result.get(i);
                }
                System.out.println(name + " AVG:" + (total / result.size()) + " TOTAL TIME: " + totalTime + " REQUEST/SEC:" + (int)(result.size() / ((double)totalTime / 1000D)));
                cd.countDown();
            }
        }


        // Cold Run
        System.out.println("WARMING UP...");
        final CountDownLatch cd1 = new CountDownLatch(30);
        final Handler<HttpClientResponse> dummyHandler = new Handler<HttpClientResponse>() {
            public void handle(final HttpClientResponse event) {
                cd1.countDown();
            }
        };
        for (int i = 0; i < 15; i++) {
            client2.get("/", dummyHandler).end();
            client1.get("/", dummyHandler).end();
        }
        cd1.await();


        System.out.println("STARTING REAL TESTS....");
        final CountDownLatch cd1_2 = new CountDownLatch(1);
        final ResultCollector rc1_2 = new ResultCollector("Keep Alive Off, Warm", cd1_2);
        for (int i = 0; i < REQUESTS_PER_CLIENT; i++) {
            client1.get("/", rc1_2.createHandler()).end();
        }
        System.out.println("WAITING FOR PART 1 TO COMPLETE");
        cd1_2.await();
        final CountDownLatch cd2_2 = new CountDownLatch(1);
        final ResultCollector rc2_2 = new ResultCollector("Keep Alive On, Warm", cd2_2);
        for (int i = 0; i < REQUESTS_PER_CLIENT; i++) {
            client2.get("/", rc2_2.createHandler()).end();
        }
        System.out.println("WAITING FOR PART 2 TO COMPLETE");
        cd2_2.await();
    }

    public static void main(String[] args) {
        final VertxPerformance p = new VertxPerformance();

//        System.out.println("PRESS KEY TO RUN.....");
        try {
//            System.in.read();
            p.run();
//            System.in.read();
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("DONE");

    }

}


````