Have the pipeline process working. Added the unit test for this class.

BUG=None
TEST=None

Change-Id: I7fe9fd5b1610959399000b1dfc9b6db55c5c28fb
Reviewed-on: https://gerrit-int.chromium.org/39473
Reviewed-by: Luis Lozano <llozano@chromium.org>
Tested-by: Yuheng Long <yuhenglong@google.com>
Commit-Queue: Yuheng Long <yuhenglong@google.com>

1 files changed, 81 insertions, 14 deletions

diff --git a/bestflags/pipeline_process.py b/bestflags/pipeline_process.py
index 6b878b30..21d3c584 100644
--- a/bestflags/pipeline_process.py
+++ b/bestflags/pipeline_process.py
@@ -1,37 +1,76 @@
 """Pipeline process that encapsulates the actual content.
 
-The actual stages include the Steering algorithm, the builder and the executor.
+The actual stages include the builder and the executor.
 """
 
 __author__ = 'yuhenglong@google.com (Yuheng Long)'
 
 import multiprocessing
 
+# Pick an integer at random.
+POISONPILL = 975
+
 
 class PipelineProcess(multiprocessing.Process):
-  """A process that encapsulates the actual content.
+  """A process that encapsulates the actual content pipeline stage.
 
-  It continuously pull tasks from the queue until a poison pill is received.
+  The actual pipeline stage can be the builder or the tester.  This process
+  continuously pull tasks from the queue until a poison pill is received.
   Once a job is received, it will hand it to the actual stage for processing.
-  """
 
-  # Poison pill means shutdown
-  POISON_PILL = None
+  Each pipeline stage contains three modules.
+  The first module continuously pulls task from the input queue. It searches the
+  cache to check whether the task has encountered before. If so, duplicate
+  computation can be avoided.
+  The second module consists of a pool of workers that do the actual work, e.g.,
+  the worker will compile the source code and get the image in the builder
+  pipeline stage.
+  The third module is a helper that put the result cost to the cost field of the
+  duplicate tasks. For example, if two tasks are equivalent, only one task, say
+  t1 will be executed and the other task, say t2 will not be executed. The third
+  mode gets the result from t1, when it is available and set the cost of t2 to
+  be the same as that of t1.
+  """
 
-  def __init__(self, method, task_queue, result_queue):
+  def __init__(self, num_processes, name, cache, stage, task_queue, helper,
+               worker, result_queue):
     """Set up input/output queue and the actual method to be called.
 
     Args:
-      method: The actual pipeline stage to be invoked.
+      num_processes: Number of helpers subprocessors this stage has.
+      name: The name of this stage.
+      cache: The computed tasks encountered before.
+      stage: An int value that specifies the stage for this pipeline stage, for
+        example, build stage or test stage. This value will be used to retrieve
+        the keys in different stage. I.e., the flags set is the key in build
+        stage and the checksum is the key in the test stage. The key is used to
+        detect duplicates.
       task_queue: The input task queue for this pipeline stage.
+      helper: The method hosted by the helper module to fill up the cost of the
+        duplicate tasks.
+      worker: The method hosted by the worker pools to do the actual work, e.g.,
+        compile the image.
       result_queue: The output task queue for this pipeline stage.
     """
 
     multiprocessing.Process.__init__(self)
-    self._method = method
+
+    self._name = name
     self._task_queue = task_queue
     self._result_queue = result_queue
 
+    self._helper = helper
+    self._worker = worker
+
+    self._cache = cache
+    self._stage = stage
+    self._num_processes = num_processes
+
+    # the queues used by the modules for communication
+    manager = multiprocessing.Manager()
+    self._helper_queue = manager.Queue()
+    self._work_queue = manager.Queue()
+
   def run(self):
     """Busy pulling the next task from the queue for execution.
 
@@ -41,11 +80,39 @@ class PipelineProcess(multiprocessing.Process):
     The process will terminate on receiving the poison pill from previous stage.
     """
 
+    # the worker pool
+    self._pool = multiprocessing.Pool(self._num_processes)
+
+    # the helper process
+    helper_process = multiprocessing.Process(target=self._helper,
+                                             args=(self._cache,
+                                                   self._helper_queue,
+                                                   self._work_queue,
+                                                   self._result_queue))
+    helper_process.start()
+    mycache = self._cache.keys()
+
     while True:
-      next_task = self.task_queue.get()
-      if next_task is None:
+      task = self._task_queue.get()
+      if task == POISONPILL:
         # Poison pill means shutdown
-        self.result_queue.put(None)
+        self._result_queue.put(POISONPILL)
         break
-      self._method(next_task)
-      self.result_queue.put(next_task)
+
+      task_key = task.get_key(self._stage)
+      if task_key in mycache:
+        # The task has been encountered before. It will be sent to the helper
+        # module for further processing.
+        self._helper_queue.put(task)
+      else:
+        # Let the workers do the actual work.
+        self._pool.apply_async(self._worker, args=(task, self._work_queue,
+                                                   self._result_queue))
+        mycache.append(task_key)
+
+    # Shutdown the workers pool and the helper process.
+    self._pool.close()
+    self._pool.join()
+
+    self._helper_queue.put(POISONPILL)
+    helper_process.join()