Change in vdsm[master]: WIP: threadpool: add compatibility module
fromani at redhat.com
fromani at redhat.com
Mon Jun 30 16:30:44 UTC 2014
Francesco Romani has uploaded a new change for review.
Change subject: WIP: threadpool: add compatibility module
......................................................................
WIP: threadpool: add compatibility module
WORK IN PROGRESS
Add a compatibility layer with the existing
thread pool implementation, used in the storage
subsystem.
Change-Id: I88f788605b23697f51a5fc6690f902f22272a89e
Signed-off-by: Francesco Romani <fromani at redhat.com>
---
M debian/vdsm-python-threadpool.install
M lib/threadpool/Makefile.am
A lib/threadpool/compat.py
A lib/threadpool/examples/demo_threadpool.py
R lib/threadpool/examples/demo_watchedpool.py
M vdsm.spec.in
6 files changed, 161 insertions(+), 0 deletions(-)
git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/28/29428/1
diff --git a/debian/vdsm-python-threadpool.install b/debian/vdsm-python-threadpool.install
index 4d510ca..bea1282 100644
--- a/debian/vdsm-python-threadpool.install
+++ b/debian/vdsm-python-threadpool.install
@@ -1,4 +1,5 @@
./usr/lib/python2.7/dist-packages/threadpool/__init__.py
+./usr/lib/python2.7/dist-packages/threadpool/compat.py
./usr/lib/python2.7/dist-packages/threadpool/futures.py
./usr/lib/python2.7/dist-packages/threadpool/schedqueue.py
./usr/lib/python2.7/dist-packages/threadpool/watchedpool.py
diff --git a/lib/threadpool/Makefile.am b/lib/threadpool/Makefile.am
index 106d90f..5259a3c 100644
--- a/lib/threadpool/Makefile.am
+++ b/lib/threadpool/Makefile.am
@@ -22,6 +22,7 @@
dist_threadpool_PYTHON = \
__init__.py \
+ compat.py \
futures.py \
schedqueue.py \
watchedpool.py \
diff --git a/lib/threadpool/compat.py b/lib/threadpool/compat.py
new file mode 100644
index 0000000..22546b1
--- /dev/null
+++ b/lib/threadpool/compat.py
@@ -0,0 +1,88 @@
+#
+# Copyright 2014 Red Hat, Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+#
+# Refer to the README and COPYING files for full details of the license
+#
+
+"""
+plain thread pool
+"""
+
+import logging
+import threading
+
+from . import worker
+from . import workqueue
+
+
+def _process_task(id, cmd, args, callback):
+ if id is None: # should retry.
+ pass
+ elif callback is None:
+ threading.current_thread().setName(id)
+ logging.debug("Task: %s running: %s with: %s" %
+ (id, repr(cmd), repr(args)))
+ cmd(args)
+ else:
+ threading.current_thread().setName(id)
+ callback(cmd(args))
+
+
+class ThreadPool(object):
+ def __init__(self, numThreads, waitTimeout=3, maxTasks=100):
+ self._work_interval = waitTimeout
+ self._max_size = maxTasks
+ self._nworkers = numThreads
+ self._running = False
+ self._work_queue = None
+ self._workers = []
+ self.start()
+
+ def start(self):
+ if not self._running:
+ self._work_queue = workqueue.WorkQueue(
+ self._work_interval, self._max_size)
+ for _ in range(self._nworkers):
+ worker = worker.Worker(self._work_queue)
+ worker.start()
+ self._workers.append(worker)
+ logging.info('pool started')
+ self._running = True
+
+ def joinAll(self, waitForThreads=True):
+ """
+ terminate all pooled threads. Thread safe.
+ """
+ self._running = False
+
+ for worker in self._workers:
+ worker.stop()
+ if waitForThreads:
+ worker.join()
+ del worker
+
+ def queueTask(self, id, task, args=None, taskCallback=None):
+ if not self._running:
+ return False
+ if not callable(task):
+ return False
+
+ self._work_queue.post(0,
+ _process_task,
+ (id, task, args, taskCallback))
+
+ return True
diff --git a/lib/threadpool/examples/demo_threadpool.py b/lib/threadpool/examples/demo_threadpool.py
new file mode 100644
index 0000000..18e9fd2
--- /dev/null
+++ b/lib/threadpool/examples/demo_threadpool.py
@@ -0,0 +1,70 @@
+#
+# Copyright 2014 Red Hat, Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+#
+# Refer to the README and COPYING files for full details of the license
+#
+
+from random import randrange
+
+from compat import ThreadPool
+
+
+# Sample task 1: given a start and end value, shuffle integers,
+# then sort them
+
+def sortTask(data):
+ print "SortTask starting for ", data
+ numbers = range(data[0], data[1])
+ for a in numbers:
+ rnd = randrange(0, len(numbers) - 1)
+ a, numbers[rnd] = numbers[rnd], a
+ print "SortTask sorting for ", data
+ numbers.sort()
+ print "SortTask done for ", data
+ return "Sorter ", data
+
+# Sample task 2: just sleep for a number of seconds.
+
+def waitTask(data):
+ print "WaitTask starting for ", data
+ print "WaitTask sleeping for %d seconds" % data
+ sleep(data)
+ return "Waiter ", data
+
+# Both tasks use the same callback
+
+def taskCallback(data):
+ print "Callback called for", data
+
+# Create a pool with three worker threads
+
+pool = ThreadPool(100)
+
+# Insert tasks into the queue and let them run
+print "Running tasks: ", pool.getRunningTasks(), "\n"
+pool.queueTask(sortTask, (1000, 100000), taskCallback)
+print "Running tasks: ", pool.getRunningTasks(), "\n"
+pool.queueTask(waitTask, 5, taskCallback)
+pool.queueTask(sortTask, (200, 200000), taskCallback)
+pool.queueTask(waitTask, 2, taskCallback)
+print "Running tasks: ", pool.getRunningTasks(), "\n"
+pool.queueTask(sortTask, (3, 30000), taskCallback)
+pool.queueTask(waitTask, 7, taskCallback)
+
+# When all tasks are finished, allow the threads to terminate
+pool.joinAll()
+print "ThreadPool sample done.\n"
diff --git a/lib/threadpool/examples/demo.py b/lib/threadpool/examples/demo_watchedpool.py
similarity index 100%
rename from lib/threadpool/examples/demo.py
rename to lib/threadpool/examples/demo_watchedpool.py
diff --git a/vdsm.spec.in b/vdsm.spec.in
index 4688695..808c3a0 100644
--- a/vdsm.spec.in
+++ b/vdsm.spec.in
@@ -1438,6 +1438,7 @@
%files python-threadpool
%{python_sitelib}/threadpool/__init__.py*
+%{python_sitelib}/threadpool/compat.py*
%{python_sitelib}/threadpool/futures.py*
%{python_sitelib}/threadpool/schedqueue.py*
%{python_sitelib}/threadpool/watchedpool.py*
--
To view, visit http://gerrit.ovirt.org/29428
To unsubscribe, visit http://gerrit.ovirt.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I88f788605b23697f51a5fc6690f902f22272a89e
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Francesco Romani <fromani at redhat.com>
More information about the vdsm-patches
mailing list