Streaming thread safety #581
Closed
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Because it doesn't actually do anything now.
Using `myid()` with `workers()` meant that when the context was initialized with a single worker the processor list would be: `[OSProc(1), OSProc(1)]`. `procs()` will always include PID 1 and any other workers, which is what we want.
This shouldn't be necessary since we `wait()` for the given task in another `errormonitor()`'d task.
This was observed when running the `Single task running forever` test with multiple threads.
Necessary because `stream_pull_values!()` may throw different exceptions depending on whether the exception occurred locally or remotely.
Otherwise it's possible that the scheduler will close the DTask before all outputs have been sent, which would cause the downstream tasks to hang. This is how it could happen: 1. A streaming task starts. 2. The output handler task calls `take!(::ProcessRingBuffer)` on an output buffer, finds it empty, and `yield()`'s. 3. The task executes, pushes its output to the output buffers, reaches `max_evals` and finishes. 4. The scheduler finishes the corresponding DTask. 5. The `take!(::ProcessRingBuffer)` call resumes. The buffer isn't empty anymore but it calls `task_may_cancel(; must_force=true)` before continuing and throws an exception since the scheduler has finished the DTask. The result is that the last output is never sent, and the exeption is swallowed by the output handler started by `initialize_output_stream!()`. 6. Downstream tasks don't get that last result so they never reach `max_evals` and spin forever. Fixed by storing the handler tasks in the `StreamStore` and closing them in `close(::StreamStore)`. Also increased the timeout of the 'Single task running forever' task because it will sometimes timeout before the default 10s is up.
|
Blegh, evidently didn't get all of them: https://buildkite.com/julialang/dagger-dot-jl/builds/1394#01937fb0-b82c-45c0-b5b0-02471b806bab/436-1441 |
jpsamaroo
force-pushed
the
jps/stream2
branch
4 times, most recently
from
16827af to
a765cbe
Compare
|
Most of these changes were cherry-picked or manually ported into #463, which is generally passing CI, so I'm going to close this. Thanks for putting these fixes together! |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There's a few changes in here, I would recommend reviewing each commit individually. The most important one is 96a0c46.
Side note: I see that we have both
take!(::StreamStore, ::UInt)andtake!(::StreamingValue). The former is unused AFAICT, and it differs from the latter in that it will wait for the store to be notified when empty instead of spinning. Should we use it instead? Or delete it?