Always detect when a connection is closed behind psycopg2's back.

There's a race condition that only seems to happen over Unix-domain
sockets. Sometimes, the closed socket is reported by the kernel to
libpq like this (captured with strace):

  sendto(3, "Q\0\0\0\34select pg_backend_pid()\0", 29, MSG_NOSIGNAL, NULL, 0) = 29
  recvfrom(3, "E\0\0\0mSFATAL\0C57P01\0Mterminating "..., 16384, 0, NULL, NULL) = 110
  recvfrom(3, 0x12d0330, 16384, 0, 0, 0)  = -1 ECONNRESET (Connection reset by peer)

That is, psycopg2/libpq sees no error when sending the first query
after the connection is closed, but gets an error reading the result.
In that case, everything worked fine.

But sometimes, the error manifests like this:

  sendto(3, "Q\0\0\0\34select pg_backend_pid()\0", 29, MSG_NOSIGNAL, NULL, 0) = -1 EPIPE (Broken pipe)
  recvfrom(3, "E\0\0\0mSFATAL\0C57P01\0Mterminating "..., 16384, 0, NULL, NULL) = 110
  recvfrom(3, "", 16274, 0, NULL, NULL)   = 0
  recvfrom(3, "", 16274, 0, NULL, NULL)   = 0

i.e. libpq received an error when sending the query. This manifests as
a slightly different exception from a slightly different place. More
importantly, in this case connection.closed is left at 0 rather than
being set to 2, and that is the bug I'm fixing here.

Note that we see almost identical behaviour for sync and async
connections, and the fixes are the same. So I added extremely similar
test cases.

Finally, there is still a bug here: for async connections, we
sometimes raise DatabaseError (incorrect) and sometimes raise
OperationalError (correct). Will fix that next.
This commit is contained in:
Greg Ward 2016-06-28 17:46:21 -04:00 committed by Daniele Varrazzo
parent 3c124a0b87
commit b203a7c775
2 changed files with 51 additions and 0 deletions

View File

@ -986,6 +986,9 @@ pq_execute(cursorObject *curs, const char *query, int async, int no_result, int
/* don't let pgres = NULL go to pq_fetch() */
if (curs->pgres == NULL) {
if (CONNECTION_BAD == PQstatus(curs->conn->pgconn)) {
curs->conn->closed = 2;
}
pthread_mutex_unlock(&(curs->conn->lock));
Py_BLOCK_THREADS;
if (!PyErr_Occurred()) {
@ -1013,6 +1016,9 @@ pq_execute(cursorObject *curs, const char *query, int async, int no_result, int
CLEARPGRES(curs->pgres);
if (PQsendQuery(curs->conn->pgconn, query) == 0) {
if (CONNECTION_BAD == PQstatus(curs->conn->pgconn)) {
curs->conn->closed = 2;
}
pthread_mutex_unlock(&(curs->conn->lock));
Py_BLOCK_THREADS;
PyErr_SetString(OperationalError,

View File

@ -29,6 +29,8 @@ import psycopg2.extensions
from testutils import (unittest, ConnectingTestCase, skip_before_postgres,
skip_if_no_namedtuple, skip_if_no_getrefcount, slow)
import psycopg2.extras
class CursorTests(ConnectingTestCase):
@ -537,6 +539,49 @@ class CursorTests(ConnectingTestCase):
self.assertRaises(exception, cur.callproc, procname, parameter_sequence)
self.conn.rollback()
def test_external_close_sync(self):
# If a "victim" connection is closed by a "control" connection
# behind psycopg2's back, psycopg2 always handles it correctly:
# raise OperationalError, set conn.closed to 2. This reproduces
# issue #443, a race between control_conn closing victim_conn and
# psycopg2 noticing.
control_conn = self.conn
connect_func = self.connect
wait_func = lambda conn: None
self._test_external_close(control_conn, connect_func, wait_func)
def test_external_close_async(self):
# Issue #443 is in the async code too. Since the fix is duplicated,
# so is the test.
control_conn = self.conn
connect_func = lambda: self.connect(async=True)
wait_func = psycopg2.extras.wait_select
self._test_external_close(control_conn, connect_func, wait_func)
def _test_external_close(self, control_conn, connect_func, wait_func):
# The short sleep before using victim_conn the second time makes it
# much more likely to lose the race and see the bug. Repeating the
# test several times makes it even more likely.
for i in range(10):
victim_conn = connect_func()
wait_func(victim_conn)
with victim_conn.cursor() as cur:
cur.execute('select pg_backend_pid()')
wait_func(victim_conn)
pid1 = cur.fetchall()[0][0]
with control_conn.cursor() as cur:
cur.execute('select pg_terminate_backend(%s)', (pid1,))
time.sleep(0.001)
with self.assertRaises(psycopg2.DatabaseError):
with victim_conn.cursor() as cur:
cur.execute('select 1')
wait_func(victim_conn)
self.assertEqual(victim_conn.closed, 2)
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)