Skip to content

drpcstream: make invoke sequence atomic under write lock #52

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
shubhamdhama wants to merge 15 commits into
base: stream-multiplexing
Choose a base branch
from
Open

drpcstream: make invoke sequence atomic under write lock #52

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
0f1fbbe
*: extract PacketAssembler for frame-to-packet assembly
shubhamdhama Mar 26, 2026
162c5ab
fixup! *: extract PacketAssembler for frame-to-packet assembly
shubhamdhama Mar 28, 2026
d6577f9
fixup! *: extract PacketAssembler for frame-to-packet assembly
shubhamdhama Mar 30, 2026
1818a7d
fixup! *: extract PacketAssembler for frame-to-packet assembly
shubhamdhama Mar 30, 2026
de30052
*: move packet assembly tests from stream_test to packet_assembler_test
shubhamdhama Mar 30, 2026
6ea6efb
drpcwire: reuse PacketAssembler buffer across packets
shubhamdhama Mar 30, 2026
b7ec558
drpcmanager: replace manageStreams loop with per-stream goroutines
shubhamdhama Mar 29, 2026
dcda07e
*: use per-stream Finished signal instead of shared sfin channel
shubhamdhama Mar 29, 2026
6abe605
drpcmanager: use atomic counter for client stream ID generation
shubhamdhama Mar 29, 2026
0129d8f
drpcmanager: replace streamBuffer with streamRegistry
shubhamdhama Mar 30, 2026
e77dd9b
fixup! drpcmanager: replace streamBuffer with streamRegistry
shubhamdhama Apr 1, 2026
a4092e3
fixup! drpcmanager: replace streamBuffer with streamRegistry
shubhamdhama Apr 1, 2026
adb62df
drpcmanager: remove InactivityTimeout option
shubhamdhama Mar 31, 2026
553b574
drpc: enable stream multiplexing
shubhamdhama Apr 6, 2026
8ae283a
drpcstream: make invoke sequence atomic under write lock
shubhamdhama Apr 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion drpcclient/dialoptions.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,7 +129,6 @@ func DialContext(ctx context.Context, address string, opts ...DialOption) (conn
Stream: drpcstream.Options{
MaximumBufferSize: 0, // unlimited
},
SoftCancel: false,
},
}), nil
}
21 changes: 2 additions & 19 deletions drpcconn/conn.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,12 +135,7 @@ func (c *Conn) Invoke(ctx context.Context, rpc string, enc drpc.Encoding, in, ou
}

func (c *Conn) doInvoke(stream *drpcstream.Stream, enc drpc.Encoding, rpc string, data []byte, metadata []byte, out drpc.Message) (err error) {
if len(metadata) > 0 {
if err := stream.RawWrite(drpcwire.KindInvokeMetadata, metadata); err != nil {
return err
}
}
if err := stream.RawWrite(drpcwire.KindInvoke, []byte(rpc)); err != nil {
if err := stream.WriteInvoke(rpc, metadata); err != nil {
return err
}
if err := stream.RawWrite(drpcwire.KindMessage, data); err != nil {
Expand Down Expand Up @@ -171,25 +166,13 @@ func (c *Conn) NewStream(ctx context.Context, rpc string, enc drpc.Encoding) (_
return nil, err
}

if err := c.doNewStream(stream, rpc, metadata); err != nil {
if err := stream.WriteInvoke(rpc, metadata); err != nil {
return nil, errs.Combine(err, stream.Close())
}

return stream, nil
}

func (c *Conn) doNewStream(stream *drpcstream.Stream, rpc string, metadata []byte) error {
if len(metadata) > 0 {
if err := stream.RawWrite(drpcwire.KindInvokeMetadata, metadata); err != nil {
return err
}
}
if err := stream.RawWrite(drpcwire.KindInvoke, []byte(rpc)); err != nil {
return err
}
return nil
}

// encodeMetadata retrieves and encodes metadata from the provided
// (outgoing/client) context.
func (c *Conn) encodeMetadata(ctx context.Context) (metadata []byte, err error) {
Expand Down
Loading