(inqueue, outqueue, initializer=None, initargs=(), maxtasks=None,
wrap_exception=False)
| 95 | |
| 96 | |
| 97 | def worker(inqueue, outqueue, initializer=None, initargs=(), maxtasks=None, |
| 98 | wrap_exception=False): |
| 99 | if (maxtasks is not None) and not (isinstance(maxtasks, int) |
| 100 | and maxtasks >= 1): |
| 101 | raise AssertionError("Maxtasks {!r} is not valid".format(maxtasks)) |
| 102 | put = outqueue.put |
| 103 | get = inqueue.get |
| 104 | if hasattr(inqueue, '_writer'): |
| 105 | inqueue._writer.close() |
| 106 | outqueue._reader.close() |
| 107 | |
| 108 | if initializer is not None: |
| 109 | initializer(*initargs) |
| 110 | |
| 111 | completed = 0 |
| 112 | while maxtasks is None or (maxtasks and completed < maxtasks): |
| 113 | try: |
| 114 | task = get() |
| 115 | except (EOFError, OSError): |
| 116 | util.debug('worker got EOFError or OSError -- exiting') |
| 117 | break |
| 118 | |
| 119 | if task is None: |
| 120 | util.debug('worker got sentinel -- exiting') |
| 121 | break |
| 122 | |
| 123 | job, i, func, args, kwds = task |
| 124 | try: |
| 125 | result = (True, func(*args, **kwds)) |
| 126 | except Exception as e: |
| 127 | if wrap_exception and func is not _helper_reraises_exception: |
| 128 | e = ExceptionWithTraceback(e, e.__traceback__) |
| 129 | result = (False, e) |
| 130 | try: |
| 131 | put((job, i, result)) |
| 132 | except Exception as e: |
| 133 | wrapped = MaybeEncodingError(e, result[1]) |
| 134 | util.debug("Possible encoding error while sending result: %s" % ( |
| 135 | wrapped)) |
| 136 | put((job, i, (False, wrapped))) |
| 137 | |
| 138 | task = job = result = func = args = kwds = None |
| 139 | completed += 1 |
| 140 | util.debug('worker exiting after %d tasks' % completed) |
| 141 | |
| 142 | def _helper_reraises_exception(ex): |
| 143 | 'Pickle-able helper function for use by _guarded_task_generation.' |
searching dependent graphs…