drpcstream: gate data-frame sends on the per-stream send window#71
drpcstream: gate data-frame sends on the per-stream send window#71suj-krishnan wants to merge 1 commit into
Conversation
50c697f to
a4ad9e8
Compare
58a5fed to
30ffb52
Compare
a4ad9e8 to
6ff7baf
Compare
30ffb52 to
9b6d87b
Compare
6ff7baf to
7fb0331
Compare
9b6d87b to
2512021
Compare
7fb0331 to
9c3bc23
Compare
2512021 to
c086e23
Compare
9c3bc23 to
efc77b4
Compare
c086e23 to
b14cb1f
Compare
efc77b4 to
d88576b
Compare
b14cb1f to
6594c14
Compare
b1f5883 to
4d08d0b
Compare
fe9d197 to
228866a
Compare
228866a to
de7b9dc
Compare
fcfa2e4 to
10b863f
Compare
4d08d0b to
deefd7b
Compare
7d03098 to
336c61c
Compare
deefd7b to
af4331b
Compare
ef342ce to
0d5cf31
Compare
shubhamdhama
left a comment
There was a problem hiding this comment.
These are the comments for the first commit.
| defer s.checkFinished() | ||
|
|
||
| // Wait for the write lock without holding s.mu (see Close). | ||
| // Wait for the write lock without holding s.mu, matching CloseSend's |
There was a problem hiding this comment.
This comment is related to the other PR.
| if s.sendw != nil { | ||
| s.sendw.close(io.EOF) | ||
| } | ||
| s.mu.Unlock() |
There was a problem hiding this comment.
This doesn't need to be wrapped in s.mu locks. Both are concurrency safe.
| @@ -588,12 +624,29 @@ | |||
| if s.sigs.term.IsSet() { | |||
| s.mu.Unlock() | |||
| return nil | |||
| } | |||
|
|
|||
| // Close the send window before taking the write lock so a send parked on | |||
| // credit wakes and releases the lock, instead of stalling the close on a | |||
| // grant that may never come (mirrors SendCancel's signal-first ordering). | |||
| if s.sendw != nil { | |||
| s.sendw.close(termClosed) | |||
| } | |||
| s.mu.Unlock() | |||
There was a problem hiding this comment.
Same here: This doesn't need to be wrapped in s.mu locks. Both are concurrency safe.
| s.mu.Unlock() | ||
| return nil | ||
| } | ||
| s.mu.Unlock() |
There was a problem hiding this comment.
This one too: This doesn't need to be wrapped in s.mu locks. This is concurrency safe.
| if s.sigs.term.IsSet() { | ||
| s.mu.Unlock() | ||
| return nil | ||
| } |
There was a problem hiding this comment.
Also from a correctness point of view this isn't required, but we can keep it as a "fast-path". So not asking for any change here.
| // disabled) leaves sends ungated. acquire parks until credit arrives, | ||
| // the ctx is canceled, or the window closes (stream termination). | ||
| if kind == drpcwire.KindMessage && s.sendw != nil { | ||
| if err := s.sendw.acquire(s.Context(), int64(len(fr.Data))); err != nil { |
There was a problem hiding this comment.
s.Context() is useless here. close does the trick of aborting. I would recommend just keep the n and remove the context.
There was a problem hiding this comment.
Good catch - the PR is updated.
| if s.sigs.send.IsSet() || s.sigs.term.IsSet() { | ||
| s.mu.Unlock() | ||
| return nil | ||
| } |
There was a problem hiding this comment.
Close, SendError, and CloseSend now all have nearly identical lock unlocks, maybe we should add in a small helper that could reduce the duplication?
There was a problem hiding this comment.
Have removed the changes from this PR for now as they are slightly adjacent to flow control - we'll tackle it separately
1520087 to
6649f34
Compare
Wire the sendWindow credit gate into rawWriteLocked: a KindMessage frame acquires len(frame) bytes of per-stream send credit before it is handed to the writer. Control frames (invoke/metadata) bypass the gate. The window is opt-in: a stream has no send window by default, so data writes stay ungated (unlimited) and behavior is unchanged until one is installed. terminate closes the window with the send-side error (sigs.send is first-wins, holding io.EOF when a cancel/error path pre-set it), so a send parked on credit returns the same error as one parked in WriteFrame or a later send. SendCancel and Cancel already terminate before taking the write lock, so they wake a credit-parked writer. acquire no longer takes a context: the stream's context Done channel is only closed once the stream is finished (after all ops complete), which cannot happen while an acquire is parked (the parked write is itself an in-flight op), so it never fired there. close is the sole abort path, and every termination route reaches it. Per-stream only. NOTE: Close/SendError/CloseSend still take the write lock while holding s.mu, so a send parked on credit can deadlock them; the lock-ordering rework that makes those paths preempt a parked writer is handled separately. Co-Authored-By: roachdev-claude <roachdev-claude-bot@cockroachlabs.com>
6649f34 to
48cda98
Compare
Wires the
sendWindowcredit gate into the stream write path:rawWriteLocked, aKindMessageframe acquireslen(frame)bytes of per-stream send credit before it is handed to the writer.terminatecloses the window with the send-side error (sigs.send.Err()— first-wins, soio.EOFwhen a cancel/error path pre-set it): a send parked on credit returns the same error as one parked inMuxWriter.WriteFrameor a later send.SendError's pre-close usesio.EOFandClose's usestermClosedfor the same parked/unparked agreement.