2
\$\begingroup\$

I have made a connection pool for SQL Connection objects. The purpose of this object is to provide multi threaded support for the accessing of the underlying database.

The code works by using a reflection proxy for overriding the close() method of the returned connections, so it can detect when the consumer of the API is done processing, and therefore detect when the connection can be reused.

Connections ready for reuse are put in a Queue<>, this provides a round-robin access to the connections. If there are no usable connections remaining in this queue, then a call will be made to create a new connection from a ConnectionCreator interface.

If a exception occurs while scanning for a proper connection, then these exceptions will be put in a list, and if another exception occurs while creating the new connection, these exceptions will be added to the new exception using addSuppressed.

The API exposes the following methods to the outside:

  • close() - Closes the pool
  • openConnection() - Gets a pooled connection, or creates a new one if no connection exists
  • getAmountOfPooledConnections() - Get the amount of connections inside the pool
  • closeAllConnections() - Close all connections in the pool, but doesn't mark the object as closed.

My questions:

  1. Did I employ double-locking correctly in giveBackConnection(Connection). Under no situation there may be connections be stuck in queue after closure of the object.
  2. Did I follow correct practices for suppressing the exceptions, or should suppressed exceptions be used for something else?
  3. Anything else you notice about the code?

Code

ConnectionPool.java

package me.ferrybig.connectionpool;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.logging.Logger;

public class ConnectionPool implements AutoCloseable {

    private static final int CONNECTION_TIMEOUT = 5000;
    public static final int DEFAULT_POOL_SIZE = 16;

    private static final Logger LOG = Logger.getLogger(ConnectionPool.class.getName());

    private final Queue<Connection> queue = new ArrayBlockingQueue<>(5);
    private final ConnectionCreator creator;
    private volatile boolean isGlobalClosed = false;

    public ConnectionPool(ConnectionCreator creator) {
        this(creator, DEFAULT_POOL_SIZE);
    }

    public ConnectionPool(ConnectionCreator creator, int poolSize) {
        this.creator = Objects.requireNonNull(creator, "creator == null");
    }

    @Override
    public void close() throws SQLException {
        if (isGlobalClosed) {
            return;
        }
        isGlobalClosed = true;
        closeAllConnections();
    }

    /**
     * Get the connection to the database to access all information. This method
     * supports multiple threads without creating a new connectio for each
     * thread.
     *
     * @return Connection -> The connection to the database
     * @throws SQLException
     */
    public Connection openConnection() throws SQLException {
        if (isGlobalClosed) {
            throw new SQLException("Pool closed");
        }
        Connection con = getOrCreateConnection();
        LOG.finest("Opening pooled connection");
        return (Connection) Proxy.newProxyInstance(this.getClass().getClassLoader(),
                new Class<?>[]{Connection.class}, new InvocationHandler() {
            boolean isClosed = false;

            @Override
            public Object invoke(Object proxy, Method method, Object[] args)
                    throws Throwable {
                if (method.getName().equals("close") && method.getParameterCount() == 0) {
                    assert method.getReturnType() == Void.TYPE;
                    if (isClosed) {
                        return null;
                    }
                    isClosed = true;
                    giveBackConnection(con);
                    return null;
                }
                return method.invoke(con, args);
            }
        });
    }

    /**
     * Gets a pending connection, or a new one if none left
     *
     * @return Gets a pending connection, or a new one if none left
     * @throws SQLException If none valid left and the creation of a new one
     * failed
     */
    private Connection getOrCreateConnection() throws SQLException {
        List<SQLException> suppressed = new ArrayList<>();
        Connection con;
        while ((con = queue.poll()) != null) {
            try {
                if (!con.isClosed() && con.isValid(CONNECTION_TIMEOUT)) {
                    return con;
                } else {
                    con.close();
                }
            } catch (SQLException e) {
                suppressed.add(e);
            }
        }
        try {
            return createNewConnection();
        } catch (SQLException | RuntimeException e) {
            suppressed.forEach(e::addSuppressed);
            throw e;
        }
    }

