Add pipeline settlement signal - #108
Merged
Merged
Conversation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #108 +/- ##
==========================================
+ Coverage 99.76% 99.87% +0.11%
==========================================
Files 15 16 +1
Lines 837 788 -49
==========================================
- Hits 835 787 -48
+ Misses 2 1 -1 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
This file contains hidden or 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
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.
Blocking sinks in rill return as soon as their outcome is known. That's deliberate – the sooner we get
control back, the sooner we can cancel the source that might still be producing work, and whatever is
still in flight across the pipeline stages.
What that model doesn't give us is an "everything is done" event. Sometimes we need one: to free a
shared resource, or to safely look at side effects the callbacks produced. Basically
sync.WaitGroup.Wait(), but for a pipeline. That's what we're adding.One catch, and it's the whole reason
cancel()is in that snippet: settlement doesn't stop anything, itonly tells us when the pipeline ran out of work. If the source keeps producing, we keep waiting,
so stopping the source has to come first.
How it works
We say a pipeline stage is settled once it won't do any more work: it has fully consumed its input, and
every callback it started has returned.
Most stages already expose this, they just don't call it that.
Map,Filter,FromSliceand friendsclose their output channel only after they've settled, so a closed output is the settlement signal.
Sinks (e.g.
ForEach,ToSlice,Reduce) are the hole – no output channel, nothing to close, no signal. We give them an equivalent one.The useful part is that it composes backwards. A settled sink means the stage before it closed its
output, which means that stage settled, which means the stage before it closed its output, and so on up
to the source. One signal at the end of a linear pipeline covers the whole thing. And it's a real
happens-before edge, not a timing coincidence – the tests pin that.
Two things to know:
Tee) need one settlement per terminal sink. A settled sink saysnothing about callbacks still running in a sibling branch.
input is consumed and the work is done.
Compatibility
Existing call sites keep working, since the options are variadic. But every sink's signature changed, so
code that assigns a sink to an exact function type has to be updated. Technically, this is a breaking
API change.