[oe-commits] Richard Purdie : oe/utils: Add simple threaded pool implementation

[git at git.openembedded.org]

Module: openembedded-core.git
Branch: master-next
Commit: 44ae778fefca5112900b870be7a485360c50bc2e
URL:    http://git.openembedded.org/?p=openembedded-core.git&a=commit;h=44ae778fefca5112900b870be7a485360c50bc2e

Author: Richard Purdie <richard.purdie at linuxfoundation.org>
Date:   Mon Jun  1 22:15:34 2015 +0100

oe/utils: Add simple threaded pool implementation

Python 2.7 doesn't have a threaded pool implementation, just a multiprocessing
one. We have need of a threaded implementation so add some simple class code
to support this.

Signed-off-by: Richard Purdie <richard.purdie at linuxfoundation.org>

---

 meta/lib/oe/utils.py | 41 +++++++++++++++++++++++++++++++++++++++++
 1 file changed, 41 insertions(+)

diff --git a/meta/lib/oe/utils.py b/meta/lib/oe/utils.py
index 7173e10..0de8800 100644
--- a/meta/lib/oe/utils.py
+++ b/meta/lib/oe/utils.py
@@ -207,3 +207,44 @@ def multiprocess_exec(commands, function):
 def squashspaces(string):
     import re
     return re.sub("\s+", " ", string).strip()
+
+#
+# Python 2.7 doesn't have threaded pools (just multiprocessing)
+# so implement a version here
+#
+
+from Queue import Queue
+from threading import Thread
+
+class ThreadedWorker(Thread):
+    """Thread executing tasks from a given tasks queue"""
+    def __init__(self, tasks):
+        Thread.__init__(self)
+        self.tasks = tasks
+        self.daemon = True
+        self.start()
+
+    def run(self):
+        while True:
+            func, args, kargs = self.tasks.get()
+            try:
+                func(*args, **kargs)
+            except Exception, e:
+                print e
+            finally:
+                self.tasks.task_done()
+
+class ThreadedPool:
+    """Pool of threads consuming tasks from a queue"""
+    def __init__(self, num_threads):
+        self.tasks = Queue(num_threads)
+        for _ in range(num_threads): ThreadedWorker(self.tasks)
+
+    def add_task(self, func, *args, **kargs):
+        """Add a task to the queue"""
+        self.tasks.put((func, args, kargs))
+
+    def wait_completion(self):
+        """Wait for completion of all the tasks in the queue"""
+        self.tasks.join()
+