Python Executer that kills processes after a timeout

Question

Imagine you want to run a bunch of tasks in parallel, but you can't be sure that all of your tasks will eventually terminate or rather decide to eat up all your memory and CPU time. This should be easy, right? Create a new thread, run your task, and kill the thread if it doesn't terminate in a given time slot.

Well, here's how Python fails to handle this:

• we can't use threads in the first place, because due to the Global Interpreter Lock our tasks won't actually run in parallel, so we won't benefit from multicore CPUs at all. Also, we can't terminate threads in Python, so there's that.
• multiprocessing: Apparently the way to parallelize stuff in Python is to fork your main process and to handle all the inter-process communication by yourself. Also, communication works by serializing objects using pickle, which comes with a few minor limitations like the inability to serialize functions that are not defined at the top level of a module. But at least we can terminate a process that is running havoc.
• Process Pools solve a lot of the hassle that comes with inter-process communication, but there is no function that allows us to define a timeout. Bummer.
• Executors are an option, especially the ProcessPoolExecutor. It has a map() function that allows us to define a timeout after which a task is skipped. skipped doesn't mean that the underlying process gets killed. In fact, the process keeps running until it terminates by itself, which may be never.

So finally I decided to build something that actually solves my problem. The interface is quite similar to Python's Executor class, but it ensures that processes are actually killed after a timeout, at the cost of forking a process for each function call. This means you shouldn't use this class for lightweight tasks or the overhead will considerably slow you down.

import os
from concurrent.futures import ThreadPoolExecutor
from multiprocessing import Manager, Process
from typing import Callable, Iterable, Dict, Any


class ProcessKillingExecutor:
    """
    The ProcessKillingExecutor works like an `Executor <https://docs.python.org/dev/library/concurrent.futures.html#executor-objects>`_
    in that it uses a bunch of processes to execute calls to a function with different arguments asynchronously.

    But other than the `ProcessPoolExecutor <https://docs.python.org/dev/library/concurrent.futures.html#concurrent.futures.ProcessPoolExecutor>`_,
    the ProcessKillingExecutor forks a new Process for each function call that terminates after the function returns or
    if a timeout occurs.

    This means that contrary to the Executors and similar classes provided by the Python Standard Library, you can
    rely on the fact that a process will get killed if a timeout occurs and that absolutely no side can occur between
    function calls.

    Note that descendant processes of each process will not be terminated – they will simply become orphaned.
    """

    def __init__(self, max_workers: int = None):
        """
        Initializes a new ProcessKillingExecutor instance.
        :param max_workers: The maximum number of processes that can be used to execute the given calls.
        """
        super().__init__()
        self.processes = max_workers or os.cpu_count()
        self.manager = Manager()

    def map(self, func: Callable, iterable: Iterable, timeout: float = None, callback_timeout: Callable = None,
            daemon: bool = True):
        """
        Returns an iterator (actually, a generator) equivalent to map(fn, iter).
        :param func: the function to execute
        :param iterable: an iterable of function arguments
        :param timeout: after this time, the process executing the function will be killed if it did not finish
        :param callback_timeout: this function will be called, if the task times out. It gets the same arguments as
                                 the original function
        :param daemon: run the child process as daemon
        :return: An iterator equivalent to: map(func, *iterables) but the calls may be evaluated out-of-order.
        """
        executor = ThreadPoolExecutor(max_workers=self.processes)
        params = ({'func': func, 'args': args, 'timeout': timeout, 'callback_timeout': callback_timeout,
                   'daemon': daemon} for args in iterable)
        return executor.map(self._submit_unpack_kwargs, params)

    def _submit_unpack_kwargs(self, kwargs: Dict):
        """unpack the kwargs and call submit"""
        return self.submit(**kwargs)

    def submit(self, func: Callable = None, args: Any = (), kwargs: Dict = {}, timeout: float = None,
               callback_timeout: Callable[[Any], Any] = None, daemon: bool = True):
        """
        Submits a callable to be executed with the given arguments.
        Schedules the callable to be executed as func(*args, **kwargs) in a new process.
        Returns the result, if the process finished successfully, or None, if it fails or a timeout occurs.
        :param func: the function to execute
        :param args: the arguments to pass to the function. Can be one argument or a tuple of multiple args.
        :param kwargs: the kwargs to pass to the function
        :param timeout: after this time, the process executing the function will be killed if it did not finish
        :param callback_timeout: this function will be called with the same arguments, if the task times out.
        :param daemon: run the child process as daemon
        :return: the result of the function, or None if the process failed or timed out
        """
        args = args if isinstance(args, tuple) else (args,)
        shared_dict = self.manager.dict()
        process_kwargs = {'func': func, 'args': args, 'kwargs': kwargs, 'share': shared_dict}
        p = Process(target=self._process_run, kwargs=process_kwargs, daemon=daemon)
        p.start()
        p.join(timeout=timeout)
        if 'return' in shared_dict:
            return shared_dict['return']
        else:
            if callback_timeout:
                callback_timeout(*args, **kwargs)
            if p.is_alive():
                p.terminate()
            return None

    @staticmethod
    def _process_run(func: Callable[[Any], Any] = None, args: Any = (), kwargs: Dict = {}, share: Dict = None):
        """
        Executes the specified function as func(*args, **kwargs).
        The result will be stored in the shared dictionary
        :param func: the function to execute
        :param args: the arguments to pass to the function
        :param kwargs: the kwargs to pass to the function
        :param share: a dictionary created using Manager.dict()
        """
        result = func(*args, **kwargs)
        share['return'] = result

