feat: Implement linger for AppendRecordStream
#48
Conversation
|
Do we want to make |
Tonic has a hard dependency on tokio right? So it's ok... |
Yeah, that's what I thought! |
There was a problem hiding this comment.
Can you add a unit test exercising the combination of triggers - count, size, and linger. Using a paused tokio runtime makes it easy to advance time by doing a sleep (https://docs.rs/tokio/latest/tokio/time/fn.pause.html#auto-advance).
Also, can we try to use a Sleep directly instead of Interval? I think it can be easier to reason about.
Resolves: #37 Signed-off-by: Vaibhav Rabber <[email protected]>
Signed-off-by: Vaibhav Rabber <[email protected]>
Signed-off-by: Vaibhav Rabber <[email protected]>
Signed-off-by: Vaibhav Rabber <[email protected]>
vrongmeal
force-pushed
the
vrongmeal/linger-time
branch
from
5dc2a29 to
b12765a
Compare
Signed-off-by: Vaibhav Rabber <[email protected]>
vrongmeal
changed the title
linger_time for AppendRecordStreamlinger for AppendRecordStream
| @@ -30,6 +34,7 @@ impl Default for AppendRecordStreamOpts { | |||
| max_batch_size: ByteSize::mib(1), | |||
| match_seq_num: None, | |||
| fencing_token: None, | |||
| linger: None, | |||
There was a problem hiding this comment.
Let's default to 5ms (batching is generally helpful), and let it be disabled by making with_linger take an Option<Duration>.
|
Re: BTW same goes for |
| types, | ||
| }; | ||
|
|
||
| #[rstest] |
There was a problem hiding this comment.
nice! can you run these in the CI?
| // Reset the linger sleep future since the old one is polled ready. | ||
| self.linger_sleep = self.opts.linger_sleep_fut(); |
There was a problem hiding this comment.
| if self | ||
| .linger_sleep | ||
| .as_mut() | ||
| .is_some_and(|fut| fut.poll_unpin(cx).is_pending()) | ||
| { | ||
| return Poll::Pending; | ||
| } |
There was a problem hiding this comment.
The idea is that we want to trigger an append if any of these conditions can be satisfied:
max_batch_recordsaccumulated from underlying streammax_batch_sizeaccumulated from underlying streamlingersleep expired - and there is at least 1 accumulated record
So waiting upfront for linger time to expire when the underlying record stream could potentially satisfy the first 2 conditions, does not seem right to me...
BTW, I have a feeling this adaptor from AppendRecord to batched AppendInput given this combination of triggers may be clearest to implement using async_stream::stream!, and let that generate the state machine. Hand-written futures and streams tend to be hard to write correctly (and review)...
There was a problem hiding this comment.
This bugged me at first thinking linger is used to rate limit in many cases but makes sense if it matches the other conditions.
Will use async_stream::stream in that case, would make it much clearer.
|
Also, the comments regarding |
|
Going to close this PR in favour of #55 soon. |
Resolves: #37