Workerpool.AddManyTasks: check tasks type

Each task has to be a sequence, or the RunTask call will fail.

Signed-off-by: Guido Trotter <ultrotter@google.com>
Reviewed-by: Michael Hanselmann <hansmi@google.com>

diff --git a/lib/workerpool.py b/lib/workerpool.py
index 0887ba4..0ca9155 100644 (file)
--- a/lib/workerpool.py
+++ b/lib/workerpool.py
@@ -27,6 +27,8 @@ import collections
 import logging
 import threading
 
+from ganeti import compat
+
 
 class BaseWorker(threading.Thread, object):
   """Base worker class for worker pools.
@@ -211,6 +213,9 @@ class WorkerPool(object):
     @param tasks: list of args passed to L{BaseWorker.RunTask}
 
     """
+    assert compat.all(isinstance(task, (tuple, list)) for task in tasks), \
+      "Each task must be a sequence"
+
     self._lock.acquire()
     try:
       self._WaitWhileQuiescingUnlocked()

diff --git a/test/ganeti.workerpool_unittest.py b/test/ganeti.workerpool_unittest.py
index ca8fc1a..cd15123 100755 (executable)
--- a/test/ganeti.workerpool_unittest.py
+++ b/test/ganeti.workerpool_unittest.py
@@ -169,6 +169,28 @@ class TestWorkerpool(unittest.TestCase):
 
     self.assertEquals(ctx.GetDoneTasks(), 22)
 
+  def testManyTasksSequence(self):
+    ctx = CountingContext()
+    wp = workerpool.WorkerPool("Test", 3, CountingBaseWorker)
+    try:
+      self._CheckWorkerCount(wp, 3)
+      self.assertRaises(AssertionError, wp.AddManyTasks,
+                        ["Hello world %s" % i for i in range(10)])
+      self.assertRaises(AssertionError, wp.AddManyTasks,
+                        [i for i in range(10)])
+
+      wp.AddManyTasks([(ctx, "Hello world %s" % i, ) for i in range(10)])
+      wp.AddTask(ctx, "A separate hello")
+
+      wp.Quiesce()
+
+      self._CheckNoTasks(wp)
+    finally:
+      wp.TerminateWorkers()
+      self._CheckWorkerCount(wp, 0)
+
+    self.assertEquals(ctx.GetDoneTasks(), 11)
+
   def _CheckNoTasks(self, wp):
     wp._lock.acquire()
     try: