Process notifies when data is received, not when the result is parsed

Notifies process access the connection, is not limited to the result, so
There is the possibility of loss of protocol sync in multithread programs.

Closes ticket #55.

3 files changed, 14 insertions, 4 deletions

diff --git a/NEWS b/NEWS
index 5e3aa14..a14cbf3 100644
--- a/NEWS
+++ b/NEWS
@@ -7,6 +7,8 @@ What's new in psycopg 2.4.2
     transaction property.
   - Psycopg doesn't execute queries at connection time to find the
     default isolation level.
+  - Fixed bug with multithread code potentially causing loss of sync
+    with the server communication or lock of the client (ticket #55).
   - Don't build mx.DateTime support if the module can't be imported
     (ticket #53).
   - Fixed escape for negative numbers prefixed by minus operator
diff --git a/psycopg/connection_int.c b/psycopg/connection_int.c
index 0544957..18a0a14 100644
--- a/psycopg/connection_int.c
+++ b/psycopg/connection_int.c
@@ -174,8 +174,6 @@ conn_notifies_process(connectionObject *self)
     PyObject *notify = NULL;
     PyObject *pid = NULL, *channel = NULL, *payload = NULL;
 
-    /* TODO: we are called without the lock! */
-
     while ((pgn = PQnotifies(self->pgconn)) != NULL) {
 
         Dprintf("conn_notifies_process: got NOTIFY from pid %d, msg = %s",
diff --git a/psycopg/pqpath.c b/psycopg/pqpath.c
index a188deb..e1a4f5f 100644
--- a/psycopg/pqpath.c
+++ b/psycopg/pqpath.c
@@ -665,11 +665,14 @@ pq_is_busy(connectionObject *conn)
 
     res = PQisBusy(conn->pgconn);
 
+    Py_BLOCK_THREADS;
+    conn_notifies_process(conn);
+    Py_UNBLOCK_THREADS;
+
     pthread_mutex_unlock(&(conn->lock));
     Py_END_ALLOW_THREADS;
 
     conn_notice_process(conn);
-    conn_notifies_process(conn);
 
     return res;
 }
@@ -781,6 +784,14 @@ pq_execute(cursorObject *curs, const char *query, int async)
             }
             return -1;
         }
+
+        /* Process notifies here instead of when fetching the tuple as we are
+         * into the same critical section that received the data. Without this
+         * care, reading notifies may disrupt other thread communications.
+         * (as in ticket #55). */
+        Py_BLOCK_THREADS;
+        conn_notifies_process(curs->conn);
+        Py_UNBLOCK_THREADS;
     }
 
     else if (async == 1) {
@@ -1370,7 +1381,6 @@ pq_fetch(cursorObject *curs)
     }
 
     conn_notice_process(curs->conn);
-    conn_notifies_process(curs->conn);
 
     /* error checking, close the connection if necessary (some critical errors
        are not really critical, like a COPY FROM error: if that's the case we