Python Postgres psycopg2 ThreadedConnectionPool exhausted

Question

I have looked into several 'too many clients' related topic here but still can't solve my problem, so I have to ask this again, for me specific case.

Basically, I set up my local Postgres server and need to do tens of thousands of queries, so I used the Python psycopg2package. Here are my codes:

import psycopg2
import pandas as pd
import numpy as np
from flashtext import KeywordProcessor
from psycopg2.pool import ThreadedConnectionPool
from concurrent.futures import ThreadPoolExecutor

df = pd.DataFrame({'S':['California', 'Ohio', 'Texas'], 'T':['Dispatcher', 'Zookeeper', 'Mechanics']})
# df = pd.concat([df]*10000) # repeat df 10000 times

DSN = "postgresql://User:password@localhost/db"
tcp = ThreadedConnectionPool(1, 800, DSN)

def do_one_query(inputS, inputT):
    conn = tcp.getconn()
    c = conn.cursor()

    q = r"SELECT * from eridata where "State" = 'California' and "Title" = 'Dispatcher' limit 1;"   

    c.execute(q)
    all_results = c.fetchall()
    for row in all_results:
        return row
    tcp.putconn(conn, close=True)

cnt=0
for idx, row in df.iterrows():

    cnt+=1
    with ThreadPoolExecutor(max_workers=1) as pool:
        ret = pool.submit(do_one_query,  row["S"], row["T"])
        print ret.result()
    print cnt

The code runs well with a small df. If I repeat df by 10000 times, I got error message saying connection pool exhausted . I though the connections I used have been closed by this line:

tcp.putconn(conn, close=True)

But I guess actually they are not closed? How can I get around this issue?

Answer 1

Don't pass the parameter close=True. putconn already "closes" the connection. Think of putconn like "put connection back in pool", which is probably what you want.

Answer 2

I know that the question is rather old and you do not need the answer anymore. But for the sake of future readers...

The reason behind using a pool is to have some sort of control over the number of connections to database so that we do not overload the database server. In your case it is 800 which is surprisingly big. It seems like you increased the number to be more than what your application requires at its most so that you do not receive "Pool Exhausted" exception anymore.

To solve that allow me to clarify first. psycopg ThreadPool is a thread safe connection pool but it does not control the max number of connections. it simply raises exception. Read this from the docs:

def getconn(self):
        """Get a connection from the pool.

        If there is no idle connection available, then a new one is opened;
        unless there are already :attr:`.maxconn` connections open, then a
        :class:`PoolError` exception is raised.

        Any connection that is broken, or has been idle for more than
        :attr:`.idle_timeout` seconds, is closed and discarded.
        """

As you might have noticed the exception is raised if max is hit. So while it is a thread safe pool it does not have any control over the number of connections.

Solution: Since you are multithreading, you are gonna need to control your threads behavior and your lucky tool alongside the pool is "semaphore": (Below is just a simplified code snippet of my own to sever the demonstration purpose)

_connPool: Optional[psycopg2.pool.ThreadedConnectionPool] = None
_poolSemaphore = threading.Semaphore(10) #10 is max no of connections in this case

def start():
    try:
        Engine._connPool = psycopg2.pool.ThreadedConnectionPool(1, 10,
                                                                user=DBConnectionInfo.user,
                                                                password=DBConnectionInfo.password,
                                                                host=DBConnectionInfo.host,
                                                                port=DBConnectionInfo.port,
                                                                database=DBConnectionInfo.database)
        print("Engine started")
    except Exception as error:
        print("Engine failed to start. connection pool did not initialize:", error)

def getConnection():
    Engine._poolSemaphore.acquire(blocking=True)
    print("Pool is delivering connection")
    return Engine._connPool.getconn()

def putConnectionBack(conn: psycopg2):
    Engine._connPool.putconn(conn, close=False)
    Engine._poolSemaphore.release()
    print("Pool took back a connection")

The above code opens 10 connections at most. The semaphore allows pool to serve 10 threads (safely) with connections but blocks the rest until one of those ten returns a connection back to pool. It is just then that a semaphore.release() is triggered and a blocked thread runs (from the very same point it was blocked) and receives the connection. So consider getConn() as the process of borrowing a connection from the pool. Be aware that it is very important to return it to pool when your done. What happens it you don't, you might ask. Well idle connections get killed by pool after some timeout period but it does not trigger your semaphore and your blocked thread won't get triggered back to work. In another words the Pool works fine but your code goes to halt.