refactor threadpool to become more eventlet.GreenPool like

and to not directly subclass anything from threading.
Also don't use threading.Thread but thread.start_thread
as we are wrapping the thread anyway.

Author: hpk42

execnet/multi.py

@@ -237,20 +237,20 @@ def safe_terminate(timeout, list_of_paired_functions):
     workerpool = WorkerPool(len(list_of_paired_functions)*2)
 
     def termkill(termfunc, killfunc):
-        termreply = workerpool.dispatch(termfunc)
+        termreply = workerpool.spawn(termfunc)
         try:
             termreply.get(timeout=timeout)
         except IOError:
             killfunc()
 
     replylist = []
     for termfunc, killfunc in list_of_paired_functions:
-        reply = workerpool.dispatch(termkill, termfunc, killfunc)
+        reply = workerpool.spawn(termkill, termfunc, killfunc)
         replylist.append(reply)
     for reply in replylist:
         reply.get()
     workerpool.shutdown()
-    workerpool.join()
+    workerpool.waitall()
 
 
 default_group = Group()

execnet/threadpool.py

@@ -4,6 +4,11 @@
 (c) 2009, holger krekel
 """
 import threading
+try:
+    import thread
+except ImportError:
+    import _thread as thread
+
 import time
 import sys
 
@@ -22,7 +27,7 @@
 class Reply(object):
     """ reply instances provide access to the result
         of a function execution that got dispatched
-        through WorkerPool.dispatch()
+        through WorkerPool.spawn()
     """
     _excinfo = None
     def __init__(self, task):
@@ -54,12 +59,11 @@ def get(self, timeout=None):
             reraise(excinfo[0], excinfo[1], excinfo[2]) # noqa
         return result
 
-class WorkerThread(threading.Thread):
+class WorkerThread:
     def __init__(self, pool):
-        threading.Thread.__init__(self)
         self._queue = queue.Queue()
         self._pool = pool
-        self.setDaemon(1)
+        self._finishevent = threading.Event()
 
     def _run_once(self):
         reply = self._queue.get()
@@ -79,16 +83,23 @@ def _run_once(self):
         # at this point, reply, task and all other local variables go away
         return True
 
+    def start(self):
+        self.id = thread.start_new_thread(self.run, ())
+
+    @property
+    def dead(self):
+        return self._finishevent.isSet()
+
     def run(self):
         try:
             while self._run_once():
-                self._pool._ready[self] = True
+                self._pool._ready.add(self)
         finally:
-            del self._pool._alive[self]
             try:
-                del self._pool._ready[self]
+                self._pool._ready.remove(self)
             except KeyError:
                 pass
+            self._finishevent.set()
 
     def send(self, task):
         reply = Reply(task)
@@ -98,8 +109,11 @@ def send(self, task):
     def stop(self):
         self._queue.put(SystemExit)
 
+    def join(self, timeout=None):
+        self._finishevent.wait(timeout)
+
 class WorkerPool(object):
-    """ A WorkerPool allows to dispatch function executions
+    """ A WorkerPool allows to spawn function executions
         to threads.  Each Worker Thread is reused for multiple
         function executions. The dispatching operation
         takes care to create and dispatch to existing
@@ -116,29 +130,30 @@ def __init__(self, maxthreads=None):
             create up to `maxthreads` worker threads.
         """
         self.maxthreads = maxthreads
-        self._ready = {}
-        self._alive = {}
+        self._running = set()
+        self._ready = set()
 
-    def dispatch(self, func, *args, **kwargs):
+    def spawn(self, func, *args, **kwargs):
         """ return Reply object for the asynchronous dispatch
             of the given func(*args, **kwargs) in a
             separate worker thread.
         """
         if self._shuttingdown:
             raise IOError("WorkerPool is already shutting down")
         try:
-            thread, _ = self._ready.popitem()
+            thread = self._ready.pop()
         except KeyError: # pop from empty list
