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.

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

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",

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