WIP for connection pool

bigchaindb/util.py

@@ -1,6 +1,9 @@
 
 import json
 import time
+import contextlib
+import threading
+import queue
 import multiprocessing as mp
 from datetime import datetime
 
@@ -31,6 +34,31 @@ class ProcessGroup(object):
             self.processes.append(proc)
 
 
+# Inspired by:
+# - http://stackoverflow.com/a/24741694/597097
+def pool(builder, limit=None):
+    lock = threading.Lock()
+    local_pool = queue.Queue()
+    size = 0
+
+    @contextlib.contextmanager
+    def pooled():
+        nonlocal size
+        if size == limit:
+            instance = local_pool.get()
+        else:
+            with lock:
+                if size == limit:
+                    instance = local_pool.get()
+                else:
+                    size += 1
+                    instance = builder()
+        yield instance
+        local_pool.put(instance)
+
+    return pooled
+
+
 def serialize(data):
     """Serialize a dict into a JSON formatted string.
 

tests/test_util.py

@@ -1,7 +1,9 @@
-from bigchaindb import util
+import pytest
 
 
 def test_transform_create(b, user_private_key, user_public_key):
+    from bigchaindb import util
+
     tx = util.create_tx(user_public_key, user_public_key, None, 'CREATE')
     tx = util.transform_create(tx)
     tx = util.sign_tx(tx, b.me_private)
@@ -10,3 +12,13 @@ def test_transform_create(b, user_private_key, user_public_key):
     assert tx['transaction']['new_owner'] == user_public_key
     assert util.verify_signature(tx)
 
+@pytest.mark.skipif(reason='asdf')
+def test_pool():
+    from bigchaindb import util
+
+    pool = util.pool(lambda: 'hello', limit=4)
+
+    assert pool().__enter__() == 'hello'
+    assert pool().__enter__() == 'hello'
+    assert pool().__enter__() == 'hello'
+