drpcmanager: replace streamBuffer with streamRegistry#47
Open
shubhamdhama wants to merge 3 commits intoshubham/use-atomic-counterfrom
Open
drpcmanager: replace streamBuffer with streamRegistry#47shubhamdhama wants to merge 3 commits intoshubham/use-atomic-counterfrom
shubhamdhama wants to merge 3 commits intoshubham/use-atomic-counterfrom
Conversation
shubhamdhama
force-pushed
the
shubham/use-atomic-counter
branch
from
f5869a5 to
f204e04
Compare
shubhamdhama
force-pushed
the
shubham/add-stream-registry
branch
from
86bb186 to
e16d021
Compare
Replace the single-stream streamBuffer with a stream registry that maps stream IDs to stream objects. The registry currently holds at most one active stream (two briefly during handoff), but provides the foundation for stream multiplexing where callers will look up streams by ID directly.
shubhamdhama
force-pushed
the
shubham/use-atomic-counter
branch
from
f204e04 to
6abe605
Compare
shubhamdhama
force-pushed
the
shubham/add-stream-registry
branch
from
e16d021 to
0129d8f
Compare
| r.mu.Lock() | ||
| defer r.mu.Unlock() | ||
|
|
||
| if stream == nil { |
There was a problem hiding this comment.
This is input validation and should be done before taking the lock.
| } | ||
|
|
||
| if r.closed { | ||
| return managerClosed.New("register") |
| r.mu.RLock() | ||
| defer r.mu.RUnlock() | ||
|
|
||
| s, ok := r.streams[id] |
There was a problem hiding this comment.
Don't you have to check if registry is closed?
| func (r *streamRegistry) Unregister(id uint64) { | ||
| r.mu.Lock() | ||
| defer r.mu.Unlock() | ||
|
|
There was a problem hiding this comment.
Don't you have to check if registry is closed?
|
|
||
| // ForEach calls fn for each registered stream. The registry is read-locked | ||
| // during iteration. | ||
| func (r *streamRegistry) ForEach(fn func(*drpcstream.Stream)) { |
| // reg tracks active streams. Currently holds at most one active stream; | ||
| // a second may briefly coexist during stream handoff (old stream's | ||
| // Unregister races with new stream's Register). | ||
| reg *streamRegistry |
There was a problem hiding this comment.
nit: why not just call it activeStreams?
| // empty. It iterates the map because the registry may briefly hold two streams | ||
| // during stream handoff. This method should be removed once multiplexing is | ||
| // supported and callers look up streams by ID directly. | ||
| func (r *streamRegistry) GetLatest() *drpcstream.Stream { |
There was a problem hiding this comment.
Keep track of the latest streamID and avoid iterating over the map holding the lock.
drpcmanager/manager.go
Outdated
| m.log("TERM", func() string { return fmt.Sprint(err) }) | ||
| m.sigs.tport.Set(m.tr.Close()) | ||
| m.sbuf.Close() | ||
| m.reg.ForEach(func(s *drpcstream.Stream) { s.Close() }) |
There was a problem hiding this comment.
Should this be s.Cancel() instead of s.Close()? s.Close() writes to the transport, and the current code already holds the lock and writes to the transport for every stream. I'm assuming this path isn't intended for mux streams?
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
2 participants
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.
Replace the single-stream streamBuffer with a stream registry that maps
stream IDs to stream objects. The registry currently holds at most one
active stream (two briefly during handoff), but provides the foundation
for stream multiplexing where callers will look up streams by ID directly.