1
\$\begingroup\$

When I talk to people regarding connection pooling in SQLite most of them always laugh and say "You don't know about SQLite", "It is not a client server DB, and only overhead in creating new connection is opening a file connection".

I agree but connection pooling helps in SQLite if we want to make use of PRAGMA cache_size. This cache works at sqlite connection level and if we close connection the cache will be discarded (it is also discarded when database file changes).

So in order to make use of cache_size when set with higher values it is better to pool connections.

Here is my simple implementation in python

import queue
import sqlite3
from contextlib import contextmanager


class ConnectionPool:
    def __init__(self, max_connections, database):
        self.max_connections = max_connections
        self.database = database
        self.pool = queue.Queue(maxsize=max_connections)

        for _ in range(max_connections):
            conn = self.create_connection()
            self.pool.put(conn)

    def create_connection(self):
        return sqlite3.connect(self.database)

    def get_connection(self, timeout):
        try:
            return self.pool.get(timeout=timeout)
        except queue.Empty:
            raise RuntimeError("Timeout: No available pool in the pool.")

    def release_connection(self, conn):
        self.pool.put(conn)

    @contextmanager
    def connection(self, timeout=10):
        conn = self.get_connection(timeout)
        try:
            yield conn
        finally:
            self.release_connection(conn)


if __name__ == "__main__":
    pool = ConnectionPool(5, 'cp.db')

    with pool.connection() as connection:
        cursor = connection.cursor()
        cursor.execute('SELECT 1')
        result = cursor.fetchall()
        print(result)

Please feel free to give feedbacks and suggestion for improving above connection pool implementation

\$\endgroup\$

2 Answers 2

2
\$\begingroup\$

I would make a few suggestions. The last two, listed separately, may be of marginal value to you and do increase the complexity of the code considerably. I have included a sample implementation that incorporates these two ideas mostly for my future reference and use. Ironically, I could now use somebody to review this for me!

  1. I would prepend to your class's attribute names a '_' signifying that they are "private."
  2. If a client uses the context manager method connection to obtain a connection, there is a default timeout argument value. However, the same is not true if the client wants to explicitly call get_connection and release_connection. But if these two methods are not meant to be called by the client, then rename these methods so that they have a leading '_'. Otherwise, you can have defaults for both styles of acquisition with that default value specified only once for consistency:
    def get_connection(self, timeout=None):
        if timeout is None:
            timeout = 10
        ... # rest of code omitted

    @contextmanager
    def connection(self, timeout=None):
        conn = self.get_connection(timeout)
        ... # rest of code omitted
  1. You might wish to implement a close method that closes all the connections. This method should only be called when it is known that all the connections have been returned to the pool for good.
  2. I have updated the signature on the __init__ method so that the client can add additional parameters the call to sqlite3.connect.
  3. If you timeout during acquisition, the RuntimeError message should be changed to "Timeout: No available connection in the pool."
  4. A connection pool is just a special case of a more general "resource" pool where the resource in question is a reusable connection. Consider creating an abstract base class, ResourcePool, that can be used to allocate any type of resource. This class would have an abstract method, allocate_resource, that is overridden in a subclass (e.g. your ConectionPool class) and creates an instance of a specific type of resource. Thus, ResourcePool class would have method names such as allocate_resource instead of create_connection, get_resource instead of get_connection, etc. Your ConnectionPool subclass can define, for example, a get_connection method that simply delegates to the base class get_resource method so that the client can use connection-specific method names.

If you want to be slightly more efficient at the cost of additional complexity, then:

  1. You might wish to lazily create connections. Store as attributes the maximum number of connections to create, the number of connections created and the number of connections currently in the pool, i.e. the Queue instance, that are immediately available for allocation. As long as there are unused connections in the queue, you can simply get the next available one. But if the queue is empty, then if the number of connections that have been created is less than the maximum number of connections permitted, you simply create another connection, update the count of created connections and return the new connection. This logic is a critical section that needs to be serialized under control of a threading.Lock instance. Lazy allocation will save you something if the maximum number of concurrent connections in use at one time is considerably less than your pool size and if creating a connection uses a lot of resources. This is probably not your situation.
  2. If a connection is returned to the pool with uncommitted updates, it would be wise to perform a rollback on the connection.
  3. You might also wish to use a collections.deque instance instead of a queue.Queue, since it would be a bit faster. But your code also becomes a bit more complicated.

Update

Your implementation, which used a queue.Queue instance to implement the pool (as well as my previous sample implementation) assumes that the connections would be obtained by threads running concurrently for if it were only a single thread running, what would be the point of that thread obtaining multiple connections to the same database? You would then really only need a pool size of 1. But in this case, you would still be better off just implementing a singleton connection and forget about using a pool.

So the problem now become this: A sqlite3 connection can only be used on the thread that created it (at least that is the case for my Python 3.8.5). So such a pool is really not suitable for a multithreading environment.