Trivial test case:

# due to serialization issues, this function must be defined at the module level
def some_task(n):
    import time
    time.sleep(n/4)
    return n ** 2

if __name__ == "__main__":
    def fun_timeout(n):
        print('timeout:', n)
    executor = ProcessKillingExecutor(max_workers=2)
    generator = executor.map(some_task, range(10), timeout=2, callback_timeout=fun_timeout)
    for elem in generator:
        print(elem)

I'm using this class to parse lots of PDFs using pdfminer. The problem is, that for some documents, pdfminer won't terminate, therefore the need to reliably kill processes after a while.

What do you think? Is it a good solution for the problem?

Answer 1

The obligatory other solution solving your life:

def timer(f):
    def wrapper(job_args, *args, **kwargs):
        fn_args, timeout, timeout_callback = job_args[:3]
        q = Queue()
        p = Process(target=f, args=(q, fn_args), kwargs=kwargs)
        p.start()
        p.join(timeout=timeout)
        p.terminate()
        p.join()
        if not q.empty():
            return q.get()
        return timeout_callback(fn_args, args, kwargs)
    return wrapper


@timer
def job(q, file, *args, **kwargs):
    sleep(3)
    print(file, getpid())
    q.put(file+"_done")


def timeout_callback(*args, **kwargs):
    print("Timeout")


def main():
    timeout = 2
    data = ["file1", "file2", "file3", "file4", "file5"]
    tp = ThreadPoolExecutor(2)

    data = [(x, timeout, timeout_callback) for x in data]
    for got in tp.map(job, data):
       print(got)

---

In your init

def __init__(self, max_workers: int = None):
        ...
        super().__init__()

but you don't have a super class in

class ProcessKillingExecutor:

you could add

class ProcessKillingExecutor(object):

for clarity, else it looks as if you're calling a the super of a base class.

---

The manager might be unnecessary; you are only ever transferring one value. What you are looking for might be a Queue.

---

You are only allowed 80 characters, because it becomes more readable

"""
    The ProcessKillingExecutor works like an `Executor <https://docs.python.org/dev/library/concurrent.futures.html#executor-objects>`_
    in that it uses a bunch of processes to execute calls to a function with different arguments asynchronously.

    But other than the `ProcessPoolExecutor <https://docs.python.org/dev/library/concurrent.futures.html#concurrent.futures.ProcessPoolExecutor>`_,
    the ProcessKillingExecutor forks a new Process for each function call that terminates after the function returns or
    if a timeout occurs.

    This means that contrary to the Executors and similar classes provided by the Python Standard Library, you can
    rely on the fact that a process will get killed if a timeout occurs and that absolutely no side can occur between
    function calls.

    Note that descendant processes of each process will not be terminated – they will simply become orphaned.
    """