    /**
     * Create a new connection
     *
     * @return The newly made connection
     * @throws SQLException If the creation of the <code>Connection</code>
     * failed
     */
    private Connection createNewConnection() throws SQLException {
        LOG.finer("Creating new connection");
        Connection c = creator.newConnection();
        if (c == null) {
            throw new SQLException("creator returned null connection");
        }
        return c;
    }

    /**
     * Puts a new connection back in the global pool
     *
     * @param orginal The connection to put back
     * @throws SQLException If it didn't fit into out pool and failed closing
     */
    private void giveBackConnection(Connection orginal) throws SQLException {
        LOG.finest("Closing pooled connection");
        if (isGlobalClosed) {
            List<SQLException> suppressed = closeRemainingConnections();
            try {
                LOG.fine("Closing connection");
                orginal.close();
            } catch (SQLException | RuntimeException e) {
                suppressed.forEach(e::addSuppressed);
                throw e;
            }
        }
        if (!queue.offer(orginal)) {
            LOG.fine("Closing connection");
            orginal.close();
        } else if (isGlobalClosed) { // variable MAY have changed
            closeAllConnections();
        }
    }

    /**
     *
     * @return
     */
    public int getAmountOfPooledConnections() {
        return queue.size();
    }

    public void closeAllConnections() throws SQLException {
        List<SQLException> suppressed = closeRemainingConnections();
        if (suppressed.isEmpty()) {
            return;
        }
        SQLException e = new SQLException("Unable to close all connections");
        suppressed.forEach(e::addSuppressed);
        throw e;
    }

    private List<SQLException> closeRemainingConnections() {
        LOG.finer("Closing remaining connection");
        List<SQLException> suppressed = new ArrayList<>();
        Connection c;
        while ((c = queue.poll()) != null) {
            try {
                c.close();
            } catch (SQLException e) {
                suppressed.add(e);
            }
        }
        return suppressed;
    }

    public interface ConnectionCreator {

        public Connection newConnection() throws SQLException;
    }
}

Test: ConnectionPoolTest.java

package me.ferrybig.connectionpool;

import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Assert;
import org.junit.Test;
import org.junit.Assume;

public class ConnectionPoolTest {

    @Test
    public void creatorNotCalledWhenNoConnectionIsNeededTest() throws Exception {
        AtomicBoolean b = new AtomicBoolean(false);
        ConnectionPool con = new ConnectionPool(() -> {
            b.set(true);
            Assert.fail();
            return null;
        });
        con.close();
        Assert.assertEquals(0, con.getAmountOfPooledConnections());
        Assert.assertFalse(b.get());
    }

    @Test(expected = RuntimeException.class)
    public void runtimeExceptionsCorrectlyPassedFromConnectionCreatorTest()
            throws Exception {
        ConnectionPool con = new ConnectionPool(() -> {
            throw new RuntimeException();
        });
        con.openConnection();
        Assert.fail();
    }

    @Test
    public void doesntCreateMoreConnectionsThanNeededTest() throws Exception {
        AtomicInteger created = new AtomicInteger(0);
        try (ConnectionPool con = new ConnectionPool(() -> {
            created.incrementAndGet();
            return (Connection) Proxy.newProxyInstance(this.getClass().getClassLoader(),
                    new Class<?>[]{Connection.class}, (Object proxy, Method method, Object[] args) -> {
                        if ("close".equals(method.getName())) {
                            return null;
                        }
                        if ("isClosed".equals(method.getName())) {
                            return false;
                        }
                        if ("isValid".equals(method.getName())) {
                            return true;
                        }
                        Assert.fail("Not implemented for this test");
                        return null;
                    });
        })) {
            Assume.assumeTrue("Test `creatorNotCalledWhenNoConnectionIsNeededTest` is broken", created.get() == 0);
            Assert.assertEquals(0, con.getAmountOfPooledConnections());

            con.openConnection().close();
            Assert.assertEquals(1, con.getAmountOfPooledConnections());

            con.openConnection().close();
            Assert.assertEquals(1, con.getAmountOfPooledConnections());

            Connection c1 = con.openConnection(),
                    c2 = con.openConnection(),
                    c3 = con.openConnection(),
                    c4 = con.openConnection();
            Assert.assertEquals(0, con.getAmountOfPooledConnections());
            Assert.assertEquals(4, created.get());

            c1.close();
            c2.close();
            c3.close();
            c4.close();
            Assert.assertEquals(4, con.getAmountOfPooledConnections());
            Assert.assertEquals(4, created.get());

            con.closeAllConnections();
            Assert.assertEquals(0, con.getAmountOfPooledConnections());
            Assert.assertEquals(4, created.get());

            con.openConnection().close();
            Assert.assertEquals(1, con.getAmountOfPooledConnections());
            Assert.assertEquals(5, created.get());
        }
    }

