Make sure all pools are stopped when shuting down

riak/client/__init__.py

@@ -130,6 +130,13 @@ def __init__(self, protocol='pbc', transport_options={},
         self._bucket_types = WeakValueDictionary()
         self._tables = WeakValueDictionary()
 
+    def __del__(self):
+        if self._multiget_pool:
+            self._multiget_pool.stop()
+
+        if self._multiput_pool:
+            self._multiput_pool.stop()
+
     def _get_protocol(self):
         return self._protocol
 

riak/client/multi.py

@@ -7,6 +7,8 @@
 from riak.riak_object import RiakObject
 from riak.ts_object import TsObject
 
+import atexit
+
 if PY2:
     from Queue import Queue
 else:
@@ -102,9 +104,11 @@ def stop(self):
         """
         Signals the worker threads to exit and waits on them.
         """
-        self._stop.set()
-        for worker in self._workers:
-            worker.join()
+
+        if not self.stopped():
+            self._stop.set()
+            for worker in self._workers:
+                worker.join()
 
     def stopped(self):
         """
@@ -193,6 +197,16 @@ def _worker_method(self):
 RIAK_MULTIPUT_POOL = MultiPutPool()
 
 
+def stop_pools():
+    """Stop worker pools at exit."""
+
+    RIAK_MULTIGET_POOL.stop()
+    RIAK_MULTIPUT_POOL.stop()
+
+
+atexit.register(stop_pools)
+
+
 def multiget(client, keys, **options):
     """Executes a parallel-fetch across multiple threads. Returns a list
     containing :class:`~riak.riak_object.RiakObject` or