ref: Allow open messages to carry a payload so we don't have to
send two messages at the start of a flow.
This speeds up the single RPC latency between 1.5% and 9%.
MultiPart: 1/2
Change-Id: Ie4356a2c07b221cbec86c26f333f44f2ab5d4bbc
diff --git a/runtime/internal/flow/conn/flow.go b/runtime/internal/flow/conn/flow.go
index c2b8d6c..dd6ffc6 100644
--- a/runtime/internal/flow/conn/flow.go
+++ b/runtime/internal/flow/conn/flow.go
@@ -90,20 +90,6 @@
sent := 0
var left []byte
err := f.worker.Run(f.ctx, func(tokens int) (int, bool, error) {
- if !f.opened {
- // TODO(mattr): we should be able to send multiple messages
- // in a single writeMsg call.
- err := f.conn.mp.writeMsg(f.ctx, &message.OpenFlow{
- ID: f.id,
- InitialCounters: defaultBufferSize,
- BlessingsKey: f.bkey,
- DischargeKey: f.dkey,
- })
- if err != nil {
- return 0, false, err
- }
- f.opened = true
- }
size := 0
var bufs [][]byte
if len(left) > 0 {
@@ -136,7 +122,22 @@
d.Flags |= message.DisableEncryptionFlag
}
sent += size
- return size, done, f.conn.mp.writeMsg(f.ctx, d)
+
+ var err error
+ if f.opened {
+ err = f.conn.mp.writeMsg(f.ctx, d)
+ } else {
+ err = f.conn.mp.writeMsg(f.ctx, &message.OpenFlow{
+ ID: f.id,
+ InitialCounters: defaultBufferSize,
+ BlessingsKey: f.bkey,
+ DischargeKey: f.dkey,
+ Flags: d.Flags,
+ Payload: d.Payload,
+ })
+ f.opened = true
+ }
+ return size, done, err
})
if alsoClose || err != nil {
f.close(f.ctx, err)
