Return to Answer

import multiprocessing as mp import multiprocessing.queues as mpq import functools import dill from typing import Tuple, Callable, Dict, Optional, Iterable, List class TimeoutError(Exception): def __init__(self, func, timeout): self.t = timeout self.fname = func.__name__ def __str__(self): return f"function '{self.fname}' timed out after {self.t}s" def _lemmiwinks(func: Callable, args: Tuple[object], kwargs: Dict[str, object], q: mp.Queue): """lemmiwinks crawls into the unknown""" q.put(dill.loads(func)(*args, **kwargs)) def killer_call(func: Callable = None, timeout: int = 10) -> Callable: """ Single function call with a timeout Args: func: the function timeout: The timeout in seconds """ if not isinstance(timeout, int): raise ValueError(f'timeout needs to be an int. Got: {timeout}') if func is None: return functools.partial(killer_call, timeout=timeout) @functools.wraps(killer_call) def _inners(*args, **kwargs) -> object: q_worker = mp.Queue() proc = mp.Process(target=_lemmiwinks, args=(dill.dumps(func), args, kwargs, q_worker)) proc.start() try: return q_worker.get(timeout=timeout) except mpq.Empty: raise TimeoutError(func, timeout) finally: try: proc.terminate() except: pass return _inners if __name__ == '__main__': @killer_call(timeout=4) def bar(x): import time time.sleep(x) return x print(bar(2)) bar(10) 

import multiprocessing as mp import multiprocessing.queues as mpq import functools import dill from typing import Tuple, Callable, Dict, Optional, Iterable, List, Any class TimeoutError(Exception): def __init__(self, func: Callable, timeout: int): self.t = timeout self.fname = func.__name__ def __str__(self): return f"function '{self.fname}' timed out after {self.t}s" def _lemmiwinks(func: Callable, args: Tuple, kwargs: Dict[str, Any], q: mp.Queue): """lemmiwinks crawls into the unknown""" q.put(dill.loads(func)(*args, **kwargs)) def killer_call(func: Callable = None, timeout: int = 10) -> Callable: """ Single function call with a timeout Args: func: the function timeout: The timeout in seconds """ if not isinstance(timeout, int): raise ValueError(f'timeout needs to be an int. Got: {timeout}') if func is None: return functools.partial(killer_call, timeout=timeout) @functools.wraps(killer_call) def _inners(*args, **kwargs) -> Any: q_worker = mp.Queue() proc = mp.Process(target=_lemmiwinks, args=(dill.dumps(func), args, kwargs, q_worker)) proc.start() try: return q_worker.get(timeout=timeout) except mpq.Empty: raise TimeoutError(func, timeout) finally: try: proc.terminate() except: pass return _inners if __name__ == '__main__': @killer_call(timeout=4) def bar(x): import time time.sleep(x) return x print(bar(2)) bar(10) 

• Uses dill No need for that if __name__ == '__main__' rubbish
• Raises TimeoutError uses exceptions to alert on timeout - can easily be modified
• Cross Platform: Windows & Mac OS X
• Compatibility: Python 3.6+ (I also tested on python 2.7 and it works with small syntax adjustments)

• Raises TimeoutError uses exceptions to alert on timeout - can easily be modified
• Cross Platform: Windows & Mac OS X
• Compatibility: Python 3.6+ (I also tested on python 2.7 and it works with small syntax adjustments)

For full explanation and extension to parallel maps, see here https://flipdazed.github.io/blog/quant%20dev/parallel-functions-with-timeouts