multiprocessing.pool: Properly handle encoding errors, so pickling er… · ask/celery@eaa4d5d (original) (raw)
`@@ -55,6 +55,21 @@ def soft_timeout_sighandler(signum, frame):
`
55
55
`raise SoftTimeLimitExceeded()
`
56
56
``
57
57
``
``
58
`+
class MaybeEncodingError(Exception):
`
``
59
`+
"""Wraps unpickleable object."""
`
``
60
+
``
61
`+
def init(self, exc, value):
`
``
62
`+
self.exc = str(exc)
`
``
63
`+
self.value = repr(value)
`
``
64
`+
super(MaybeEncodingError, self).init(self.exc, self.value)
`
``
65
+
``
66
`+
def str(self):
`
``
67
`+
return "Error sending result: '%s'. Reason: '%s'." % (self.value,
`
``
68
`+
self.exc)
`
``
69
`+
def repr(self):
`
``
70
`+
return "<MaybeEncodingError: %s>" % str(self)
`
``
71
+
``
72
+
58
73
`def worker(inqueue, outqueue, ackqueue, initializer=None, initargs=(),
`
59
74
`maxtasks=None):
`
60
75
`assert maxtasks is None or (type(maxtasks) == int and maxtasks > 0)
`
`@@ -90,7 +105,13 @@ def worker(inqueue, outqueue, ackqueue, initializer=None, initargs=(),
`
90
105
`result = (True, func(*args, **kwds))
`
91
106
`except Exception, e:
`
92
107
`result = (False, e)
`
93
``
`-
put((job, i, result))
`
``
108
`+
try:
`
``
109
`+
put((job, i, result))
`
``
110
`+
except Exception, exc:
`
``
111
`+
wrapped = MaybeEncodingError(exc, result[1])
`
``
112
`+
debug('Got possible encoding error while sending result: %s' % wrapped)
`
``
113
`+
put((job, i, (False, wrapped)))
`
``
114
+
94
115
`completed += 1
`
95
116
`debug('worker exiting after %d tasks' % completed)
`
96
117
``
`@@ -589,7 +610,7 @@ def imap_unordered(self, func, iterable, chunksize=1):
`
589
610
``
590
611
`def apply_async(self, func, args=(), kwds={},
`
591
612
`callback=None, accept_callback=None, timeout_callback=None,
`
592
``
`-
waitforslot=False):
`
``
613
`+
waitforslot=False, error_callback=None):
`
593
614
`'''
`
594
615
`` Asynchronous equivalent of apply() builtin.
``
595
616
``
`@@ -607,7 +628,8 @@ def apply_async(self, func, args=(), kwds={},
`
607
628
` '''
`
608
629
`assert self._state == RUN
`
609
630
`result = ApplyResult(self._cache, callback,
`
610
``
`-
accept_callback, timeout_callback)
`
``
631
`+
accept_callback, timeout_callback,
`
``
632
`+
error_callback)
`
611
633
`if waitforslot:
`
612
634
`self._putlock.acquire()
`
613
635
`self._taskqueue.put(([(result._job, None, func, args, kwds)], None))
`
`@@ -742,7 +764,7 @@ def _terminate_pool(cls, taskqueue, inqueue, outqueue, ackqueue, pool,
`
742
764
`class ApplyResult(object):
`
743
765
``
744
766
`def init(self, cache, callback, accept_callback=None,
`
745
``
`-
timeout_callback=None):
`
``
767
`+
timeout_callback=None, error_callback=None):
`
746
768
`self._cond = threading.Condition(threading.Lock())
`
747
769
`self._job = job_counter.next()
`
748
770
`self._cache = cache
`
`@@ -751,6 +773,7 @@ def init(self, cache, callback, accept_callback=None,
`
751
773
`self._time_accepted = None
`
752
774
`self._ready = False
`
753
775
`self._callback = callback
`
``
776
`+
self._errback = error_callback
`
754
777
`self._accept_callback = accept_callback
`
755
778
`self._timeout_callback = timeout_callback
`
756
779
`cache[self._job] = self
`
`@@ -786,6 +809,8 @@ def _set(self, i, obj):
`
786
809
`self._success, self._value = obj
`
787
810
`if self._callback and self._success:
`
788
811
`self._callback(self._value)
`
``
812
`+
if self._errback and not self._success:
`
``
813
`+
self._errback(self._value)
`
789
814
`self._cond.acquire()
`
790
815
`try:
`
791
816
`self._ready = True
`