error_callback in multiprocessing.Pool apply_async in Python 2?

Question

Function apply_async of multiprocessing.Pool class has argument error_callback in Python 3. But this argument is missing in Python 2.

Is there any trick to achieve the same functionality in Python 2 ? Ideally I would like to write code which runs in both Python 2 and 3.

Answer 1

TL;DR

Instead of adding extra arguments to return, use the successful() method that AnycResult objects provide you. That's what it's there for.

Explanation

Calls to apply_async return an AsyncResult object. These objects have a method called successful() that will tell you if the target process exited via an exception. Calling successful() will return true or false depending on whether the subprocess exited with an exception. Here is an example:

import multiprocessing

def good_func(arg):
    arg += 1
    return arg

def bad_func(arg):
    arg += 1
    raise Exception # force an exception to be raised
    return arg

pool = multiprocessing.Pool()

good_result = pool.apply_async(good_func, (1,))
bad_result = pool.apply_async(bad_func, (1,))

pool.close()
pool.join()

print(good_result.successful()) # prints True
print(bad_result.successful())  # prints False

This code returns:

True
False

You can then add a conditional expression that calls your cleanup or error handling code if the call to successfull() returns false.

Furthermore, if you need the traceback, you can wrap the body of your subprocess function in a try/except block and return a string version of the traceback in the event of an exception. It could look something like this version of bad_func:

import sys
import traceback
import multiprocessing

def good_func(arg):
    arg += 1
    return arg

def bad_func(arg):
    try:
        arg += 1
        raise Exception
        return a
    except Exception as error:
        # capture the exception and bundle the traceback
        # in a string, then raise new exception with the string traceback
        raise Exception("".join(traceback.format_exception(*sys.exc_info())))

pool = multiprocessing.Pool()

good_result = pool.apply_async(good_func, (1,))
bad_result = pool.apply_async(bad_func, (1,))

pool.close()
pool.join()

print(good_result.successful()) # prints True
print(good_result.get())        # prints good result
print(bad_result.successful())  # prints False
print(bad_result.get())         # prints traceback

This code produces this output:

True
2
False
Traceback (most recent call last):
  File "async_err.py", line 29, in
    print(bad_result.get())
  File "/user/peteoss/encap/Python-2.7.6/lib/python2.7/multiprocessing /pool.py", line 554, in get
    raise self._value
Exception: Traceback (most recent call last):
  File "async_err.py", line 13, in bad_func
    raise Exception
Exception

You should note a few things:

1. You will get two tracebacks: one from the failed called to get(), and the second one is the traceback that you bundled into a string in your subprocess.

2. If you wrap the subprocess function in a try/except block, you will need to reraise an exception when you handle it, otherwise get() will return false.

3. The AsyncResult.successful() and AsyncResult.get() methods are available in all versions of CPython that have the multiprocessing library, so this method meets your requirement of writing version-agnostic code.

Answer 2

I haven't tried python3 yet. But for me, to catch the errors in the child process, I put the function that runs in child process within a

import traceback


try:
    your code that can make error
except Exception as e:
    print e
    return False, traceback.format_exc()
else:
    return True, result

So that I will know if something goes wrong.

EDIT: I change the return format as OP's comment so that the child process returns a tuple (is_success, result or error traceback message )

So that main process will first read the flag is_success and then handles the second argument accordingly.