-            if self.maxthreads and len(self._alive) >= self.maxthreads:
-                raise IOError("can't create more than %d threads." %
+            if self.maxthreads and len(self._running) >= self.maxthreads:
+                raise IOError("maximum of %d threads are busy, "
+                              "can't create more." %
                               (self.maxthreads,))
             thread = self._newthread()
         return thread.send((func, args, kwargs))
 
     def _newthread(self):
         thread = WorkerThread(self)
-        self._alive[thread] = True
         thread.start()
+        self._running.add(thread)
         return thread
 
     def shutdown(self):
@@ -147,22 +162,23 @@ def shutdown(self):
         """
         if not self._shuttingdown:
             self._shuttingdown = True
-            for t in list(self._alive):
+            for t in self._running:
                 t.stop()
 
-    def join(self, timeout=None):
+    def waitall(self, timeout=None):
         """ wait until all worker threads have terminated. """
         deadline = delta = None
         if timeout is not None:
             deadline = time.time() + timeout
-        for thread in list(self._alive):
+        while self._running:
+            thread = self._running.pop()
             if deadline:
                 delta = deadline - time.time()
                 if delta <= 0:
                     raise IOError("timeout while joining threads")
             thread.join(timeout=delta)
-            if thread.isAlive():
-                raise IOError("timeout while joining threads")
+            if not thread.dead:
+                raise IOError("timeout while joining thread %s" % thread.id)
 
 if __name__ == '__channelexec__':
     maxthreads = channel.receive()  # noqa
@@ -176,7 +192,7 @@ def join(self, timeout=None):
         if task is None:
             gw._trace("thread-dispatcher got None, exiting")
             execpool.shutdown()
-            execpool.join()
+            execpool.waitall()
             raise gw._StopExecLoop
         gw._trace("dispatching exec task to thread pool")
-        execpool.dispatch(gw.executetask, task)
+        execpool.spawn(gw.executetask, task)

testing/test_termination.py

@@ -18,7 +18,7 @@ def doit():
         return 17
 
     pool = WorkerPool()
-    reply = pool.dispatch(doit)
+    reply = pool.spawn(doit)
     x = reply.get(timeout=5.0)
     assert x == 17
 
@@ -102,7 +102,7 @@ def flush(self):
     popen = subprocess.Popen([str(anypython), str(p)], stdout=subprocess.PIPE)
     # sync with start-up
     popen.stdout.readline()
-    reply = WorkerPool(1).dispatch(popen.communicate)
+    reply = WorkerPool(1).spawn(popen.communicate)
     reply.get(timeout=50)
     out, err = capfd.readouterr()
     lines = [x for x in err.splitlines()

testing/test_threadpool.py

@@ -14,26 +14,21 @@ def f(i):
         while q.qsize():
             py.std.time.sleep(0.01)
     for i in range(num):
-        pool.dispatch(f, i)
+        pool.spawn(f, i)
     for i in range(num):
         q.get()
-    assert len(pool._alive) == 4
+    assert len(pool._running) == 4
     pool.shutdown()
-    # XXX I replaced the following join() with a time.sleep(1), which seems
-    # to fix the test on Windows, and doesn't break it on Linux... Completely
-    # unsure what the idea is, though, so it would be nice if someone with some
-    # more understanding of what happens here would either fix this better, or
-    # remove this comment...
-    # pool.join(timeout=1.0)
-    py.std.time.sleep(1)
-    assert len(pool._alive) == 0
-    assert len(pool._ready) == 0
+    pool.waitall(timeout=1.0)
+    #py.std.time.sleep(1)  helps on windows?
+    assert len(pool._running) == 0
+    assert len(pool._running) == 0
 
 def test_get():
     pool = WorkerPool()
     def f():
         return 42
-    reply = pool.dispatch(f)
+    reply = pool.spawn(f)
     result = reply.get()
     assert result == 42
 
@@ -42,15 +37,15 @@ def test_get_timeout():
     def f():
         py.std.time.sleep(0.2)
         return 42
-    reply = pool.dispatch(f)
+    reply = pool.spawn(f)
     with py.test.raises(IOError):
         reply.get(timeout=0.01)
 
 def test_get_excinfo():
     pool = WorkerPool()
     def f():
         raise ValueError("42")
-    reply = pool.dispatch(f)
+    reply = pool.spawn(f)
     with py.test.raises(ValueError):
         reply.get(1.0)
     with pytest.raises(EOFError):
@@ -61,34 +56,34 @@ def test_maxthreads():
     def f():
         py.std.time.sleep(0.5)
     try:
-        pool.dispatch(f)
-        py.test.raises(IOError, pool.dispatch, f)
+        pool.spawn(f)
+        py.test.raises(IOError, pool.spawn, f)
     finally:
         pool.shutdown()
 
-def test_join_timeout():
+def test_waitall_timeout():
     pool = WorkerPool()
     q = queue.Queue()
     def f():
         q.get()
-    reply = pool.dispatch(f)
+    reply = pool.spawn(f)
     pool.shutdown()
-    py.test.raises(IOError, pool.join, 0.01)
+    py.test.raises(IOError, pool.waitall, 0.01)
     q.put(None)
     reply.get(timeout=1.0)
-    pool.join(timeout=0.1)
+    pool.waitall(timeout=0.1)
 
 @py.test.mark.skipif("not hasattr(os, 'dup')")
 def test_pool_clean_shutdown():
     capture = py.io.StdCaptureFD()
     pool = WorkerPool()
     def f():
         pass
-    pool.dispatch(f)
-    pool.dispatch(f)
+    pool.spawn(f)
+    pool.spawn(f)
     pool.shutdown()
-    pool.join(timeout=1.0)
-    assert not pool._alive
+    pool.waitall(timeout=1.0)
+    assert not pool._running
     assert not pool._ready
     out, err = capture.reset()
     print(out)

tox.ini

@@ -31,3 +31,5 @@ commands=
 [pytest]
 timeout = 20
 addopts = -rxXs
+rsyncdirs = execnet testing
+