Make Dagger.finish_stream() propagate downstream
#579
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
return_type() is kinda broken in v1.10, see: JuliaLang/julia#52385 In any case Base.promote_op() is the official public API for this operation so we should use it anyway.
This always us to handle all the other kinds of task specs.
This should get the docs building again.
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 is a bit nicer than commenting/uncommenting a line in the code.
Otherwise it may spin (see comments for details). Also refactored it into a while-loop instead of using a @goto.
This is useful for testing and benchmarking.
This is currently necessary for the streaming branch, we'll have to change this later but it's good to have CI working for now.
This works by converting the output buffers into a safely-serializeable container and sending that to the new node.
This makes them be displayed as if they were running under the original task.
This makes the tests a little easier to control.
Instead of taking/putting values sequentially (which may block), runs "pull" and "push" tasks for each input and output, respectively. Uses buffers to communicate values between pullers/pushers and the streaming task, instead of only using one buffer per task-to-task connection.
Switch from RemoteFetcher to RemoteChannelFetcher
Pass object rather than type to `stream_{push,pull}_values!`
ProcessRingBuffer: Don't exit on graceful interrupt when non-empty
- Added some whitespace. - Deleted the unused `rand_finite()` methods. - Allow passing the `timeout` to `test_finishes()` - Fix bug in one of the tests where we weren't waiting for all the tasks to finish, which would occasionally cause test failures because of the race condition.
JamesWrigley
force-pushed
the
finish-stream
branch
from
4ea4fbb
to
e1ccbfe
Compare
Previously a streaming task calling `Dagger.finish_stream()` would only stop the caller, but now it will also stop all downstream tasks. This is done by: - Getting the output handler tasks to close their `RemoteChannel` when exiting. - Making the input handler tasks close their buffers when the `RemoteChannel` is closed. - Exiting `stream!()` when an input buffer is closed.
`unwrap_nested_exception()` now supports `DTaskFailedException` so we can match against the real exceptions thrown.
JamesWrigley
force-pushed
the
finish-stream
branch
from
bf1a4ee
to
999bdd7
Compare
jpsamaroo
force-pushed
the
jps/stream2
branch
from
245d06a
to
e772af0
Compare
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.
Previously a streaming task calling
Dagger.finish_stream()would only stop the caller, but now it will also stop all downstream tasks. This is done by:RemoteChannelwhen exiting.RemoteChannelis closed.stream!()when an input buffer is closed.One question, what exactly are the
DropBuffertests for? Are they just for testingDropBuffer(its usefulness is not clear to me), or to test that streaming tasks should throw exceptions whenfetch()'d after cancelling? For now I've just disabled those tests because with the new behaviour a task stopping for any reason will also stop the downstream tasks. So currently in the tests thextask will reachmax_evalsand finish, which also causeAto finish, which means everything stops gracefully andAis never cancelled so it doesn't throw an exception.