Fix PYTHON-509, rare connection leaks in Python <= 2.7.0

Author: A. Jesse Jiryu Davis

pymongo/thread_util.py

@@ -15,6 +15,7 @@
 """Utilities to abstract the differences between threads and greenlets."""
 
 import threading
+import sys
 import weakref
 
 have_greenlet = True
@@ -24,6 +25,10 @@
  have_greenlet = False
 
 
+# Do we have to work around http://bugs.python.org/issue1868?
+issue1868 = (sys.version_info[:3] <= (2, 7, 1))
+
+
 class Ident(object):
  def __init__(self):
  self._refs = {}
@@ -47,23 +52,42 @@ def watch(self, callback):
 
 
 class ThreadIdent(Ident):
+ class _DummyLock(object):
+ def acquire(self):
+ pass
+
+ def release(self):
+ pass
+
  def __init__(self):
  super(ThreadIdent, self).__init__()
  self._local = threading.local()
+ if issue1868:
+ self._lock = threading.Lock()
+ else:
+ self._lock = ThreadIdent._DummyLock()
 
  # We watch for thread-death using a weakref callback to a thread local.
  # Weakrefs are permitted on subclasses of object but not object() itself.
  class ThreadVigil(object):
  pass
 
+ def _make_vigil(self):
+ # Assigning to a threadlocal isn't thread-safe in Python <= 2.7.0
+ self._lock.acquire()
+ vigil = getattr(self._local, 'vigil', None)
+ if not vigil:
+ self._local.vigil = vigil = ThreadIdent.ThreadVigil()
+
+ self._lock.release()
+ return vigil
+
  def get(self):
- if not hasattr(self._local, 'vigil'):
- self._local.vigil = ThreadIdent.ThreadVigil()
- return id(self._local.vigil)
+ return id(self._make_vigil())
 
  def watch(self, callback):
- tid = self.get()
- self._refs[tid] = weakref.ref(self._local.vigil, callback)
+ vigil = self._make_vigil()
+ self._refs[id(vigil)] = weakref.ref(vigil, callback)
 
 
 class GreenletIdent(Ident):

test/slow/__init__.py

@@ -0,0 +1,35 @@
+# Copyright 2013 10gen, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Test built in connection-pooling with lots of threads."""
+
+import unittest
+import sys
+
+sys.path[0:0] = [""]
+
+from test.test_pooling_base import _TestMaxPoolSize
+
+
+class TestPoolWithLotsOfThreads(_TestMaxPoolSize, unittest.TestCase):
+ use_greenlets = False
+
+ def test_max_pool_size_with_leaked_request_super_massive(self):
+ # Like test_max_pool_size_with_leaked_request_massive but even more
+ # threads. Tests that socket reclamation works under high load,
+ # especially in Python <= 2.7.0. You may need to raise ulimit.
+ # See http://bugs.python.org/issue1868.
+ nthreads = 1000
+ self._test_max_pool_size(
+ 2, 1, max_pool_size=2 * nthreads, nthreads=nthreads)

test/slow/test_high_concurrency.py

@@ -238,7 +238,7 @@ def run_mongo_thread(self):
  r.lock.release()
  else:
  r.lock.release()
- r.ready.wait(2) # Wait two seconds
+ r.ready.wait(timeout=60)
  assert r.ready.isSet(), "Rendezvous timed out"
 
  for i in range(self.end_request):
@@ -652,10 +652,10 @@ def leak_request():
  # Access the thread local from the main thread to trigger the
  # ThreadVigil's delete callback, returning the request socket to
  # the pool.
- # In Python 2.6 and lesser, a dead thread's locals are deleted
+ # In Python 2.7.0 and lesser, a dead thread's locals are deleted
  # and those locals' weakref callbacks are fired only when another
- # thread accesses the locals and finds the thread state is stale.
- # This is more or less a bug in Python <= 2.6. Accessing the thread
+ # thread accesses the locals and finds the thread state is stale,
+ # see http://bugs.python.org/issue1868. Accessing the thread
  # local from the main thread is a necessary part of this test, and
  # realistic: in a multithreaded web server a new thread will access
  # Pool._ident._local soon after an old thread has died.
@@ -664,20 +664,34 @@ def leak_request():
  # Pool reclaimed the socket
  self.assertEqual(1, len(cx_pool.sockets))
  self.assertEqual(the_sock[0], id(one(cx_pool.sockets).sock))