My conclusion is that such a pool needs to be using instead asyncio and thus your original implementation needs to be using a asyncio.Queue instance. So my new sample implementation would be:

Sample asyncio Implementation Using a deque and Lazy Connection Creation

import asyncio
from abc import ABC, abstractmethod
from collections import deque
from contextlib import asynccontextmanager

class ResourcePool(ABC):
    def __init__(self, max_resources):
        if max_resources < 1:
            raise ValueError(f'Invalid max_resources argument: {max_resources}')
        self._max_resources = max_resources
        self._request_lock = asyncio.Lock()
        self._resource_returned = asyncio.Condition()
        self._pool = deque()
        self._n_resources_created = 0
        self._is_hashable_resource = self.is_hashable_resource()
        self._in_use = set()

    async def close(self):
        """Close all resources."""

        async with self._request_lock:
            while self._pool:
                res = self._pool.popleft()
                await self.close_resource(res)
            while self._in_use:
                res = self._in_use.pop()
                await self.close_resource(res)

    async def close_resource(self, res):
        """Subclasses should override this method if the resource does
not implement a close method."""
        await res.close()

    def is_hashable_resource(self):
        """Override this function and return False if the resource cannot be
added to a set. Otherwise, we can do some additional error checking and
ensure that all resources are closed when method close is called."""
        return True

    @abstractmethod
    async def create_resource(self):
        pass

    async def get_resource(self, timeout=None):
        if timeout is None:
            timeout = 10

        while True:
            async with self._request_lock:
                if self._pool:
                    res = self._pool.popleft()
                    if self._is_hashable_resource:
                        self._in_use.add(res)
                    return res

                # Can we create another resource?
                if self._n_resources_created < self._max_resources:
                    self._n_resources_created += 1
                    res = await self.create_resource()
                    if self._is_hashable_resource:
                        self._in_use.add(res)
                    return res

                # We must wait for a resource to be returned
                async def wait_for_resource():
                    async with self._resource_returned:
                        await self._resource_returned.wait()

                try:
                    await asyncio.wait_for(wait_for_resource(), timeout=timeout)
                    # The pool now has at least one resource available and
                    # we will succeed on next iteration.
                except asyncio.TimeoutError:
                    raise RuntimeError("Timeout: No available resource in the pool.")


    async def release_resource(self, res):
        if self._is_hashable_resource:
            if res not in self._in_use:
                raise Exception('Releasing unknown object')
            # Could raise exception if two threads are releasing the same resource:
            self._in_use.remove(res)
        self._pool.append(res)
        # If someone is waiting for a resource:
        async with self._resource_returned:
            self._resource_returned.notify()

    @asynccontextmanager
    async def resource(self, timeout=None):
        res = await self.get_resource(timeout)
        try:
            yield res
        finally:
            await self.release_resource(res)


import aiosqlite3

class ConnectionPool(ResourcePool):
    def __init__(self, max_connections, database):
        super().__init__(max_connections)
        self._database = database

    async def create_resource(self):
        return await aiosqlite3.connect(self._database, asyncio.get_running_loop())

    def get_connection(self, timeout=None):
        return self.get_resource(timeout)

    async def release_resource(self, connection):
        await connection.rollback() # clean up
        return await super().release_resource(connection)

    def release_connection(self, connection):
        return self.release_resource(connection)

    def connection(self, timeout=None):
        return self.resource(timeout)


if __name__ == "__main__":
    async def worker(pool, n):
        try:
            for i in range(10):
                async with pool.connection() as conn:
                    cursor = await conn.cursor()
                    await cursor.execute('SELECT COUNT(*) FROM braintree_transaction')
                    row = await cursor.fetchone()
                    await cursor.close()
                    print(n, i, row[0])
                    await asyncio.sleep(.0001)
        except Exception as e:
            print(e)

    connection_pool = ConnectionPool(2, database='fncpl.db')
    tasks = [worker(connection_pool, n) for n in range(3)]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.gather(*tasks))
    loop.run_until_complete(connection_pool.close())
    loop.close()

A sample implementation that is based on threads (not suitable for a sqlite3 pool) would be:

Sample Multi-threaded Implementation

from abc import ABC, abstractmethod
from collections import deque
from threading import Lock, Condition
from contextlib import contextmanager