    @Test(expected = SQLException.class)
    public void cannotGetNewConnectionAfterClosureTest() throws SQLException {
        ConnectionPool con = new ConnectionPool(() -> {
            throw new RuntimeException();
        });
        con.close();
        con.openConnection();
    }
}
\$\endgroup\$
  • 1
    \$\begingroup\$ Is there a compelling reason not to use one of the existing libraries that does this? \$\endgroup\$ – Eric Stein Feb 14 '16 at 1:10
2
\$\begingroup\$
public ConnectionPool(ConnectionCreator creator, int poolSize) {
    this.creator = Objects.requireNonNull(creator, "creator == null");
}

You're not respecting the passed poolSize...
Neither did you implement a way to pass a differing CONNECTION_TIMEOUT that would allow me to timeout faster (or slower) than 5 seconds.

private volatile boolean isGlobalClosed = false;

I personally think this one is better off as an AtomicBoolean. Makes the behaviour a little more intuitive wrt. interleafing and related. That's a personal note though.

private final Queue<Connection> queue = new ArrayBlockingQueue<>(5);

You're not respecting the poolSize here either... Neither the DEFAULT_POOL_SIZE nor the one given in the constructor are anywhere near relevant. You just completely ignore them and keep at most 5 Connections available in your pool. This isn't the standard behaviour of a ...Pool class.

Instead I'd expect to be allowed to ask a Facade my requests. The facade then would delegate my request / query to the next available Connection. For that I'd use:

public class PooledConnection implements Connection {
    // ...  

You'd have to share this across threads manually, but alas, the same problem is present in the current code.

<interjection>
How other languages do it:
.NET has the following in their docs for SQLConnection:

Login and logout events will not be raised on the server when a connection is fetched from or returned to the connection pool, because the connection is not actually closed when it is returned to the connection pool. For more information, see SQL Server Connection Pooling (ADO.NET).

The interesting part here is that you don't request the Connection from a Pool, but it's inherently constructed from a Pool. Your implementation does not allow creating a connection like that.

The link given states:

Whenever a user calls Open on a connection, the pooler looks for an available connection in the pool. If a pooled connection is available, it returns it to the caller instead of opening a new connection. When the application calls Close on the connection, the pooler returns it to the pooled set of active connections instead of closing it. Once the connection is returned to the pool, it is ready to be reused on the next Open call.

</interjection>

private void giveBackConnection(Connection orginal) throws SQLException {

right... giveBackConnection is a rather clunky name IMO. I'd say reclaimConnection or even returnConnection would be more appropriate.

public interface ConnectionCreator {

    public Connection newConnection() throws SQLException;
}

This is called Supplier in Java 8.

\$\endgroup\$
  • \$\begingroup\$ Thx for your review. The point about the poolsize ignored and no ability to set a time are indeed good points and I will be adding this to the next revision of my class. I personally think it isn't a good idea to implement Connection in this class as this would effectively require 1 connection for every thread, instead of reused connections. giveBackConnection will be renamed to reclaimConnection. Using Supplier instead of a custom interface seems like a anti-pattern to me because that class only allows unchecked exceptions, so subclasses are forced to return null. Have a nice day. \$\endgroup\$ – Ferrybig Feb 15 '16 at 15:12

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.