First stab at adding dynamic requires()

[erikbern]

• Scheduler now handles a new state "pending"
• Some funky code in the worker to halt execution and resume later
• Might needs more testing
• Put together somewhere over Greenland :)

TODO: there are still some bugs when running with multiple workers, I'm investigating it

[erikbern]

oops meant the new state is "suspended"

[erikbern]

[erikbern]

Some notes:

• This probably won't be merged in a while and even so, it's an experimental feature.
• However at some point it has the potential to completely replace the requires method
• Tasks are always resumed by the same worker that started running the task. This is not strictly needed if tasks are idempotent – but I think it's a reasonable constraint
• We could get away without the new "suspended" state and just keep things in "running" although it might be confusing down the road
• There is a lot of overhead in the scheduler, adding even more on top of what was already there. Need to come up with better algorithms at some point
• I'm not sure if x.output() should be returned into the yield, maybe it should be the task x itself
• This doesn't work with multiple workers, because we fork after everything static is scheduled and then dynamically added tasks are only known to a single worker. Not sure how to get around this, maybe by switching to threads or using shared memory.

[erikbern]

comments...plz :(

[miku]

I meant to +1 on this since I saw this PR, sorry - I remember the ML discussion about this. The only thing I'd miss would be the separation of run and requires, which I find helpful conceptually and visually, when reading code. I'll test a couple of our real-world workflows with your model in the next days.

[erikbern]

Yeah it could potentially be a bit confusing with two different ways to specify the requirements... but in most cases it's probably nice to use requires() when you can, just for the reason you mentioned

[freider]

What's up with the CI error? Same "random" error as we've seen before?

[erikbern]

I think it was passing the tests earlier right?

Btw can anyone take a look at this PR? I don't want it to be sitting here forever

[freider]

The default value of new_deps is [], not None, is this intended?

[erikbern]

fixed

[freider]

This has great potential! I added some inline comments. There are a couple of gotchas I've been thinking about and I'm not sure if they are addressed:

• The worker will still quit when no tasks are available right? (unless you use the flag to continue running). In the case that you have failing or non-existing external dependencies as part of the dynamic requirements, that would leave suspended tasks in the task graph, without any worker actually working on them. A test for what happens in these cases would be nice.
• Do we need some specialized task pruning for suspended tasks? Haven't really thought through it, but they should be cleaned up whenever the worker responsible times out right? Should double check the pruning code to make sure they aren't left behind forever, blocking the task execution of future workers.
• Would be great if we had some unit/integration test checking what happens to the task state in the scheduler if you have multiple clients scheduling the same dynamic task in an "interleaved" way. I.e. can the task state somehow get reset from being suspended to something else and thereby mess things up (doesn't seem like it from the code, but a test would be nice to simplify future refactoring)

[erikbern]

I might try to refactor this to avoid the "suspended" state and just keep the "running" state instead. Then what would be nice is if we can do this across workers and not just do it on the same worker. I might add another mechanism to enforce "local" execution of dependencies later since we rely on it for a bunch of stuff (this would also replace the clunky delegates decorator)

[erikbern]

Old stuff, will abandon.

Definitely planning to resubmit this one though