Task timeout

[muchas]

Hi there,

Great project, thanks for your work!

Do you have any way to force the timeout on long running tasks?

pr.map(fn, stage, timeout=3)  # fn would time out after 3 seconds and skip the computation

[cgarciae]

@muchas I've been thinking of implementing something like this for the pr module because for some reason sometimes a process randomly hangs. What is your use case?

[muchas]

@cgarciae My pipeline contains nondeterministic, computationally expensive function which for most elements runs fast enough, but for some of them it takes too much time, blocking available processes. I'm fine with filtering those elements out.

Till now I was using pebble decorator:

@concurrency.process(timeout=5)
def possibly_long_computation():
    ...

However the above solution does not combine well with pypeln as I'm getting the following error:

AssertionError: daemonic processes are not allowed to have children

Thank you for looking into it!

EDIT:
pebble solution seems to work with threads

[cgarciae]

I'll try to implement this soon. I'll been doing a refactor + documentation improvement so I didn't want to add new stuff, but this feature seems very useful.

[cgarciae]

Checkout the new timeout parameter for most functions in the pypeln API, e.g pl.process.map