class ResourcePool(ABC):
    def __init__(self, max_resources):
        if max_resources < 1:
            raise ValueError(f'Invalid max_resources argument: {max_resources}')
        self._max_resources = max_resources
        self._request_lock = Lock()
        self._resource_returned = Condition()
        self._pool = deque()
        self._n_resources_created = 0
        self._is_hashable_resource = self.is_hashable_resource()
        self._in_use = set()

    def close(self):
        """Close all resources."""

        with self._request_lock:
            while self._pool:
                self.close_resource(self._pool.popleft())
            while self._in_use:
                self.close_resource(self._in_use.pop())

    def close_resource(self, res):
        """Subclasses should override this method if the resource does
not implement a close method."""
        res.close()

    def __del__(self):
        """Close all resources when the pool is garbage
collected. Note that due to how close is implemented, it may be
called multiple times."""
        self.close()

    def is_hashable_resource(self):
        """Override this function and return False if the resource cannot be
added to a set. Otherwise, we can do some additional error checking and
ensure that all resources are closed when method close is called."""
        return True

    @abstractmethod
    def create_resource(self):
        pass

    def get_resource(self, timeout=None):
        if timeout is None:
            timeout = 10

        with self._request_lock:
            while True:
                if self._pool:
                    res = self._pool.popleft()
                    if self._is_hashable_resource:
                        self._in_use.add(res)
                    return res

                # Can we create another resource?
                if self._n_resources_created < self._max_resources:
                    self._n_resources_created += 1
                    res = self.create_resource()
                    if self._is_hashable_resource:
                        self._in_use.add(res)
                    return res

                # We must wait for a resource to be returned
                with self._resource_returned:
                    if not self._resource_returned.wait(timeout=timeout):
                        raise RuntimeError("Timeout: No available resource in the pool.")
                    # The pool now has at least one resource available and
                    # we will succeed on next iteration.

    def release_resource(self, res):
        if self._is_hashable_resource:
            if res not in self._in_use:
                raise Exception('Releasing unknown object')
            # Could raise exception if two threads are releasing the same resource:
            self._in_use.remove(res)
        self._pool.append(res)
        # Notify the thread waiting for a resource, if any:
        with self._resource_returned:
            self._resource_returned.notify()

    @contextmanager
    def resource(self, timeout=None):
        res = self.get_resource(timeout)
        try:
            yield res
        finally:
            self.release_resource(res)


import mysql.connector

class ConnectionPool(ResourcePool):
    def __init__(self, max_connections, database, user, password, time_zone='America/New_York'):
        super().__init__(max_connections)
        self._database = database
        self._user = user
        self._password = password
        self._time_zone = time_zone

    def create_resource(self):
        return mysql.connector.connect(database=self._database,
                                       user=self._user,
                                       password=self._password,
                                       charset='utf8mb4',
                                       use_unicode=True,
                                       init_command=f"set session time_zone='{self._time_zone}'"
                                       )

    def get_connection(self, timeout=None):
        return self.get_resource(timeout)

    def release_resource(self, connection):
        connection.rollback() # clean up
        return super().release_resource(connection)

    def release_connection(self, connection):
        return self.release_resource(connection)

    def connection(self, timeout=None):
        return self.resource(timeout)

if __name__ == '__main__':
    from multiprocessing.pool import ThreadPool

    def worker(pool, n):
        try:
            for i in range(10):
                with pool.connection() as conn:
                    cursor = conn.cursor()
                    cursor.execute('SELECT COUNT(*) FROM transaction')
                    cnt = cursor.fetchone()[0]
                    cursor.close()
                    print(n, i, cnt)
        except Exception as e:
            print(e)

    connection_pool = ConnectionPool(2, database='test_db', user='***', password='***')
    thread_pool = ThreadPool(3)
    for n in range(3):
        thread_pool.apply_async(worker, args=(connection_pool, n))
    thread_pool.close()
    thread_pool.join()

Note that in both of these implementations if the resource being created can be hashed, then we can create a set to which we add resources that have been given out to clients. This allows us to provide an additional check to ensure that a client cannot return to the pool a resource that was never taken from the pool. It also allows the close method to close all resources including those that have not been returned back to the pool yet. There is no reason why this logic cannot be incorporated into the asyncio version.

\$\endgroup\$
0
\$\begingroup\$

There is some review context missing here, such as automated tests and realistic callers. You are going for a somewhat subtle point:

connection pooling helps in SQLite if we want to make use of PRAGMA cache_size. ... if we close connection the cache will be discarded

Ok, fair enough. Implicit in that remark is "big queries will go faster" if they benefit from cache.

Within the if __name__ guard we have a client which is nice enough, and I thank you for that example. But it should go further. I really wanted to see a test suite that makes repeated queries and contrasts a pair of benchmark timings:

  1. "slow" without pool
  2. "fast" with pool

If the explicit goal of this code is to improve cache interaction, maybe it should accept a parameter specifying the pragma cache_size? It could default to None, meaning "no change".


My experience in interacting with colleagues of diverse skill levels, around {postgres, mysql} connections mediated by sqlalchemy, is that the "invisible" pool is "voodoo". That is, they often have trouble reasoning about what's happening with the pool, and how code in a PR would alter the behavior.

Consider adding DEBUG logging to the various Markov birth / death events. Each log message should also mention pooled connection counts so we can track connection lifecycles.

Successful calls to connection() can be "fast" or can take e.g. nearly ten seconds. Pick a threshold, perhaps 200 msec, and log at increased priority (INFO?) for each "slow" success.


This codebase appears to meet its design objectives.

I would be willing to delegate or accept maintenance tasks on it.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

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