Instantly share code, notes, and snippets.

Embed
What would you like to do?
processify
import os
import sys
import traceback
from functools import wraps
from multiprocessing import Process, Queue
def processify(func):
'''Decorator to run a function as a process.
Be sure that every argument and the return value
is *pickable*.
The created process is joined, so the code does not
run in parallel.
'''
def process_func(q, *args, **kwargs):
try:
ret = func(*args, **kwargs)
except Exception:
ex_type, ex_value, tb = sys.exc_info()
error = ex_type, ex_value, ''.join(traceback.format_tb(tb))
ret = None
else:
error = None
q.put((ret, error))
# register original function with different name
# in sys.modules so it is pickable
process_func.__name__ = func.__name__ + 'processify_func'
setattr(sys.modules[__name__], process_func.__name__, process_func)
@wraps(func)
def wrapper(*args, **kwargs):
q = Queue()
p = Process(target=process_func, args=[q] + list(args), kwargs=kwargs)
p.start()
ret, error = q.get()
p.join()
if error:
ex_type, ex_value, tb_str = error
message = '%s (in subprocess)\n%s' % (ex_value.message, tb_str)
raise ex_type(message)
return ret
return wrapper
@processify
def test_function():
return os.getpid()
@processify
def test_deadlock():
return range(30000)
@processify
def test_exception():
raise RuntimeError('xyz')
def test():
print os.getpid()
print test_function()
print len(test_deadlock())
test_exception()
if __name__ == '__main__':
test()
@hangtwenty

This comment has been minimized.

Show comment
Hide comment
@hangtwenty

hangtwenty Aug 21, 2014

so helpful! thank you!

hangtwenty commented Aug 21, 2014

so helpful! thank you!

@YS-L

This comment has been minimized.

Show comment
Hide comment
@YS-L

YS-L Oct 1, 2015

Nice utility, thanks!

It seems that if the decorated function returns a sufficiently large object, a deadlock can occur at the p.join() line, for example:

@processify
def will_deadlock()
    return range(30000)


if __name__ == '__main__':
    will_deadlock() # deadlocks here

Simply removing the p.join() line solves the problem. The multiprocessing doc mentioned a possible deadlock issue regarding joining processes that use queues, which seems to be relevant here.

YS-L commented Oct 1, 2015

Nice utility, thanks!

It seems that if the decorated function returns a sufficiently large object, a deadlock can occur at the p.join() line, for example:

@processify
def will_deadlock()
    return range(30000)


if __name__ == '__main__':
    will_deadlock() # deadlocks here

Simply removing the p.join() line solves the problem. The multiprocessing doc mentioned a possible deadlock issue regarding joining processes that use queues, which seems to be relevant here.

@dgerosa

This comment has been minimized.

Show comment
Hide comment
@dgerosa

dgerosa May 11, 2017

Hello,

this is incredibly useful! Thanks! You should really upload it to Pypi...

Davide

dgerosa commented May 11, 2017

Hello,

this is incredibly useful! Thanks! You should really upload it to Pypi...

Davide

@karatheodory

This comment has been minimized.

Show comment
Hide comment
@karatheodory

karatheodory May 29, 2017

Thank you for the great work, it is very helpful!

One problem I've had with it is when the original exception type has more parameters than usual (in my case it is sqlalchemy.exc.ProgrammingError, which takes 4 arguments, so I've had TypeError: __init__() takes at least 4 arguments (2 given) instead of the original exception). We can go this way:

        if error:
            ex_type, ex_value, tb_str = error
            message = '%s (in subprocess)\n%s' % (ex_value.message, tb_str)
            try:
                exception = ex_type(message)
            except Exception:
                # Failed to keep the original exception type
                exception = Exception('%s\n(original exception type: %s)' % (message, ex_type))
            raise exception

karatheodory commented May 29, 2017

Thank you for the great work, it is very helpful!

One problem I've had with it is when the original exception type has more parameters than usual (in my case it is sqlalchemy.exc.ProgrammingError, which takes 4 arguments, so I've had TypeError: __init__() takes at least 4 arguments (2 given) instead of the original exception). We can go this way:

        if error:
            ex_type, ex_value, tb_str = error
            message = '%s (in subprocess)\n%s' % (ex_value.message, tb_str)
            try:
                exception = ex_type(message)
            except Exception:
                # Failed to keep the original exception type
                exception = Exception('%s\n(original exception type: %s)' % (message, ex_type))
            raise exception
@samoand

This comment has been minimized.

Show comment
Hide comment
@samoand

samoand Aug 26, 2017

Can you please add a license to this code? I was putting together something like that, with splitting input and mapping it to pool of processes. It would be very helpful to use your code as a base. Thanks.

samoand commented Aug 26, 2017

Can you please add a license to this code? I was putting together something like that, with splitting input and mapping it to pool of processes. It would be very helpful to use your code as a base. Thanks.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment