paulb@40 | 1 | #!/usr/bin/env python |
paulb@40 | 2 | |
paulb@40 | 3 | """ |
paulb@40 | 4 | A simple parallel processing API for Python, inspired somewhat by the thread |
paulb@40 | 5 | module, slightly less by pypar, and slightly less still by pypvm. |
paulb@40 | 6 | |
paulb@67 | 7 | Copyright (C) 2005, 2006, 2007 Paul Boddie <paul@boddie.org.uk> |
paulb@41 | 8 | |
paulb@79 | 9 | This program is free software; you can redistribute it and/or modify it under |
paulb@79 | 10 | the terms of the GNU Lesser General Public License as published by the Free |
paulb@79 | 11 | Software Foundation; either version 3 of the License, or (at your option) any |
paulb@79 | 12 | later version. |
paulb@41 | 13 | |
paulb@79 | 14 | This program is distributed in the hope that it will be useful, but WITHOUT |
paulb@79 | 15 | ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS |
paulb@79 | 16 | FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more |
paulb@79 | 17 | details. |
paulb@41 | 18 | |
paulb@79 | 19 | You should have received a copy of the GNU Lesser General Public License along |
paulb@79 | 20 | with this program. If not, see <http://www.gnu.org/licenses/>. |
paulb@40 | 21 | """ |
paulb@40 | 22 | |
paulb@128 | 23 | __version__ = "0.3.1" |
paulb@40 | 24 | |
paulb@40 | 25 | import os |
paulb@40 | 26 | import sys |
paulb@40 | 27 | import select |
paulb@40 | 28 | import socket |
paulb@40 | 29 | |
paulb@40 | 30 | try: |
paulb@40 | 31 | import cPickle as pickle |
paulb@40 | 32 | except ImportError: |
paulb@40 | 33 | import pickle |
paulb@40 | 34 | |
paulb@84 | 35 | # Communications. |
paulb@84 | 36 | |
paulb@40 | 37 | class AcknowledgementError(Exception): |
paulb@40 | 38 | pass |
paulb@40 | 39 | |
paulb@40 | 40 | class Channel: |
paulb@40 | 41 | |
paulb@40 | 42 | "A communications channel." |
paulb@40 | 43 | |
paulb@40 | 44 | def __init__(self, pid, read_pipe, write_pipe): |
paulb@40 | 45 | |
paulb@40 | 46 | """ |
paulb@40 | 47 | Initialise the channel with a process identifier 'pid', a 'read_pipe' |
paulb@40 | 48 | from which messages will be received, and a 'write_pipe' into which |
paulb@40 | 49 | messages will be sent. |
paulb@40 | 50 | """ |
paulb@40 | 51 | |
paulb@40 | 52 | self.pid = pid |
paulb@40 | 53 | self.read_pipe = read_pipe |
paulb@40 | 54 | self.write_pipe = write_pipe |
paulb@40 | 55 | self.closed = 0 |
paulb@40 | 56 | |
paulb@40 | 57 | def __del__(self): |
paulb@40 | 58 | |
paulb@40 | 59 | # Since signals don't work well with I/O, we close pipes and wait for |
paulb@40 | 60 | # created processes upon finalisation. |
paulb@40 | 61 | |
paulb@40 | 62 | self.close() |
paulb@40 | 63 | |
paulb@40 | 64 | def close(self): |
paulb@40 | 65 | |
paulb@40 | 66 | "Explicitly close the channel." |
paulb@40 | 67 | |
paulb@40 | 68 | if not self.closed: |
paulb@40 | 69 | self.closed = 1 |
paulb@40 | 70 | self.read_pipe.close() |
paulb@40 | 71 | self.write_pipe.close() |
paulb@40 | 72 | #self.wait(os.WNOHANG) |
paulb@40 | 73 | |
paulb@40 | 74 | def wait(self, options=0): |
paulb@40 | 75 | |
paulb@40 | 76 | "Wait for the created process, if any, to exit." |
paulb@40 | 77 | |
paulb@40 | 78 | if self.pid != 0: |
paulb@40 | 79 | try: |
paulb@40 | 80 | os.waitpid(self.pid, options) |
paulb@40 | 81 | except OSError: |
paulb@40 | 82 | pass |
paulb@40 | 83 | |
paulb@40 | 84 | def _send(self, obj): |
paulb@40 | 85 | |
paulb@40 | 86 | "Send the given object 'obj' through the channel." |
paulb@40 | 87 | |
paulb@40 | 88 | pickle.dump(obj, self.write_pipe) |
paulb@40 | 89 | self.write_pipe.flush() |
paulb@40 | 90 | |
paulb@40 | 91 | def send(self, obj): |
paulb@40 | 92 | |
paulb@40 | 93 | """ |
paulb@40 | 94 | Send the given object 'obj' through the channel. Then wait for an |
paulb@40 | 95 | acknowledgement. (The acknowledgement makes the caller wait, thus |
paulb@40 | 96 | preventing processes from exiting and disrupting the communications |
paulb@40 | 97 | channel and losing data.) |
paulb@40 | 98 | """ |
paulb@40 | 99 | |
paulb@40 | 100 | self._send(obj) |
paulb@40 | 101 | if self._receive() != "OK": |
paulb@40 | 102 | raise AcknowledgementError, obj |
paulb@40 | 103 | |
paulb@40 | 104 | def _receive(self): |
paulb@40 | 105 | |
paulb@40 | 106 | "Receive an object through the channel, returning the object." |
paulb@40 | 107 | |
paulb@40 | 108 | obj = pickle.load(self.read_pipe) |
paulb@40 | 109 | if isinstance(obj, Exception): |
paulb@40 | 110 | raise obj |
paulb@40 | 111 | else: |
paulb@40 | 112 | return obj |
paulb@40 | 113 | |
paulb@40 | 114 | def receive(self): |
paulb@40 | 115 | |
paulb@40 | 116 | """ |
paulb@40 | 117 | Receive an object through the channel, returning the object. Send an |
paulb@40 | 118 | acknowledgement of receipt. (The acknowledgement makes the sender wait, |
paulb@40 | 119 | thus preventing processes from exiting and disrupting the communications |
paulb@40 | 120 | channel and losing data.) |
paulb@40 | 121 | """ |
paulb@40 | 122 | |
paulb@40 | 123 | try: |
paulb@40 | 124 | obj = self._receive() |
paulb@40 | 125 | return obj |
paulb@40 | 126 | finally: |
paulb@40 | 127 | self._send("OK") |
paulb@40 | 128 | |
paulb@84 | 129 | # Management of processes and communications. |
paulb@84 | 130 | |
paulb@40 | 131 | class Exchange: |
paulb@40 | 132 | |
paulb@40 | 133 | """ |
paulb@40 | 134 | A communications exchange that can be used to detect channels which are |
paulb@67 | 135 | ready to communicate. Subclasses of this class can define the 'store_data' |
paulb@67 | 136 | method in order to enable the 'add_wait', 'wait' and 'finish' methods. |
paulb@40 | 137 | """ |
paulb@40 | 138 | |
paulb@116 | 139 | def __init__(self, channels=None, limit=None, reuse=0, autoclose=1): |
paulb@40 | 140 | |
paulb@40 | 141 | """ |
paulb@67 | 142 | Initialise the exchange with an optional list of 'channels'. |
paulb@67 | 143 | |
paulb@67 | 144 | If the optional 'limit' is specified, restrictions on the addition of |
paulb@67 | 145 | new channels can be enforced and observed through the 'add_wait', 'wait' |
paulb@67 | 146 | and 'finish' methods. To make use of these methods, create a subclass of |
paulb@67 | 147 | this class and define a working 'store_data' method. |
paulb@67 | 148 | |
paulb@116 | 149 | If the optional 'reuse' parameter is set to a true value, channels and |
paulb@116 | 150 | processes will be reused for waiting computations. |
paulb@116 | 151 | |
paulb@67 | 152 | If the optional 'autoclose' parameter is set to a false value, channels |
paulb@67 | 153 | will not be closed automatically when they are removed from the exchange |
paulb@67 | 154 | - by default they are closed when removed. |
paulb@40 | 155 | """ |
paulb@40 | 156 | |
paulb@67 | 157 | self.limit = limit |
paulb@116 | 158 | self.reuse = reuse |
paulb@116 | 159 | self.autoclose = autoclose |
paulb@99 | 160 | self.waiting = [] |
paulb@40 | 161 | self.readables = {} |
paulb@58 | 162 | self.removed = [] |
paulb@40 | 163 | self.poller = select.poll() |
paulb@40 | 164 | for channel in channels or []: |
paulb@40 | 165 | self.add(channel) |
paulb@40 | 166 | |
paulb@40 | 167 | def add(self, channel): |
paulb@40 | 168 | |
paulb@40 | 169 | "Add the given 'channel' to the exchange." |
paulb@40 | 170 | |
paulb@40 | 171 | self.readables[channel.read_pipe.fileno()] = channel |
paulb@40 | 172 | self.poller.register(channel.read_pipe.fileno(), select.POLLIN | select.POLLHUP | select.POLLNVAL | select.POLLERR) |
paulb@40 | 173 | |
paulb@40 | 174 | def active(self): |
paulb@40 | 175 | |
paulb@40 | 176 | "Return a list of active channels." |
paulb@40 | 177 | |
paulb@40 | 178 | return self.readables.values() |
paulb@40 | 179 | |
paulb@40 | 180 | def ready(self, timeout=None): |
paulb@40 | 181 | |
paulb@40 | 182 | """ |
paulb@40 | 183 | Wait for a period of time specified by the optional 'timeout' (or until |
paulb@40 | 184 | communication is possible) and return a list of channels which are ready |
paulb@40 | 185 | to be read from. |
paulb@40 | 186 | """ |
paulb@40 | 187 | |
paulb@40 | 188 | fds = self.poller.poll(timeout) |
paulb@40 | 189 | readables = [] |
paulb@58 | 190 | self.removed = [] |
paulb@58 | 191 | |
paulb@40 | 192 | for fd, status in fds: |
paulb@40 | 193 | channel = self.readables[fd] |
paulb@55 | 194 | removed = 0 |
paulb@40 | 195 | |
paulb@40 | 196 | # Remove ended/error channels. |
paulb@40 | 197 | |
paulb@40 | 198 | if status & (select.POLLHUP | select.POLLNVAL | select.POLLERR): |
paulb@40 | 199 | self.remove(channel) |
paulb@58 | 200 | self.removed.append(channel) |
paulb@55 | 201 | removed = 1 |
paulb@40 | 202 | |
paulb@40 | 203 | # Record readable channels. |
paulb@40 | 204 | |
paulb@55 | 205 | if status & select.POLLIN: |
paulb@55 | 206 | if not (removed and self.autoclose): |
paulb@55 | 207 | readables.append(channel) |
paulb@40 | 208 | |
paulb@40 | 209 | return readables |
paulb@40 | 210 | |
paulb@40 | 211 | def remove(self, channel): |
paulb@40 | 212 | |
paulb@40 | 213 | """ |
paulb@40 | 214 | Remove the given 'channel' from the exchange. |
paulb@40 | 215 | """ |
paulb@40 | 216 | |
paulb@40 | 217 | del self.readables[channel.read_pipe.fileno()] |
paulb@40 | 218 | self.poller.unregister(channel.read_pipe.fileno()) |
paulb@40 | 219 | if self.autoclose: |
paulb@40 | 220 | channel.close() |
paulb@40 | 221 | channel.wait() |
paulb@40 | 222 | |
paulb@67 | 223 | # Enhanced exchange methods involving channel limits. |
paulb@67 | 224 | |
paulb@67 | 225 | def add_wait(self, channel): |
paulb@67 | 226 | |
paulb@67 | 227 | """ |
paulb@67 | 228 | Add the given 'channel' to the exchange, waiting if the limit on active |
paulb@67 | 229 | channels would be exceeded by adding the channel. |
paulb@67 | 230 | """ |
paulb@67 | 231 | |
paulb@67 | 232 | self.wait() |
paulb@67 | 233 | self.add(channel) |
paulb@67 | 234 | |
paulb@67 | 235 | def wait(self): |
paulb@67 | 236 | |
paulb@67 | 237 | """ |
paulb@67 | 238 | Test for the limit on channels, blocking and reading incoming data until |
paulb@67 | 239 | the number of channels is below the limit. |
paulb@67 | 240 | """ |
paulb@67 | 241 | |
paulb@67 | 242 | # If limited, block until channels have been closed. |
paulb@67 | 243 | |
paulb@67 | 244 | while self.limit is not None and len(self.active()) >= self.limit: |
paulb@67 | 245 | self.store() |
paulb@67 | 246 | |
paulb@116 | 247 | def start_waiting(self, channel): |
paulb@99 | 248 | |
paulb@99 | 249 | """ |
paulb@116 | 250 | Start a waiting process given the reception of data on the given |
paulb@116 | 251 | 'channel'. |
paulb@99 | 252 | """ |
paulb@99 | 253 | |
paulb@99 | 254 | if self.waiting: |
paulb@99 | 255 | callable, args, kw = self.waiting.pop() |
paulb@116 | 256 | |
paulb@116 | 257 | # Try and reuse existing channels if possible. |
paulb@116 | 258 | |
paulb@116 | 259 | if self.reuse: |
paulb@119 | 260 | |
paulb@119 | 261 | # Re-add the channel - this may update information related to |
paulb@119 | 262 | # the channel in subclasses. |
paulb@119 | 263 | |
paulb@119 | 264 | self.add(channel) |
paulb@116 | 265 | channel.send((args, kw)) |
paulb@116 | 266 | else: |
paulb@116 | 267 | self.add(start(callable, *args, **kw)) |
paulb@116 | 268 | |
paulb@116 | 269 | # Where channels are being reused, but where no processes are waiting |
paulb@116 | 270 | # any more, send a special value to tell them to quit. |
paulb@116 | 271 | |
paulb@116 | 272 | elif self.reuse: |
paulb@116 | 273 | channel.send(None) |
paulb@99 | 274 | |
paulb@67 | 275 | def finish(self): |
paulb@67 | 276 | |
paulb@67 | 277 | """ |
paulb@67 | 278 | Finish the use of the exchange by waiting for all channels to complete. |
paulb@67 | 279 | """ |
paulb@67 | 280 | |
paulb@67 | 281 | while self.active(): |
paulb@67 | 282 | self.store() |
paulb@67 | 283 | |
paulb@67 | 284 | def store(self): |
paulb@67 | 285 | |
paulb@67 | 286 | "For each ready channel, process the incoming data." |
paulb@67 | 287 | |
paulb@67 | 288 | for channel in self.ready(): |
paulb@67 | 289 | self.store_data(channel) |
paulb@116 | 290 | self.start_waiting(channel) |
paulb@67 | 291 | |
paulb@67 | 292 | def store_data(self, channel): |
paulb@67 | 293 | |
paulb@67 | 294 | """ |
paulb@67 | 295 | Store incoming data from the specified 'channel'. In subclasses of this |
paulb@67 | 296 | class, such data could be stored using instance attributes. |
paulb@67 | 297 | """ |
paulb@67 | 298 | |
paulb@67 | 299 | raise NotImplementedError, "store_data" |
paulb@67 | 300 | |
paulb@79 | 301 | # Convenience methods. |
paulb@79 | 302 | |
paulb@84 | 303 | def start(self, callable, *args, **kw): |
paulb@79 | 304 | |
paulb@79 | 305 | """ |
paulb@79 | 306 | Using pprocess.start, create a new process for the given 'callable' |
paulb@79 | 307 | using any additional arguments provided. Then, monitor the channel |
paulb@79 | 308 | created between this process and the created process. |
paulb@79 | 309 | """ |
paulb@79 | 310 | |
paulb@99 | 311 | if self.limit is not None and len(self.active()) >= self.limit: |
paulb@99 | 312 | self.waiting.insert(0, (callable, args, kw)) |
paulb@99 | 313 | return |
paulb@99 | 314 | |
paulb@84 | 315 | self.add_wait(start(callable, *args, **kw)) |
paulb@84 | 316 | |
paulb@97 | 317 | def create(self): |
paulb@97 | 318 | |
paulb@97 | 319 | """ |
paulb@97 | 320 | Using pprocess.create, create a new process and return the created |
paulb@97 | 321 | communications channel to the created process. In the creating process, |
paulb@97 | 322 | return None - the channel receiving data from the created process will |
paulb@97 | 323 | be automatically managed by this exchange. |
paulb@97 | 324 | """ |
paulb@97 | 325 | |
paulb@97 | 326 | channel = create() |
paulb@97 | 327 | if channel.pid == 0: |
paulb@97 | 328 | return channel |
paulb@97 | 329 | else: |
paulb@97 | 330 | self.add_wait(channel) |
paulb@97 | 331 | return None |
paulb@97 | 332 | |
paulb@84 | 333 | def manage(self, callable): |
paulb@84 | 334 | |
paulb@84 | 335 | """ |
paulb@84 | 336 | Wrap the given 'callable' in an object which can then be called in the |
paulb@84 | 337 | same way as 'callable', but with new processes and communications |
paulb@84 | 338 | managed automatically. |
paulb@84 | 339 | """ |
paulb@84 | 340 | |
paulb@84 | 341 | return ManagedCallable(callable, self) |
paulb@84 | 342 | |
paulb@84 | 343 | class ManagedCallable: |
paulb@84 | 344 | |
paulb@84 | 345 | "A callable managed by an exchange." |
paulb@84 | 346 | |
paulb@84 | 347 | def __init__(self, callable, exchange): |
paulb@84 | 348 | |
paulb@84 | 349 | """ |
paulb@84 | 350 | Wrap the given 'callable', using the given 'exchange' to monitor the |
paulb@84 | 351 | channels created for communications between this and the created |
paulb@94 | 352 | processes. Note that the 'callable' must be parallel-aware (that is, |
paulb@94 | 353 | have a 'channel' parameter). Use the MakeParallel class to wrap other |
paulb@94 | 354 | kinds of callable objects. |
paulb@84 | 355 | """ |
paulb@84 | 356 | |
paulb@84 | 357 | self.callable = callable |
paulb@84 | 358 | self.exchange = exchange |
paulb@84 | 359 | |
paulb@84 | 360 | def __call__(self, *args, **kw): |
paulb@84 | 361 | |
paulb@84 | 362 | "Invoke the callable with the supplied arguments." |
paulb@84 | 363 | |
paulb@84 | 364 | self.exchange.start(self.callable, *args, **kw) |
paulb@84 | 365 | |
paulb@84 | 366 | # Abstractions and utilities. |
paulb@84 | 367 | |
paulb@84 | 368 | class Map(Exchange): |
paulb@84 | 369 | |
paulb@84 | 370 | "An exchange which can be used like the built-in 'map' function." |
paulb@84 | 371 | |
paulb@107 | 372 | def __init__(self, *args, **kw): |
paulb@107 | 373 | Exchange.__init__(self, *args, **kw) |
paulb@107 | 374 | self.init() |
paulb@107 | 375 | |
paulb@107 | 376 | def init(self): |
paulb@107 | 377 | |
paulb@107 | 378 | "Remember the channel addition order to order output." |
paulb@107 | 379 | |
paulb@107 | 380 | self.channel_number = 0 |
paulb@107 | 381 | self.channels = {} |
paulb@107 | 382 | self.results = [] |
paulb@107 | 383 | |
paulb@84 | 384 | def add(self, channel): |
paulb@84 | 385 | |
paulb@84 | 386 | "Add the given 'channel' to the exchange." |
paulb@84 | 387 | |
paulb@84 | 388 | Exchange.add(self, channel) |
paulb@92 | 389 | self.channels[channel] = self.channel_number |
paulb@92 | 390 | self.channel_number += 1 |
paulb@84 | 391 | |
paulb@107 | 392 | def start(self, callable, *args, **kw): |
paulb@107 | 393 | |
paulb@107 | 394 | """ |
paulb@107 | 395 | Using pprocess.start, create a new process for the given 'callable' |
paulb@107 | 396 | using any additional arguments provided. Then, monitor the channel |
paulb@107 | 397 | created between this process and the created process. |
paulb@107 | 398 | """ |
paulb@107 | 399 | |
paulb@107 | 400 | self.results.append(None) # placeholder |
paulb@107 | 401 | Exchange.start(self, callable, *args, **kw) |
paulb@107 | 402 | |
paulb@110 | 403 | def create(self): |
paulb@110 | 404 | |
paulb@110 | 405 | """ |
paulb@110 | 406 | Using pprocess.create, create a new process and return the created |
paulb@110 | 407 | communications channel to the created process. In the creating process, |
paulb@110 | 408 | return None - the channel receiving data from the created process will |
paulb@110 | 409 | be automatically managed by this exchange. |
paulb@110 | 410 | """ |
paulb@110 | 411 | |
paulb@110 | 412 | self.results.append(None) # placeholder |
paulb@110 | 413 | return Exchange.create(self) |
paulb@110 | 414 | |
paulb@84 | 415 | def __call__(self, callable, sequence): |
paulb@84 | 416 | |
paulb@89 | 417 | "Wrap and invoke 'callable' for each element in the 'sequence'." |
paulb@89 | 418 | |
paulb@92 | 419 | if not isinstance(callable, MakeParallel): |
paulb@92 | 420 | wrapped = MakeParallel(callable) |
paulb@92 | 421 | else: |
paulb@92 | 422 | wrapped = callable |
paulb@84 | 423 | |
paulb@107 | 424 | self.init() |
paulb@84 | 425 | |
paulb@107 | 426 | # Start processes for each element in the sequence. |
paulb@84 | 427 | |
paulb@84 | 428 | for i in sequence: |
paulb@92 | 429 | self.start(wrapped, i) |
paulb@107 | 430 | |
paulb@107 | 431 | # Access to the results occurs through this object. |
paulb@107 | 432 | |
paulb@107 | 433 | return self |
paulb@84 | 434 | |
paulb@107 | 435 | def __getitem__(self, i): |
paulb@107 | 436 | self.finish() |
paulb@107 | 437 | return self.results[i] |
paulb@84 | 438 | |
paulb@107 | 439 | def __iter__(self): |
paulb@107 | 440 | self.finish() |
paulb@107 | 441 | return iter(self.results) |
paulb@84 | 442 | |
paulb@84 | 443 | def store_data(self, channel): |
paulb@84 | 444 | |
paulb@84 | 445 | "Accumulate the incoming data, associating results with channels." |
paulb@84 | 446 | |
paulb@84 | 447 | data = channel.receive() |
paulb@92 | 448 | self.results[self.channels[channel]] = data |
paulb@92 | 449 | del self.channels[channel] |
paulb@84 | 450 | |
paulb@97 | 451 | class Queue(Exchange): |
paulb@97 | 452 | |
paulb@97 | 453 | """ |
paulb@97 | 454 | An exchange acting as a queue, making data from created processes available |
paulb@97 | 455 | in the order in which it is received. |
paulb@97 | 456 | """ |
paulb@97 | 457 | |
paulb@97 | 458 | def __init__(self, *args, **kw): |
paulb@97 | 459 | Exchange.__init__(self, *args, **kw) |
paulb@97 | 460 | self.queue = [] |
paulb@97 | 461 | |
paulb@97 | 462 | def store_data(self, channel): |
paulb@97 | 463 | |
paulb@97 | 464 | "Accumulate the incoming data, associating results with channels." |
paulb@97 | 465 | |
paulb@97 | 466 | data = channel.receive() |
paulb@97 | 467 | self.queue.insert(0, data) |
paulb@97 | 468 | |
paulb@97 | 469 | def __iter__(self): |
paulb@97 | 470 | return self |
paulb@97 | 471 | |
paulb@97 | 472 | def next(self): |
paulb@97 | 473 | |
paulb@97 | 474 | "Return the next element in the queue." |
paulb@97 | 475 | |
paulb@97 | 476 | if self.queue: |
paulb@97 | 477 | return self.queue.pop() |
paulb@97 | 478 | while self.active(): |
paulb@97 | 479 | self.store() |
paulb@97 | 480 | if self.queue: |
paulb@97 | 481 | return self.queue.pop() |
paulb@97 | 482 | else: |
paulb@97 | 483 | raise StopIteration |
paulb@97 | 484 | |
paulb@84 | 485 | class MakeParallel: |
paulb@84 | 486 | |
paulb@84 | 487 | "A wrapper around functions making them able to communicate results." |
paulb@84 | 488 | |
paulb@84 | 489 | def __init__(self, callable): |
paulb@84 | 490 | |
paulb@94 | 491 | """ |
paulb@94 | 492 | Initialise the wrapper with the given 'callable'. This object will then |
paulb@94 | 493 | be able to accept a 'channel' parameter when invoked, and to forward the |
paulb@94 | 494 | result of the given 'callable' via the channel provided back to the |
paulb@94 | 495 | invoking process. |
paulb@94 | 496 | """ |
paulb@84 | 497 | |
paulb@84 | 498 | self.callable = callable |
paulb@84 | 499 | |
paulb@84 | 500 | def __call__(self, channel, *args, **kw): |
paulb@84 | 501 | |
paulb@84 | 502 | "Invoke the callable and return its result via the given 'channel'." |
paulb@84 | 503 | |
paulb@84 | 504 | channel.send(self.callable(*args, **kw)) |
paulb@84 | 505 | |
paulb@119 | 506 | class MakeReusable(MakeParallel): |
paulb@119 | 507 | |
paulb@119 | 508 | """ |
paulb@119 | 509 | A wrapper around functions making them able to communicate results in a |
paulb@119 | 510 | reusable fashion. |
paulb@119 | 511 | """ |
paulb@119 | 512 | |
paulb@119 | 513 | def __call__(self, channel, *args, **kw): |
paulb@119 | 514 | |
paulb@119 | 515 | "Invoke the callable and return its result via the given 'channel'." |
paulb@119 | 516 | |
paulb@119 | 517 | channel.send(self.callable(*args, **kw)) |
paulb@119 | 518 | t = channel.receive() |
paulb@119 | 519 | while t is not None: |
paulb@119 | 520 | args, kw = t |
paulb@119 | 521 | channel.send(self.callable(*args, **kw)) |
paulb@119 | 522 | t = channel.receive() |
paulb@119 | 523 | |
paulb@84 | 524 | # Utility functions. |
paulb@79 | 525 | |
paulb@40 | 526 | def create(): |
paulb@40 | 527 | |
paulb@40 | 528 | """ |
paulb@40 | 529 | Create a new process, returning a communications channel to both the |
paulb@40 | 530 | creating process and the created process. |
paulb@40 | 531 | """ |
paulb@40 | 532 | |
paulb@40 | 533 | parent, child = socket.socketpair() |
paulb@40 | 534 | for s in [parent, child]: |
paulb@40 | 535 | s.setblocking(1) |
paulb@40 | 536 | |
paulb@40 | 537 | pid = os.fork() |
paulb@40 | 538 | if pid == 0: |
paulb@40 | 539 | parent.close() |
paulb@73 | 540 | return Channel(pid, child.makefile("r", 0), child.makefile("w", 0)) |
paulb@40 | 541 | else: |
paulb@40 | 542 | child.close() |
paulb@73 | 543 | return Channel(pid, parent.makefile("r", 0), parent.makefile("w", 0)) |
paulb@40 | 544 | |
paulb@97 | 545 | def exit(channel): |
paulb@97 | 546 | |
paulb@97 | 547 | """ |
paulb@97 | 548 | Terminate a created process, closing the given 'channel'. |
paulb@97 | 549 | """ |
paulb@97 | 550 | |
paulb@97 | 551 | channel.close() |
paulb@97 | 552 | os._exit(0) |
paulb@97 | 553 | |
paulb@84 | 554 | def start(callable, *args, **kw): |
paulb@40 | 555 | |
paulb@40 | 556 | """ |
paulb@40 | 557 | Create a new process which shall start running in the given 'callable'. |
paulb@94 | 558 | Additional arguments to the 'callable' can be given as additional arguments |
paulb@94 | 559 | to this function. |
paulb@94 | 560 | |
paulb@94 | 561 | Return a communications channel to the creating process. For the created |
paulb@94 | 562 | process, supply a channel as the 'channel' parameter in the given 'callable' |
paulb@94 | 563 | so that it may send data back to the creating process. |
paulb@40 | 564 | """ |
paulb@40 | 565 | |
paulb@40 | 566 | channel = create() |
paulb@40 | 567 | if channel.pid == 0: |
paulb@40 | 568 | try: |
paulb@40 | 569 | try: |
paulb@84 | 570 | callable(channel, *args, **kw) |
paulb@40 | 571 | except: |
paulb@40 | 572 | exc_type, exc_value, exc_traceback = sys.exc_info() |
paulb@40 | 573 | channel.send(exc_value) |
paulb@40 | 574 | finally: |
paulb@99 | 575 | exit(channel) |
paulb@40 | 576 | else: |
paulb@40 | 577 | return channel |
paulb@40 | 578 | |
paulb@40 | 579 | def waitall(): |
paulb@40 | 580 | |
paulb@40 | 581 | "Wait for all created processes to terminate." |
paulb@40 | 582 | |
paulb@40 | 583 | try: |
paulb@40 | 584 | while 1: |
paulb@40 | 585 | os.wait() |
paulb@40 | 586 | except OSError: |
paulb@40 | 587 | pass |
paulb@40 | 588 | |
paulb@89 | 589 | def pmap(callable, sequence, limit=None): |
paulb@84 | 590 | |
paulb@89 | 591 | """ |
paulb@89 | 592 | A parallel version of the built-in map function with an optional process |
paulb@94 | 593 | 'limit'. The given 'callable' should not be parallel-aware (that is, have a |
paulb@94 | 594 | 'channel' parameter) since it will be wrapped for parallel communications |
paulb@94 | 595 | before being invoked. |
paulb@94 | 596 | |
paulb@94 | 597 | Return the processed 'sequence' where each element in the sequence is |
paulb@94 | 598 | processed by a different process. |
paulb@89 | 599 | """ |
paulb@84 | 600 | |
paulb@89 | 601 | mymap = Map(limit=limit) |
paulb@84 | 602 | return mymap(callable, sequence) |
paulb@84 | 603 | |
paulb@40 | 604 | # vim: tabstop=4 expandtab shiftwidth=4 |