+ self.assertEqual(0, len(cx_pool._tid_to_sock))
 
 
 class _TestMaxPoolSize(_TestPoolingBase):
  """Test that connection pool keeps proper number of idle sockets open,
  no matter how start/end_request are called. To be run both with threads and
  with greenlets.
  """
- def _test_max_pool_size(self, start_request, end_request):
- c = self.get_client(max_pool_size=4, auto_start_request=False)
- # If you increase nthreads over about 35, note a
- # Gevent 0.13.6 bug on Mac, Greenlet.join() hangs if more than
- # about 35 Greenlets share a MongoClient. Apparently fixed in
- # recent Gevent development.
- nthreads = 10
+ def _test_max_pool_size(
+ self, start_request, end_request, max_pool_size=4, nthreads=10):
+ """Start `nthreads` threads. Each calls start_request `start_request`
+ times, then find_one and waits at a barrier; once all reach the barrier
+ each calls end_request `end_request` times. The test asserts that the
+ pool ends with min(max_pool_size, nthreads) sockets or, if
+ start_request wasn't called, at least one socket.
+
+ This tests both max_pool_size enforcement and that leaked request
+ sockets are eventually returned to the pool when their threads end.
+
+ You may need to increase ulimit -n on Mac.
+
+ If you increase nthreads over about 35, note a
+ Gevent 0.13.6 bug on Mac: Greenlet.join() hangs if more than
+ about 35 Greenlets share a MongoClient. Apparently fixed in
+ recent Gevent development.
+ """
+ c = self.get_client(
+ max_pool_size=max_pool_size, auto_start_request=False)
 
  rendevous = CreateAndReleaseSocket.Rendezvous(
  nthreads, self.use_greenlets)
@@ -716,14 +730,27 @@ def _test_max_pool_size(self, start_request, end_request):
  the_hub.shutdown()
 
  if start_request:
- self.assertEqual(4, len(cx_pool.sockets))
+ # Trigger final cleanup in Python <= 2.7.0.
+ cx_pool._ident.get()
+
+ expected_idle = min(max_pool_size, nthreads)
+ message = (
+ '%d idle sockets (expected %d) and %d request sockets'
+ ' (expected 0)' % (
+ len(cx_pool.sockets), expected_idle,
+ len(cx_pool._tid_to_sock)))
+
+ self.assertEqual(
+ expected_idle, len(cx_pool.sockets), message)
  else:
  # Without calling start_request(), threads can safely share
  # sockets; the number running concurrently, and hence the number
- # of sockets needed, is between 1 and 10, depending on thread-
- # scheduling.
+ # of sockets needed, is between 1 and
+ # min(max_pool_size, nthreads), depending on thread-scheduling.
  self.assertTrue(len(cx_pool.sockets) >= 1)
 
+ self.assertEqual(0, len(cx_pool._tid_to_sock))
+
  def test_max_pool_size(self):
  self._test_max_pool_size(0, 0)
 
@@ -732,13 +759,20 @@ def test_max_pool_size_with_request(self):
 
  def test_max_pool_size_with_redundant_request(self):
  self._test_max_pool_size(2, 1)
+
+ def test_max_pool_size_with_redundant_request2(self):
  self._test_max_pool_size(20, 1)
 
  def test_max_pool_size_with_leaked_request(self):
  # Call start_request() but not end_request() -- when threads die, they
  # should return their request sockets to the pool.
  self._test_max_pool_size(1, 0)
 
+ def test_max_pool_size_with_leaked_request_massive(self):
+ nthreads = 100
+ self._test_max_pool_size(
+ 2, 1, max_pool_size=2 * nthreads, nthreads=nthreads)
+
  def test_max_pool_size_with_end_request_only(self):
  # Call end_request() but not start_request()
  self._test_max_pool_size(0, 1)

test/test_pooling_base.py

@@ -128,7 +128,7 @@ def __init__(self, ident, state):
  del t_watched
  del t_unwatched
 
- # Accessing the thread-local triggers cleanup in Python <= 2.6
+ # Trigger final cleanup in Python <= 2.7.0.
  # http://bugs.python.org/issue1868
  ident.get()
  self.assertEqual(3, len(ids))