pipeline_process.py revision 49358b75c25a44760e884245440dc96e55812d04 
1"""Pipeline process that encapsulates the actual content.
2
3Part of the Chrome build flags optimization.
4
5The actual stages include the builder and the executor.
6"""
7
8__author__ = 'yuhenglong@google.com (Yuheng Long)'
9
10import multiprocessing
11
12# Pick an integer at random.
13POISONPILL = 975
14
15
16class PipelineProcess(multiprocessing.Process):
17  """A process that encapsulates the actual content pipeline stage.
18
19  The actual pipeline stage can be the builder or the tester.  This process
20  continuously pull tasks from the queue until a poison pill is received.
21  Once a job is received, it will hand it to the actual stage for processing.
22
23  Each pipeline stage contains three modules.
24  The first module continuously pulls task from the input queue. It searches the
25  cache to check whether the task has encountered before. If so, duplicate
26  computation can be avoided.
27  The second module consists of a pool of workers that do the actual work, e.g.,
28  the worker will compile the source code and get the image in the builder
29  pipeline stage.
30  The third module is a helper that put the result cost to the cost field of the
31  duplicate tasks. For example, if two tasks are equivalent, only one task, say
32  t1 will be executed and the other task, say t2 will not be executed. The third
33  mode gets the result from t1, when it is available and set the cost of t2 to
34  be the same as that of t1.
35  """
36
37  def __init__(self, num_processes, name, cache, stage, task_queue, helper,
38               worker, result_queue):
39    """Set up input/output queue and the actual method to be called.
40
41    Args:
42      num_processes: Number of helpers subprocessors this stage has.
43      name: The name of this stage.
44      cache: The computed tasks encountered before.
45      stage: An int value that specifies the stage for this pipeline stage, for
46        example, build stage or test stage. This value will be used to retrieve
47        the keys in different stage. I.e., the flags set is the key in build
48        stage and the checksum is the key in the test stage. The key is used to
49        detect duplicates.
50      task_queue: The input task queue for this pipeline stage.
51      helper: The method hosted by the helper module to fill up the cost of the
52        duplicate tasks.
53      worker: The method hosted by the worker pools to do the actual work, e.g.,
54        compile the image.
55      result_queue: The output task queue for this pipeline stage.
56    """
57
58    multiprocessing.Process.__init__(self)
59
60    self._name = name
61    self._task_queue = task_queue
62    self._result_queue = result_queue
63
64    self._helper = helper
65    self._worker = worker
66
67    self._cache = cache
68    self._stage = stage
69    self._num_processes = num_processes
70
71    # the queues used by the modules for communication
72    manager = multiprocessing.Manager()
73    self._helper_queue = manager.Queue()
74    self._work_queue = manager.Queue()
75
76  def run(self):
77    """Busy pulling the next task from the queue for execution.
78
79    Once a job is pulled, this stage invokes the actual stage method and submits
80    the result to the next pipeline stage.
81
82    The process will terminate on receiving the poison pill from previous stage.
83    """
84
85    # the worker pool
86    work_pool = multiprocessing.Pool(self._num_processes)
87
88    # the helper process
89    helper_process = multiprocessing.Process(target=self._helper,
90                                             args=(self._cache,
91                                                   self._helper_queue,
92                                                   self._work_queue,
93                                                   self._result_queue))
94    helper_process.start()
95    mycache = self._cache.keys()
96
97    while True:
98      task = self._task_queue.get()
99      if task == POISONPILL:
100        # Poison pill means shutdown
101        self._result_queue.put(POISONPILL)
102        break
103
104      task_key = task.get_key(self._stage)
105      if task_key in mycache:
106        # The task has been encountered before. It will be sent to the helper
107        # module for further processing.
108        self._helper_queue.put(task)
109      else:
110        # Let the workers do the actual work.
111        work_pool.apply_async(self._worker, args=(task, self._work_queue,
112                                                  self._result_queue))
113        mycache.append(task_key)
114
115    # Shutdown the workers pool and the helper process.
116    work_pool.close()
117    work_pool.join()
118
119    self._helper_queue.put(POISONPILL)
120    helper_process.join()
121