Concurrent connection problems with use of poll

Python: Windows 32-bit 2.7.15 & 3.7.9
Psycopg2: 2.7.3.2 upgraded to 2.8.2

Description:
We have a program with multiple threads. We use SQLAlchemy as a way to setup the connections and build the queries.
We have 1 connection pool for all the threads. This used to work fine, when we were using psycopg2 2.7.3.2.

Now we are upgrading all our dependencies which includes psycopg2 and are noticing some weird behavior.
We have narrowed it down to 2 things.

• The upgrade to the newer psycopg2 version
• the use of connection.poll in one of our threads

If we use connection.poll it breaks with weird errors like SSL verification error, SSL wrong version nummer error, file descriptor cannot be a negative integer error or even deadlocks.
We don't seem to have these problems in threads where we only run plain-queries.
It also doesn't only break the connection in the threads that is executing the poll, but other threads aswell.

[dvarrazzo]

What are you using connection.poll() for? Are you using it during connection?

Yeah. The thread that uses the connection.poll is for database notifications with the LISTEN command.
We pull out the psycopg2 connection from SQLAlchemy and use the connection.poll to receive the notifications from the database server.
Like the example shows here: https://www.psycopg.org/docs/advanced.html#asynchronous-notifications

[dvarrazzo]

If you are receiving SSL connection errors I think it means you are using poll() on a connection that is not completed yet.

I think your application is polling indiscriminately other connections instead of just the one used for notifications?

I don't think there is any change related to the connection procedure between the versions you mention. However the libssl used has probably been upgraded: you should check the NEWS file to know the version change.

Can you do just one of the two things and work out better if it's a connection error or if it's about your use of poll()?

We have 2 versions of our program, both with the poll and threads. The only difference between the 2 is the version of psycopg2.

We have been narrowing down to find the specific issue. So we cut a lot of code and eventually narrowed it down to the use of poll and the version of psycopg2. So if we use the same script in python with psycopg2: 2.7.3.2, it works. When we use it in Python with 2.8.2, it breaks.
Edit: My collegue has tested a bit further, he confirmed it happens in versions higher than 2.7.7

If it helps I could post our test script.

[dvarrazzo]

Yes please: being able to reproduce the problem is useful

Python 2.7.11
pip==20.2.2
psycopg2==2.8.6
setuptools==44.1.1
SQLAlchemy==1.3.20
wheel==0.35.1

import select
from threading import Thread
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

class PollerThread(Thread):
    def __init__(self, connection):
        Thread.__init__(self)
        self.setDaemon(1)
        self.connection = connection
    def run(self):
        while True:
            result = select.select([self.connection], [], [], 3.0)
            if result != ([], [], []):
                self.connection.poll()


if __name__ == "__main__":
    dbUser = ""
    dbPassword = ""
    dbServer = ""
    dbName = ""
    fullDatabaseUrl = str("postgresql+psycopg2://%s:%s@%s/%s" % (dbUser, dbPassword, dbServer, dbName))

    Session = sessionmaker(autocommit=True)
    engine = create_engine(fullDatabaseUrl)
    Session.configure(bind=engine)
    session = Session()
    connection = session.connection().connection.connection  # <- psycopg2 connection object
    dbEventListener = PollerThread(connection)
    dbEventListener.start()
    while True:
        result = session.execute("SELECT * FROM table1")
        data = result.fetchall()
        time.sleep(0.1)

Also, thank you for your lighting-quick responses.

[dvarrazzo]

Does the problem also manifest without using sqlalchemy? Could you provide an use case which doesn't use it?

Yes. Just tried it with psycopg2 only. The 2.7 works, but the 2.8 version crashes.

import select
from threading import Thread
import psycopg2
import time


class PollerThread(Thread):
	def __init__(self, connection):
		Thread.__init__(self)
		self.setDaemon(1)
		self.connection = connection

	def run(self):
		while True:
			result = select.select([self.connection], [], [], 3.0)
			if result != ([], [], []):
				self.connection.poll()


if __name__ == "__main__":
	dbUser = ""
	dbPassword = ""
	dbServer = ""
	dbName = ""

	connection = psycopg2.connect(dbname=dbName, user=dbUser, password=dbPassword, host=dbServer)

	dbEventListener = PollerThread(connection)
	dbEventListener.start()

	while True:
		cursor = connection.cursor()
		result = cursor.execute("SELECT * FROM table1")
		data = cursor.fetchall()
		time.sleep(0.1)

[dvarrazzo]

You cannot use the same connection both for normal querying and waiting for notifications. poll() will steal the result and cause problems to the connection internal state.

Okay. But.. It is working in a previous version.. So that is a bug?

[dvarrazzo]

I have to verify that: querying and polling should be protected indeed from each other. Testing your script with latest 2.8 gets stuck somewhere. I only have Python 2.8 handy here so cannot install < 2.8.4. Will try with Python 3.6 as soon as I can.

However, in your program, do use a separate connection for polling.

[dvarrazzo]

I can confirm the bug. Running the script (slightly modified, using select 1 as query) works no problem on branch maint_2_7 with Python 3.6/3.7 and stops or crashes on maint_2_8 and Python 3.7/3.8.

[dvarrazzo]

Likely culprit is the refactoring happened in _conn_poll_advance_read().

[stratsimir]

Hi,

I hit the same bug. After some digging, I found out that pq_get_result_async() is called without proper locking, although such is required (The function should be called with the lock and holding the GIL.). The following patch fixed the issue for me:

--- ../psycopg2-2.8.6/psycopg/connection_int.c	2020-09-05 16:39:27.000000000 -0600
+++ ../psycopg2-2.8.6-patched/psycopg/connection_int.c	2021-06-02 09:17:41.528951575 -0600
@@ -904,8 +904,14 @@ _conn_poll_advance_read(connectionObject

     Dprintf("conn_poll: poll reading");

+    Py_BEGIN_ALLOW_THREADS;
+    pthread_mutex_lock(&(self->lock));
+
     busy = pq_get_result_async(self);

+    pthread_mutex_unlock(&(self->lock));
+    Py_END_ALLOW_THREADS;
+
     switch (busy) {
     case 0: /* result is ready */
         Dprintf("conn_poll: async_status -> ASYNC_DONE");