Check/set connection status at commit inside the critical section

Failing to do so was causing the issue reported in ticket #103. The issue
as reported was fixed when SET ISOLATION LEVEL was dropped, but the real
problem wasn't fixed.

NEWS

@@ -12,6 +12,7 @@ What's new in psycopg 2.4.5
   - Fixed 'rownumber' during iteration on cursor subclasses.
     Regression introduced in 2.4.4 (ticket #100).
   - Added support for 'inet' arrays.
+  - Fixed 'commit()' concurrency problem (ticket #103).
 
 
 What's new in psycopg 2.4.4

psycopg/pqpath.c

@@ -444,34 +444,35 @@ pq_commit(connectionObject *conn)
     PGresult *pgres = NULL;
     char *error = NULL;
 
+    Py_BEGIN_ALLOW_THREADS;
+    pthread_mutex_lock(&conn->lock);
+
     Dprintf("pq_commit: pgconn = %p, autocommit = %d, status = %d",
             conn->pgconn, conn->autocommit, conn->status);
 
     if (conn->autocommit || conn->status != CONN_STATUS_BEGIN) {
         Dprintf("pq_commit: no transaction to commit");
-        return 0;
+        retvalue = 0;
+    }
+    else {
+        conn->mark += 1;
+        retvalue = pq_execute_command_locked(conn, "COMMIT", &pgres, &error, &_save);
     }
-
-    Py_BEGIN_ALLOW_THREADS;
-    pthread_mutex_lock(&conn->lock);
-    conn->mark += 1;
-
-    retvalue = pq_execute_command_locked(conn, "COMMIT", &pgres, &error, &_save);
 
     Py_BLOCK_THREADS;
     conn_notice_process(conn);
     Py_UNBLOCK_THREADS;
 
+    /* Even if an error occurred, the connection will be rolled back,
+       so we unconditionally set the connection status here. */
+    conn->status = CONN_STATUS_READY;
+
     pthread_mutex_unlock(&conn->lock);
     Py_END_ALLOW_THREADS;
 
     if (retvalue < 0)
         pq_complete_error(conn, &pgres, &error);
 
-    /* Even if an error occurred, the connection will be rolled back,
-       so we unconditionally set the connection status here. */
-    conn->status = CONN_STATUS_READY;
-
     return retvalue;
 }
 

tests/test_connection.py

@@ -166,6 +166,35 @@ class ConnectionTests(unittest.TestCase):
         gc.collect()
         self.assert_(w() is None)
 
+    def test_commit_concurrency(self):
+        # The problem is the one reported in ticket #103. Because of bad
+        # status check, we commit even when a commit is already on its way.
+        # We can detect this condition by the warnings.
+        conn = self.conn
+        notices = []
+        stop = []
+
+        def committer():
+            while not stop:
+                conn.commit()
+                while conn.notices:
+                    notices.append((2, conn.notices.pop()))
+
+        cur = conn.cursor()
+        t1 = threading.Thread(target=committer)
+        t1.start()
+        i = 1
+        for i in range(1000):
+            cur.execute("select %s;",(i,))
+            conn.commit()
+            while conn.notices:
+                notices.append((1, conn.notices.pop()))
+
+        # Stop the committer thread
+        stop.append(True)
+
+        self.assert_(not notices, "%d notices raised" % len(notices))
+
 
 class IsolationLevelsTestCase(unittest.TestCase):