Merge branch 'fix-829'

NEWS

@@ -24,8 +24,6 @@ New features:
   (:ticket:`#732`).
 - Added *fetch* parameter to `~psycopg2.extras.execute_values()` function
   (:ticket:`#813`).
-- Fixed adaptation of numeric subclasses such as `~enum.IntEnum`
-  (:ticket:`#591`).
 - `!str()` on `~psycopg2.extras.Range` produces a human-readable representation
   (:ticket:`#773`).
 - `~psycopg2.extras.DictCursor` and `~psycopg2.extras.RealDictCursor` rows
@@ -33,8 +31,15 @@ New features:
 - Added `~psycopg2.extensions.Diagnostics.severity_nonlocalized` attribute on
   the `~psycopg2.extensions.Diagnostics` object (:ticket:`#783`).
 - More efficient `~psycopg2.extras.NamedTupleCursor` (:ticket:`#838`).
-- Async communication improved to fix blocking on results returned at
+
-  different times (:ticket:`#856`).
+Bug fixes:
+
+- Fixed connections occasionally broken by the unrelated use of the
+  multiprocessing module (:ticket:`#829`).
+- Fixed async communication blocking if results are returned in different
+  chunks, e.g. with notices interspersed to the results (:ticket:`#856`).
+- Fixed adaptation of numeric subclasses such as `~enum.IntEnum`
+  (:ticket:`#591`).
 
 Other changes:
 

psycopg/config.h

@@ -33,6 +33,18 @@
 #  define HIDDEN
 #endif
 
+/* support for getpid() */
+#if defined( __GNUC__)
+#define CONN_CHECK_PID
+#include <sys/types.h>
+#include <unistd.h>
+#endif
+#ifdef _WIN32
+/* Windows doesn't seem affected by bug #829: just make it compile. */
+#define pid_t int
+#endif
+
+
 /* debug printf-like function */
 #ifdef PSYCOPG_DEBUG
 extern HIDDEN int psycopg_debug_enabled;
@@ -40,8 +52,6 @@ extern HIDDEN int psycopg_debug_enabled;
 
 #if defined( __GNUC__) && !defined(__APPLE__)
 #ifdef PSYCOPG_DEBUG
-#include <sys/types.h>
-#include <unistd.h>
 #define Dprintf(fmt, args...) \
     if (!psycopg_debug_enabled) ; else \
         fprintf(stderr, "[%d] " fmt "\n", (int) getpid() , ## args)

psycopg/connection.h

@@ -141,6 +141,9 @@ struct connectionObject {
     int isolevel;
     int readonly;
     int deferrable;
+
+    /* the pid this connection was created into */
+    pid_t procpid;
 };
 
 /* map isolation level values into a numeric const */

psycopg/connection_type.c

@@ -1367,6 +1367,10 @@ connection_setup(connectionObject *self, const char *dsn, long int async)
     self->isolevel = ISOLATION_LEVEL_DEFAULT;
     self->readonly = STATE_DEFAULT;
     self->deferrable = STATE_DEFAULT;
+#ifdef CONN_CHECK_PID
+    self->procpid = getpid();
+#endif
+
     /* other fields have been zeroed by tp_alloc */
 
     pthread_mutex_init(&(self->lock), NULL);
@@ -1420,7 +1424,15 @@ connection_dealloc(PyObject* obj)
      * resulting in a double-free segfault (ticket #166). */
     PyObject_GC_UnTrack(self);
 
+    /* close the connection only if this is the same process it was created
+     * into, otherwise using multiprocessing we may close the connection
+     * belonging to another process. */
+#ifdef CONN_CHECK_PID
+    if (self->procpid == getpid())
+#endif
+    {
         conn_close(self);
+    }
 
     if (self->weakreflist) {
         PyObject_ClearWeakRefs(obj);

tests/test_connection.py

@@ -22,14 +22,16 @@
 # FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public
 # License for more details.
 
-import ctypes
 import gc
 import os
 import re
-import subprocess as sp
 import sys
-import threading
 import time
+import ctypes
+import shutil
+import tempfile
+import threading
+import subprocess as sp
 from collections import deque
 from operator import attrgetter
 from weakref import ref
@@ -360,6 +362,47 @@ class ConnectionTests(ConnectingTestCase):
         conn.close()
         self.assert_(conn.pgconn_ptr is None)
 
+    @slow
+    def test_multiprocess_close(self):
+        dir = tempfile.mkdtemp()
+        try:
+            with open(os.path.join(dir, "mptest.py"), 'w') as f:
+                f.write("""\
+import time
+import psycopg2
+
+def thread():
+    conn = psycopg2.connect(%(dsn)r)
+    curs = conn.cursor()
+    for i in range(10):
+        curs.execute("select 1")
+        time.sleep(0.1)
+
+def process():
+    time.sleep(0.2)
+""" % {'dsn': dsn})
+
+            script = ("""\
+import sys
+sys.path.insert(0, %(dir)r)
+import time
+import threading
+import multiprocessing
+import mptest
+
+t = threading.Thread(target=mptest.thread, name='mythread')
+t.start()
+time.sleep(0.2)
+multiprocessing.Process(target=mptest.process, name='myprocess').start()
+t.join()
+""" % {'dir': dir})
+
+            out = sp.check_output(
+                [sys.executable, '-c', script], stderr=sp.STDOUT)
+            self.assertEqual(out, b'', out)
+        finally:
+            shutil.rmtree(dir, ignore_errors=True)
+
 
 class ParseDsnTestCase(ConnectingTestCase):
     def test_parse_dsn(self):