Unexpected timeout behaviour in Python processes depending on return value of called function

julian 0
Viewed 30

Given the following program:

import multiprocessing
import time
from multiprocessing import Process
import numpy as np


def run_process_timeout_wrapper(function, args, timeout):

    def aux(n, out_q):
        res = function(*n)
        out_q.put(res)  # to get result back from thread target

    result_q = multiprocessing.Queue()

    p = Process(target=aux, args=(args, result_q))
    p.start()
    x = result_q.get()
    p.join(timeout=timeout)
    p.terminate()

    if p.is_alive():
        p.terminate()
        raise multiprocessing.TimeoutError("Timed out after waiting for {}s".format(timeout))

    return x


def foo(x):
    return [(np.random.rand(50000), 0.993) for _ in range(10)]


def bar(x):
    return x

def foobar(x):
    res = [(np.random.rand(50000), 0.993) for _ in range(10000)]
    return res


if __name__ == '__main__':
    t1 = time.time()
    res = run_process_timeout_wrapper(foo, (110,), 40)
    print(time.time() - t1)

    t1 = time.time()
    res1 = run_process_timeout_wrapper(bar, (110,), 40)
    print(time.time() - t1)

    t1 = time.time()
    res2 = run_process_timeout_wrapper(foobar, (110,), 1)
    print(time.time() - t1)

The output of this program is 40seconds for the first process and about 0 for the second. I understand returning the array taking a bit more time than 0 but I am confused as to what is happening. As both foo and bar should probably take less than 40 seconds (the timeout) to compute, I would expect both to finish quickly and returning.

UPDATE:

Thank you for the answer regarding the reordering of x.get() with join. However, I found another unexpected behaviour by adding a task that should time out. Running the last part (foobar function) now seems to be non terminating despite the fact that I would expect into to terminate after 1 second and raise a TimeoutError

timeout multiprocessing python
1 Answers

From the doc:

Joining processes that use queues

Bear in mind that a process that has put items in a queue will wait before terminating until all the buffered items are fed by the “feeder” thread to the underlying pipe. (The child process can call the Queue.cancel_join_thread method of the queue to avoid this behaviour.)

This means that whenever you use a queue you need to make sure that all items which have been put on the queue will eventually be removed before the process is joined. Otherwise you cannot be sure that processes which have put items on the queue will terminate. Remember also that non-daemonic processes will be joined automatically.

So quick workaround will be put result_q.get() before p.join():

x = result_q.get()
p.join(timeout=timeout)
Andrej Kesely0
Related