do it like this.

"""
    The ProcessKillingExecutor works like an `Executor
    <https://docs.python.org/dev/library/concurrent.futures.html#executor-objects>`_
    in that it uses a bunch of processes to execute calls to a function with
    different arguments asynchronously.

    But other than the `ProcessPoolExecutor
    <https://docs.python.org/dev/library/concurrent.futures.html#concurrent.futures.ProcessPoolExecutor>`_,
    the ProcessKillingExecutor forks a new Process for each function call that
    terminates after the function returns or if a timeout occurs.

    This means that contrary to the Executors and similar classes provided by
    the Python Standard Library, you can rely on the fact that a process will
    get killed if a timeout occurs and that absolutely no side can occur
    between function calls.

    Note that descendant processes of each process will not be terminated –
    they will simply become orphaned.
    """

The exemptions being urls, or in other words, it's good pratice to not line break urls.

---

Your Annotations is slightly off

def submit(self, func: Callable = None, args: Any = (), kwargs: Dict = {}, timeout: float = None,
               callback_timeout: Callable[[Any], Any] = None, daemon: bool = True):

this should be like this with no whitespace and considering that the parameters will be there:

def submit(self,
           func: Callable,
           fn_args: Any,
           p_kwargs: Dict,
           timeout: float,
           callback_timeout: Callable[[Any], Any],
           daemon: bool):

---

Encapsulation: You are not working with kwargs and args, so don't name them as such, in the same sense that you should not name your variable i; it is very confusing.

You are dealing with three different args and kwargs, namely the job, the executing process and the classes, and args and kwargs in the class, should belong to the class if needed or not.

params = ({'func': func, 'fn_args': p_args, "p_kwargs": {},
                   'timeout': timeout, 'callback_timeout': callback_timeout,
                   'daemon': daemon} for p_args in iterable)

---

Terminating and joining the processes of a program.

if p.is_alive():
        p.terminate()
        p.join()

Terminating and joining is not the same thing. Check out your task-manager, you should see the processes staying as "zombies". The call terminate() forces the exit of the process and join() does something I don't pretend to understand, but I know that if you don't call join on terminated processes, you will get zombies.

---

It's confusing that you use a manager; I get the impression that you are going to use the result to handle interchange of data between but processes, this is not the case. You are just retrieving a result from from a child process.

It is dangerous because you are always overwriting the same key "result" in the manager dict. Think of it. Every time you return from the manager.dict, there can be leakage - i.e. if it is overwritten before you return, the wrong result is returned.

You should use a Queue that is unique to every

p.start()
p.join(timeout=timeout)
if not queue.empty():
    return queue.get()
if callback_timeout:
    callback_timeout(*p_args, **p_kwargs)
if p.is_alive():
    p.terminate()
    p.join()

and

@staticmethod
def _process_run(queue: Queue, func: Callable[[Any], Any]=None, 
                 *args, **kwargs):
    """
    Executes the specified function as func(*args, **kwargs).
    The result will be stored in the shared dictionary
    :param func: the function to execute
    :param queue: a Queue
    """
    queue.put(func(*args, **kwargs))

---

This

    return None

is also unnecessary because, None is always returned from a python function that does not return anything.

---

Your imports

import os
from concurrent.futures import ThreadPoolExecutor
from multiprocessing import Manager, Process
from typing import Callable, Iterable, Dict, Any

should be

import os
from concurrent.futures import ThreadPoolExecutor
from multiprocessing import Process
from multiprocessing import Queue
from typing import Callable
from typing import Iterable
from typing import Dict
from typing import Any

as per convention.

---

I ended up with this, and I don't claim that it is any better then yours. But I'll show it anyway for completeness.

import os
from concurrent.futures import ThreadPoolExecutor
from multiprocessing import Process
from multiprocessing import Queue
from typing import Callable
from typing import Iterable
from typing import Dict
from typing import Any


