Fix multiprocessing bug #313

Author: kevinjqiu

couchdb/http.py

@@ -13,6 +13,7 @@
 
 from base64 import b64encode
 from datetime import datetime
+import multiprocessing
 import errno
 import socket
 import time
@@ -482,17 +483,20 @@ class ConnectionPool(object):
     def __init__(self, timeout, disable_ssl_verification=False):
         self.timeout = timeout
         self.disable_ssl_verification = disable_ssl_verification
-        self.conns = {} # HTTP connections keyed by (scheme, host)
+        self.conns = {} # HTTP connections keyed by (current_process_pid, scheme, host)
         self.lock = Lock()
 
-    def get(self, url):
+    @property
+    def _current_process_id(self):
+        return multiprocessing.current_process().pid
 
+    def get(self, url):
         scheme, host = util.urlsplit(url, 'http', False)[:2]
 
         # Try to reuse an existing connection.
         self.lock.acquire()
         try:
-            conns = self.conns.setdefault((scheme, host), [])
+            conns = self.conns.setdefault((self._current_process_id, scheme, host), [])
             if conns:
                 conn = conns.pop(-1)
             else:
@@ -520,7 +524,7 @@ def release(self, url, conn):
         scheme, host = util.urlsplit(url, 'http', False)[:2]
         self.lock.acquire()
         try:
-            self.conns.setdefault((scheme, host), []).append(conn)
+            self.conns.setdefault((self._current_process_id, scheme, host), []).append(conn)
         finally:
             self.lock.release()
 

couchdb/tests/client.py

@@ -7,6 +7,7 @@
 # you should have received as part of this distribution.
 
 from datetime import datetime
+import multiprocessing
 import os
 import os.path
 import shutil
@@ -19,6 +20,10 @@
 from couchdb.tests import testutil
 
 
+def _current_pid():
+    return multiprocessing.current_process().pid
+
+
 class ServerTestCase(testutil.TempDatabaseMixin, unittest.TestCase):
 
     def test_init_with_resource(self):
@@ -481,7 +486,9 @@ def test_changes_releases_conn(self):
         # that the HTTP connection made it to the pool.
         list(self.db.changes(feed='continuous', timeout=0))
         scheme, netloc = util.urlsplit(client.DEFAULT_BASE_URL)[:2]
-        self.assertTrue(self.db.resource.session.connection_pool.conns[(scheme, netloc)])
+        current_pid = _current_pid()
+        key = (current_pid, scheme, netloc)
+        self.assertTrue(self.db.resource.session.connection_pool.conns[key])
 
     def test_changes_releases_conn_when_lastseq(self):
         # Consume a changes feed, stopping at the 'last_seq' item, i.e. don't
@@ -490,8 +497,10 @@ def test_changes_releases_conn_when_lastseq(self):
         for obj in self.db.changes(feed='continuous', timeout=0):
             if 'last_seq' in obj:
                 break
+        current_pid = _current_pid()
         scheme, netloc = util.urlsplit(client.DEFAULT_BASE_URL)[:2]
-        self.assertTrue(self.db.resource.session.connection_pool.conns[(scheme, netloc)])
+        key = (current_pid, current_pid, scheme, netloc)
+        self.assertTrue(self.db.resource.session.connection_pool.conns[key])
 
     def test_changes_conn_usable(self):
         # Consume a changes feed to get a used connection in the pool.