1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
|
"""SCons.Job
This module defines the Serial and Parallel classes that execute tasks to
complete a build. The Jobs class provides a higher level interface to start,
stop, and wait on jobs.
"""
__revision__ = "__FILE__ __REVISION__ __DATE__ __DEVELOPER__"
class Jobs:
"""An instance of this class initializes N jobs, and provides
methods for starting, stopping, and waiting on all N jobs.
"""
def __init__(self, num, taskmaster):
"""
create 'num' jobs using the given taskmaster.
If 'num' is equal to 0, then a serial job will be used,
otherwise 'num' parallel jobs will be used.
"""
if num > 1:
self.jobs = []
for i in range(num):
self.jobs.append(Parallel(taskmaster, self))
else:
self.jobs = [Serial(taskmaster)]
def start(self):
"""start the jobs"""
for job in self.jobs:
job.start()
def wait(self):
""" wait for the jobs started with start() to finish"""
for job in self.jobs:
job.wait()
def stop(self):
"""
stop the jobs started with start()
This function does not wait for the jobs to finish.
"""
for job in self.jobs:
job.stop()
class Serial:
"""This class is used to execute tasks in series, and is more efficient
than Parallel, but is only appropriate for non-parallel builds. Only
one instance of this class should be in existence at a time.
This class is not thread safe.
"""
def __init__(self, taskmaster):
"""Create a new serial job given a taskmaster.
The taskmaster's next_task() method should return the next task
that needs to be executed, or None if there are no more tasks. The
taskmaster's executed() method will be called for each task when it
is successfully executed or failed() will be called if it failed to
execute (e.g. execute() raised an exception). The taskmaster's
is_blocked() method will not be called. """
self.taskmaster = taskmaster
def start(self):
"""Start the job. This will begin pulling tasks from the taskmaster
and executing them, and return when there are no more tasks. If a task
fails to execute (i.e. execute() raises an exception), then the job will
stop."""
while 1:
task = self.taskmaster.next_task()
if task is None:
break
try:
task.execute()
except:
# Let the failed() callback function arrange for the
# build to stop if that's appropriate.
self.taskmaster.failed(task)
else:
self.taskmaster.executed(task)
def stop(self):
"""Serial jobs are always finished when start() returns, so there
is nothing to do here"""
pass
def wait(self):
"""Serial jobs are always finished when start() returns, so there
is nothing to do here"""
pass
# The will hold a condition variable once the first parallel task
# is created.
cv = None
class Parallel:
"""This class is used to execute tasks in parallel, and is less
efficient than Serial, but is appropriate for parallel builds. Create
an instance of this class for each job or thread you want.
This class is thread safe.
"""
def __init__(self, taskmaster, jobs):
"""Create a new parallel job given a taskmaster, and a Jobs instance.
Multiple jobs will be using the taskmaster in parallel, but all
method calls to taskmaster methods are serialized by the jobs
themselves.
The taskmaster's next_task() method should return the next task
that needs to be executed, or None if there are no more tasks. The
taskmaster's executed() method will be called for each task when it
is successfully executed or failed() will be called if the task
failed to execute (i.e. execute() raised an exception). The
taskmaster's is_blocked() method should return true iff there are
more tasks, but they can't be executed until one or more other
tasks have been executed. next_task() will be called iff
is_blocked() returned false.
Note: calls to taskmaster are serialized, but calls to execute() on
distinct tasks are not serialized, because that is the whole point
of parallel jobs: they can execute multiple tasks
simultaneously. """
global cv
# import threading here so that everything in the Job module
# but the Parallel class will work if the interpreter doesn't
# support threads
import threading
self.taskmaster = taskmaster
self.jobs = jobs
self.thread = threading.Thread(None, self.__run)
self.stop_running = 0
if cv is None:
cv = threading.Condition()
def start(self):
"""Start the job. This will spawn a thread that will begin pulling
tasks from the task master and executing them. This method returns
immediately and doesn't wait for the jobs to be executed.
If a task fails to execute (i.e. execute() raises an exception),
all jobs will be stopped.
To stop the job, call stop().
To wait for the job to finish, call wait().
"""
self.thread.start()
def stop(self):
"""Stop the job. This will cause the job to finish after the
currently executing task is done. A job that has been stopped can
not be restarted.
To wait for the job to finish, call wait().
"""
cv.acquire()
self.stop_running = 1
# wake up the sleeping jobs so this job will end as soon as possible:
cv.notifyAll()
cv.release()
def wait(self):
"""Wait for the job to finish. A job is finished when either there
are no more tasks or the job has been stopped and it is no longer
executing a task.
This method should only be called after start() has been called.
To stop the job, call stop().
"""
self.thread.join()
def __run(self):
"""private method that actually executes the tasks"""
cv.acquire()
try:
while 1:
while self.taskmaster.is_blocked() and not self.stop_running:
cv.wait(None)
# check this before calling next_task(), because
# this job may have been stopped because of a build
# failure:
if self.stop_running:
break
task = self.taskmaster.next_task()
if task == None:
break
cv.release()
try:
try:
task.execute()
finally:
cv.acquire()
except:
# Let the failed() callback function arrange for
# calling self.jobs.stop() to to stop the build
# if that's appropriate.
self.taskmaster.failed(task)
else:
self.taskmaster.executed(task)
if not self.taskmaster.is_blocked():
cv.notifyAll()
finally:
cv.release()
|