class ProcessKillingExecutor:
    """
    The ProcessKillingExecutor works like an `Executor
    <https://docs.python.org/dev/library/concurrent.futures.html#executor-objects>`_
    in that it uses a bunch of processes to execute calls to a function with
    different arguments asynchronously.

    But other than the `ProcessPoolExecutor
    <https://docs.python.org/dev/library/concurrent.futures.html#concurrent.futures.ProcessPoolExecutor>`_,
    the ProcessKillingExecutor forks a new Process for each function call that
    terminates after the function returns or if a timeout occurs.

    This means that contrary to the Executors and similar classes provided by
    the Python Standard Library, you can rely on the fact that a process will
    get killed if a timeout occurs and that absolutely no side can occur
    between function calls.

    Note that descendant processes of each process will not be terminated –
    they will simply become orphaned.
    """

    def __init__(self, max_workers: int=None):
        self.processes = max_workers or os.cpu_count()

    def map(self,
            func: Callable,
            iterable: Iterable,
            timeout: float=None,
            callback_timeout: Callable=None,
            daemon: bool = True
            ) -> Iterable:
        """
        :param func: the function to execute
        :param iterable: an iterable of function arguments
        :param timeout: after this time, the process executing the function
                will be killed if it did not finish
        :param callback_timeout: this function will be called, if the task
                times out. It gets the same arguments as the original function
        :param daemon: define the child process as daemon
        """
        executor = ThreadPoolExecutor(max_workers=self.processes)
        params = ({'func': func, 'fn_args': p_args, "p_kwargs": {},
                   'timeout': timeout, 'callback_timeout': callback_timeout,
                   'daemon': daemon} for p_args in iterable)
        return executor.map(self._submit_unpack_kwargs, params)

    def _submit_unpack_kwargs(self, params):
        """ unpack the kwargs and call submit """

        return self.submit(**params)

    def submit(self,
               func: Callable,
               fn_args: Any,
               p_kwargs: Dict,
               timeout: float,
               callback_timeout: Callable[[Any], Any],
               daemon: bool):
        """
        Submits a callable to be executed with the given arguments.
        Schedules the callable to be executed as func(*args, **kwargs) in a new
         process.
        :param func: the function to execute
        :param fn_args: the arguments to pass to the function. Can be one argument
                or a tuple of multiple args.
        :param p_kwargs: the kwargs to pass to the function
        :param timeout: after this time, the process executing the function
                will be killed if it did not finish
        :param callback_timeout: this function will be called with the same
                arguments, if the task times out.
        :param daemon: run the child process as daemon
        :return: the result of the function, or None if the process failed or
                timed out
        """
        p_args = fn_args if isinstance(fn_args, tuple) else (fn_args,)
        queue = Queue()
        p = Process(target=self._process_run,
                    args=(queue, func, fn_args,), kwargs=p_kwargs)

        if daemon:
            p.deamon = True

        p.start()
        p.join(timeout=timeout)
        if not queue.empty():
            return queue.get()
        if callback_timeout:
            callback_timeout(*p_args, **p_kwargs)
        if p.is_alive():
            p.terminate()
            p.join()

    @staticmethod
    def _process_run(queue: Queue, func: Callable[[Any], Any]=None,
                     *args, **kwargs):
        """
        Executes the specified function as func(*args, **kwargs).
        The result will be stored in the shared dictionary
        :param func: the function to execute
        :param queue: a Queue
        """
        queue.put(func(*args, **kwargs))


def some_task(n, *args, **kwargs):
    import time
    time.sleep(n/4)
    return n ** 2


if __name__ == "__main__":
    def fun_timeout(n):
        print('timeout:', n)
    executor = ProcessKillingExecutor(max_workers=2)
    generator = executor.map(some_task, [1, 1, 2, 2, 3, 3, 4, 4], timeout=2,
                             callback_timeout=fun_timeout)
    for elem in generator:
        print(elem)

---

What would be the coolest solution? It would be to have a class inherit from ThreadPoolExecutor and override the specific part of the class that executes the the threads inherent to ThreadPoolExecutor with what you want to do.