OLD | NEW |
1 # Copyright (c) 2011 The Chromium OS Authors. All rights reserved. | 1 # Copyright (c) 2011 The Chromium OS Authors. All rights reserved. |
2 # Use of this source code is governed by a BSD-style license that can be | 2 # Use of this source code is governed by a BSD-style license that can be |
3 # found in the LICENSE file. | 3 # found in the LICENSE file. |
4 | 4 |
5 """Module containing methods/classes related to running parallel test jobs.""" | 5 """Module containing methods/classes related to running parallel test jobs.""" |
6 | 6 |
| 7 import multiprocessing |
7 import sys | 8 import sys |
8 import threading | |
9 import time | 9 import time |
10 | 10 |
11 import cros_build_lib as cros_lib | 11 import cros_build_lib as cros_lib |
12 | 12 |
13 class ParallelJob(threading.Thread): | 13 class ParallelJobTimeoutError(Exception): |
14 """Small wrapper for threading. Thread that releases a semaphores on exit.""" | 14 """Thrown when a job ran for longer than expected.""" |
| 15 pass |
15 | 16 |
16 def __init__(self, starting_semaphore, ending_semaphore, target, args): | 17 |
| 18 class ParallelJob(multiprocessing.Process): |
| 19 """Small wrapper for Process that stores output of its target method.""" |
| 20 |
| 21 MAX_TIMEOUT_SECONDS = 1800 |
| 22 SLEEP_TIMEOUT_SECONDS = 180 |
| 23 |
| 24 def __init__(self, starting_semaphore, target, args): |
17 """Initializes an instance of a job. | 25 """Initializes an instance of a job. |
18 | 26 |
19 Args: | 27 Args: |
20 starting_semaphore: Semaphore used by caller to wait on such that | 28 starting_semaphore: Semaphore used by caller to wait on such that |
21 there isn't more than a certain number of threads running. Should | 29 there isn't more than a certain number of parallel_jobs running. Should |
22 be initialized to a value for the number of threads wanting to be run | 30 be initialized to a value for the number of parallel_jobs wanting to be |
23 at a time. | 31 run at a time. |
24 ending_semaphore: Semaphore is released every time a job ends. Should be | |
25 initialized to 0 before starting first job. Should be acquired once for | |
26 each job. Threading.Thread.join() has a bug where if the run function | |
27 terminates too quickly join() will hang forever. | |
28 target: The func to run. | 32 target: The func to run. |
29 args: Args to pass to the fun. | 33 args: Args to pass to the fun. |
30 """ | 34 """ |
31 threading.Thread.__init__(self, target=target, args=args) | 35 super(ParallelJob, self).__init__(target=target, args=args) |
32 self._target = target | 36 self._target = target |
33 self._args = args | 37 self._args = args |
34 self._starting_semaphore = starting_semaphore | 38 self._starting_semaphore = starting_semaphore |
35 self._ending_semaphore = ending_semaphore | |
36 self._output = None | |
37 self._completed = False | |
38 | 39 |
39 def run(self): | 40 def run(self): |
40 """Thread override. Runs the method specified and sets output.""" | 41 """Thread override. Runs the method specified and sets output.""" |
41 try: | 42 try: |
42 self._output = self._target(*self._args) | 43 self._target(*self._args) |
43 finally: | 44 finally: |
44 # Our own clean up. | 45 self._starting_semaphore.release() |
45 self._Cleanup() | |
46 self._completed = True | |
47 # From threading.py to avoid a refcycle. | |
48 del self._target, self._args | |
49 | 46 |
50 def GetOutput(self): | 47 @classmethod |
51 """Returns the output of the method run.""" | 48 def WaitUntilJobsComplete(cls, parallel_jobs): |
52 assert self._completed, 'GetOutput called before thread was run.' | 49 """Waits until all parallel_jobs have completed before returning. |
53 return self._output | |
54 | 50 |
55 def _Cleanup(self): | 51 Given an array of parallel_jobs, returns once all parallel_jobs have |
56 """Releases semaphores for a waiting caller.""" | 52 completed or a max timeout is reached. |
57 self._starting_semaphore.release() | 53 |
58 self._ending_semaphore.release() | 54 Raises: |
| 55 ParallelJobTimeoutError: if max timeout is reached. |
| 56 """ |
| 57 def GetCurrentActiveCount(): |
| 58 """Returns the (number of active jobs, first active job).""" |
| 59 active_count = 0 |
| 60 active_job = None |
| 61 for parallel_job in parallel_jobs: |
| 62 if parallel_job.is_alive(): |
| 63 active_count += 1 |
| 64 if not active_job: |
| 65 active_job = parallel_job |
| 66 |
| 67 return (active_count, parallel_job) |
| 68 |
| 69 start_time = time.time() |
| 70 while (time.time() - start_time) < cls.MAX_TIMEOUT_SECONDS: |
| 71 (active_count, active_job) = GetCurrentActiveCount() |
| 72 if active_count == 0: |
| 73 return |
| 74 else: |
| 75 print >> sys.stderr, ( |
| 76 'Process Pool Active: Waiting on %d/%d jobs to complete' % |
| 77 (active_count, len(parallel_jobs))) |
| 78 active_job.join(cls.SLEEP_TIMEOUT_SECONDS) |
| 79 time.sleep(5) # Prevents lots of printing out as job is ending. |
| 80 |
| 81 for parallel_job in parallel_jobs: |
| 82 if parallel_job.is_alive(): |
| 83 parallel_job.terminate() |
| 84 |
| 85 raise ParallelJobTimeoutError('Exceeded max time of %d seconds to wait for ' |
| 86 'job completion.' % cls.MAX_TIMEOUT_SECONDS) |
59 | 87 |
60 def __str__(self): | 88 def __str__(self): |
61 return '%s(%s)' % (self._target, self._args) | 89 return '%s(%s)' % (self._target, self._args) |
62 | 90 |
63 | 91 |
64 def RunParallelJobs(number_of_simultaneous_jobs, jobs, jobs_args, | 92 def RunParallelJobs(number_of_simultaneous_jobs, jobs, jobs_args, |
65 print_status): | 93 print_status): |
66 """Runs set number of specified jobs in parallel. | 94 """Runs set number of specified jobs in parallel. |
67 | 95 |
68 Args: | 96 Args: |
69 number_of_simultaneous_jobs: Max number of threads to be run in parallel. | 97 number_of_simultaneous_jobs: Max number of parallel_jobs to be run in |
| 98 parallel. |
70 jobs: Array of methods to run. | 99 jobs: Array of methods to run. |
71 jobs_args: Array of args associated with method calls. | 100 jobs_args: Array of args associated with method calls. |
72 print_status: True if you'd like this to print out .'s as it runs jobs. | 101 print_status: True if you'd like this to print out .'s as it runs jobs. |
73 Returns: | 102 Returns: |
74 Returns an array of results corresponding to each thread. | 103 Returns an array of results corresponding to each parallel_job. |
75 """ | 104 """ |
76 def _TwoTupleize(x, y): | 105 def ProcessOutputWrapper(func, args, output): |
77 return (x, y) | 106 """Simple function wrapper that puts the output of a function in a queue.""" |
| 107 output.put(func(*args)) |
78 | 108 |
79 threads = [] | |
80 job_start_semaphore = threading.Semaphore(number_of_simultaneous_jobs) | |
81 join_semaphore = threading.Semaphore(0) | |
82 assert len(jobs) == len(jobs_args), 'Length of args array is wrong.' | 109 assert len(jobs) == len(jobs_args), 'Length of args array is wrong.' |
83 | |
84 # Create the parallel jobs. | |
85 for job, args in map(_TwoTupleize, jobs, jobs_args): | |
86 thread = ParallelJob(job_start_semaphore, join_semaphore, target=job, | |
87 args=args) | |
88 threads.append(thread) | |
89 | |
90 # Cache sudo access. | 110 # Cache sudo access. |
91 cros_lib.RunCommand(['sudo', 'echo', 'Caching sudo credentials'], | 111 cros_lib.RunCommand(['sudo', 'echo', 'Caching sudo credentials'], |
92 print_cmd=False, redirect_stdout=True, | 112 print_cmd=False, redirect_stdout=True, |
93 redirect_stderr=True) | 113 redirect_stderr=True) |
94 | 114 |
| 115 parallel_jobs = [] |
| 116 output_array = [] |
| 117 |
| 118 # Semaphore used to create a Process Pool. |
| 119 job_start_semaphore = multiprocessing.Semaphore(number_of_simultaneous_jobs) |
| 120 |
| 121 # Create the parallel jobs. |
| 122 for job, args in map(lambda x, y: (x, y), jobs, jobs_args): |
| 123 output = multiprocessing.Queue() |
| 124 parallel_job = ParallelJob(job_start_semaphore, |
| 125 target=ProcessOutputWrapper, |
| 126 args=(job, args, output)) |
| 127 parallel_jobs.append(parallel_job) |
| 128 output_array.append(output) |
| 129 |
95 # We use a semaphore to ensure we don't run more jobs than required. | 130 # We use a semaphore to ensure we don't run more jobs than required. |
96 # After each thread finishes, it releases (increments semaphore). | 131 # After each parallel_job finishes, it releases (increments semaphore). |
97 # Acquire blocks of num jobs reached and continues when a thread finishes. | 132 for next_parallel_job in parallel_jobs: |
98 for next_thread in threads: | 133 job_start_semaphore.acquire(block=True) |
99 job_start_semaphore.acquire(blocking=True) | 134 next_parallel_job.start() |
100 next_thread.start() | |
101 | 135 |
102 # Wait on the rest of the threads to finish. | 136 ParallelJob.WaitUntilJobsComplete(parallel_jobs) |
103 for thread in threads: | 137 return [output.get() for output in output_array] |
104 while not join_semaphore.acquire(blocking=False): | |
105 time.sleep(5) | |
106 if print_status: | |
107 print >> sys.stderr, '.', | |
108 | |
109 return [thread.GetOutput() for thread in threads] | |
OLD | NEW |