Change in vdsm[master]: misc: Safer and simpler itmap

Tuesday, 24 March 2015

Nir Soffer has uploaded a new change for review.

Change subject: misc: Safer and simpler itmap
......................................................................

misc: Safer and simpler itmap

The previous code had few issues:

- It used unlimited number of threads by default. This may lead to
  creation of 100's of threads if you do not specify a value.
- It used non-daemon threads, which could lead to unwanted delay during
  vdsm shutdown.
- It tried to yield results before all arguments were handled. This
  could lead to unwanted delay in argument processing, if the caller
  would block processing the results.
- It started one thread per value, even if maxthreads was smaller than
  number of values.
- It was too complicated.

Changes:

- The caller must specify the maximum number of threads.
- Use daemon threads
- Queue all values before yielding results
- Start up to maxthreads worker threads, each processing multiple values
- Simplify the code
- Add test for error handling

Change-Id: Iba6116ac4003702c8e921cebaf494491a6f9afaf
Signed-off-by: Nir Soffer <nsoffer(a)redhat.com&gt;
---
M tests/miscTests.py
M vdsm/storage/misc.py
2 files changed, 42 insertions(+), 42 deletions(-)


  git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/19/39119/1

diff --git a/tests/miscTests.py b/tests/miscTests.py
index 31f64fa..4b3e3c3 100644
--- a/tests/miscTests.py
+++ b/tests/miscTests.py
@@ -196,7 +196,7 @@
         # outOfProcess operation + 1. it let us know that oop and itmap operate
         # properly with their limitations
         data = frozenset(range(oop.HELPERS_PER_DOMAIN + 1))
-        ret = frozenset(misc.itmap(dummy, data, misc.UNLIMITED_THREADS))
+        ret = frozenset(misc.itmap(dummy, data, len(data)))
         self.assertEquals(ret, data)
 
     def testMoreThreadsThanArgs(self):
@@ -207,6 +207,13 @@
         data = 1
         self.assertRaises(ValueError, misc.itmap(int, data, 0).next)
 
+    def testErrors(self):
+        err = Exception()
+        def dummy(arg):
+            raise err
+        data = [1, 2, 3]
+        self.assertEqual(list(misc.itmap(dummy, data, 4)), [err] * len(data))
+
 
 class RotateFiles(TestCaseBase):
 
diff --git a/vdsm/storage/misc.py b/vdsm/storage/misc.py
index eb484c7..463fd04 100644
--- a/vdsm/storage/misc.py
+++ b/vdsm/storage/misc.py
@@ -58,7 +58,6 @@
 STR_UUID_SIZE = 36
 UUID_HYPHENS = [8, 13, 18, 23]
 MEGA = 1 << 20
-UNLIMITED_THREADS = -1
 
 log = logging.getLogger('Storage.Misc')
 
@@ -882,53 +881,47 @@
         raise exception
 
 
-def itmap(func, iterable, maxthreads=UNLIMITED_THREADS):
+def itmap(func, iterable, maxthreads):
     """
-    Make an iterator that computes the function using
-    arguments from the iterable. It works similar to tmap
-    by running each operation in a different thread, this
-    causes the results not to return in any particular
-    order so it's good if you don't care about the order
-    of the results.
-    maxthreads stands for maximum threads that we can initiate simultaneosly.
-               If we reached to max threads the function waits for thread to
-               finish before initiate the next one.
+    Return an iterator calling func with arguments from iterable in multiple threads.
+
+    Unlike tmap, the results are not returned in the original order of the
+    arguments, and number of threads is limited to maxthreads.
     """
-    if maxthreads < 1 and maxthreads != UNLIMITED_THREADS:
-        raise ValueError("Wrong input to function itmap: %s", maxthreads)
+    if maxthreads < 1:
+        raise ValueError("Invalid maxthreads value: %s" % maxthreads)
 
-    respQueue = Queue.Queue()
+    DONE = object()
+    values = Queue.Queue()
+    results = Queue.Queue()
 
-    def wrapper(value):
-        try:
-            respQueue.put(func(value))
-        except Exception as e:
-            respQueue.put(e)
+    def worker():
+        while True:
+            value = values.get()
+            if value is DONE:
+                return
+            try:
+                results.put(func(value))
+            except Exception as e:
+                results.put(e)
 
-    threadsCount = 0
-    for arg in iterable:
-        if maxthreads != UNLIMITED_THREADS:
-            if maxthreads == 0:
-                # This not supposed to happened. If it does, it's a bug.
-                # maxthreads should get to 0 only after threadsCount is
-                # greater than 1
-                if threadsCount < 1:
-                    raise RuntimeError("No thread initiated")
-                else:
-                    yield respQueue.get()
-                    # if yield returns one thread stopped, so we can run
-                    # another thread in queue
-                    maxthreads += 1
-                    threadsCount -= 1
+    count = 0
+    threads = 0
 
-        t = threading.Thread(target=wrapper, args=(arg,))
-        t.start()
-        threadsCount += 1
-        maxthreads -= 1
+    for value in iterable:
+        values.put(value)
+        count += 1
+        if threads < maxthreads:
+            t = threading.Thread(target=worker)
+            t.daemon = True
+            t.start()
+            threads += 1
 
-    # waiting for rest threads to end
-    for i in xrange(threadsCount):
-        yield respQueue.get()
+    for _ in range(threads):
+        values.put(DONE)
+
+    for _ in xrange(count):
+        yield results.get()
 
 
 def isAscii(s):


-- 
To view, visit https://gerrit.ovirt.org/39119
To unsubscribe, visit https://gerrit.ovirt.org/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: Iba6116ac4003702c8e921cebaf494491a6f9afaf
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Nir Soffer <nsoffer(a)redhat.com&gt;

    

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011