Separate Envelope writing from marshalling #546

emcfarlane · 2023-07-10T14:23:36Z

A new API for working with envelope messages. Envelopes are built up within a single buffer. We can now use a single write call and pass a bytes.Buffer as the full payload.

With this change marshallers no longer write to the stream. Control is given to the protocols connection handler. This allows the protocols implementation to easily customize the behaviour. However, it does duplicates some of the flow. Overall I think it's simpler to duplicate control flow into the connection handler to allow for protocol specific features. Changes:

Connect unary now use marshal to gain marshalAppend codecs: https://github.com/bufbuild/connect-go/blob/6183cbf6ff174060dbdc494095ed7934dee4bc7d/protocol_connect.go#L907
No longer need to partially build envelopeWriters, create the envelope and write it instead: https://github.com/bufbuild/connect-go/blob/6183cbf6ff174060dbdc494095ed7934dee4bc7d/error_writer.go#L133-L138
errSpecialEnvelope is no longer needed, protocols check for end flags directly: https://github.com/bufbuild/connect-go/blob/6183cbf6ff174060dbdc494095ed7934dee4bc7d/envelope.go#L28
gRPC read trailer functions can be moved into the Receive() functions: https://github.com/bufbuild/connect-go/blob/6183cbf6ff174060dbdc494095ed7934dee4bc7d/protocol_grpc.go#L323
Easier to avoid leaking abstractions from one protocol to the other. Error messages are set in the handler: https://github.com/bufbuild/connect-go/blob/6183cbf6ff174060dbdc494095ed7934dee4bc7d/envelope.go#L190
Small reduction in allocs/op:

- BenchmarkConnect/unary-8                     583           2035062 ns/op         5448544 B/op        232 allocs/op
+ BenchmarkConnect/unary-8                     580           2054063 ns/op         5190420 B/op        221 allocs/op

Separates marshalling and unmarshalling from writing to the stream. Conn handlers are responsible for marshal -> compress -> write logic. Let's Conn handlers use custom logic for buffering on the wire.

emcfarlane · 2023-07-10T14:38:00Z

connect.go

@@ -364,7 +364,7 @@ func receiveUnaryResponse[T any](conn StreamingClientConn) (*Response[T], error)
 	// In a well-formed stream, the response message may be followed by a block
 	// of in-stream trailers or HTTP trailers. To ensure that we receive the
 	// trailers, try to read another message from the stream.
-	if err := conn.Receive(new(T)); err == nil {


I think we can remove this extra Receive call with this protocol drain fix: #536

Awesome. That'd also let us make the function non-generic, which would cut binary size a bit.

akshayjshah

This is ... a lot to get through in one sitting 😅 Most important questions:

Is the new envelope type the best we can do? (Asked inline on the type.)
How can we break this change up into more easily-reviewed chunks? As is, it's really difficult to substantively review.

buffer_pool.go

akshayjshah · 2023-07-11T20:49:57Z

envelope.go

-		return errorf(CodeInternal, "marshal message: %w", err)
+func marshal(dst *bytes.Buffer, message any, codec Codec) *Error {
+	if message == nil {
+		return nil


This isn't correct - nil handling is codec-specific. This works for binary protobuf but isn't correct for protojson (where we should write {}).

akshayjshah · 2023-07-11T20:52:08Z

envelope.go

-	defer w.bufferPool.Put(buffer)
-	envelope := &envelope{Data: buffer}
-	return w.Write(envelope)
+	dst.Write(raw)


We've allocated this []byte; we should try to put it back into the pool when we're done with it.

akshayjshah · 2023-07-11T20:55:10Z

envelope.go

+	}
+	if readMaxBytes > 0 && bytesRead > int64(readMaxBytes) {
+		// Attempt to read to end in order to allow connection re-use
+		discardedBytes, err := io.Copy(io.Discard, src)


Can we use the discard function here, so we're not copying an unbounded number of bytes from a badly-behaved client?

akshayjshah · 2023-07-11T23:39:09Z

envelope.go

+func makeEnvelope(buffer *bytes.Buffer) envelope {
+	var head [5]byte
+	buffer.Write(head[:])
+	return envelope{Buffer: buffer}


With these changes, is envelope doing enough work to keep around? It doesn't seem to provide much value as a wrapper on top of bytes.Buffer, and I don't love that the unwrap method isn't safe to call twice.

Would the rest of the repository be more clear if we instead had something like this (and managed the envelope and message buffer separately)?

type envelope [5]byte func (e *envelope) Flags() uint8 { ... } func (e *envelope) SetFlags(uint8) { ... } func (e *envelope) HasFlags(uint8) { ... } // check if flags are set func (e *envelope) Size() int { ... } // read size from last 4 bytes func (e *envelope) SetSize(int) { ... } // write size to last 4 bytes

Envelope is only needed to build a new enveloped message. On read we need to unwrap it to access the message size and then only need to return flags as the message size is encoded in the return data payload.

To make it easier to build new messages and allow downstream to be agnostic to whether the message is enveloped or not we can encode the envelope message in the bytes.Buffer directly.

buffer := &bytes.Buffer{} marshal(buffer, msg) if buffer.Len() > min { compressBuffer := &bytes.Buffer{} compress(compressBuffer, buffer) buffer = compressBuffer // swap } write(buffer)

Can be written in an envelope as:

envelope := makeEnvelope(&bytes.Buffer{}) marshal(envelope.Buffer, msg) var flags uint8 if envelope.size() > min { compressEnvelope := makeEnvelope(&bytes.Buffer{}) compress(compressEnvelope.Buffer, envelope.unwrap()) flags |= flagCompressed envelope = compressEnvelope // swap } envelope.encodeSizeAndFlags(flags) write(envelope.Buffer)

Defiantly something better could be done though. The unwrap() is used to escape the envelope and return the data for a reader (needed when we compress the payloads). Maybe:

type envelope struct { Buffer *bytes.Buffer } func (e envelope) Encode(flags uint8) func (e envelope) Size() int // size of data, buffer.Len() - 5 func (e envelope) Unwrap() *bytes.Buffer

I think for flags we could have a type, but easy to write the logic inplace too.

type flags uint8 func (x flags) Has(y flags) bool { return x&y > 0 } func (x flags) Set(y flags) flags { return x | y }

emcfarlane · 2023-07-10T20:37:59Z

envelope.go

-	defer w.bufferPool.Put(buffer)
-	envelope := &envelope{Data: buffer}
-	return w.Write(envelope)
+	dst.Write(raw)


Needs to append to the buffer rather than replace. (we could add the original logic, it's just more complicated as still need to append(dst.Bytes(), raw).

emcfarlane · 2023-07-12T13:19:40Z

envelope.go

+func makeEnvelope(buffer *bytes.Buffer) envelope {
+	var head [5]byte
+	buffer.Write(head[:])
+	return envelope{Buffer: buffer}


Envelope is only needed to build a new enveloped message. On read we need to unwrap it to access the message size and then only need to return flags as the message size is encoded in the return data payload.

To make it easier to build new messages and allow downstream to be agnostic to whether the message is enveloped or not we can encode the envelope message in the bytes.Buffer directly.

buffer := &bytes.Buffer{} marshal(buffer, msg) if buffer.Len() > min { compressBuffer := &bytes.Buffer{} compress(compressBuffer, buffer) buffer = compressBuffer // swap } write(buffer)

Can be written in an envelope as:

envelope := makeEnvelope(&bytes.Buffer{}) marshal(envelope.Buffer, msg) var flags uint8 if envelope.size() > min { compressEnvelope := makeEnvelope(&bytes.Buffer{}) compress(compressEnvelope.Buffer, envelope.unwrap()) flags |= flagCompressed envelope = compressEnvelope // swap } envelope.encodeSizeAndFlags(flags) write(envelope.Buffer)

Defiantly something better could be done though. The unwrap() is used to escape the envelope and return the data for a reader (needed when we compress the payloads). Maybe:

type envelope struct { Buffer *bytes.Buffer } func (e envelope) Encode(flags uint8) func (e envelope) Size() int // size of data, buffer.Len() - 5 func (e envelope) Unwrap() *bytes.Buffer

I think for flags we could have a type, but easy to write the logic inplace too.

type flags uint8 func (x flags) Has(y flags) bool { return x&y > 0 } func (x flags) Set(y flags) flags { return x | y }

emcfarlane · 2023-07-18T22:14:53Z

Closing, will look at breaking apart!

emcfarlane added 3 commits July 6, 2023 14:10

Flatten marshal and unmarshal

4d239b9

Separates marshalling and unmarshalling from writing to the stream. Conn handlers are responsible for marshal -> compress -> write logic. Let's Conn handlers use custom logic for buffering on the wire.

Encode envelope as a single buffer

ea38c47

Remove conn lived buffers

e5503a1

emcfarlane requested a review from akshayjshah July 10, 2023 14:23

emcfarlane self-assigned this Jul 10, 2023

emcfarlane marked this pull request as ready for review July 10, 2023 14:32

emcfarlane commented Jul 10, 2023

View reviewed changes

akshayjshah reviewed Jul 11, 2023

View reviewed changes

emcfarlane commented Jul 12, 2023

View reviewed changes

Merge branch 'main' into emcfarlane/marshallers

1035d78

emcfarlane closed this Jul 18, 2023

emcfarlane mentioned this pull request Aug 11, 2023

Initial stab at API for how users configure the middleware connectrpc/vanguard-go#10

Merged

emcfarlane mentioned this pull request Sep 12, 2023

Separate envelope writing from marshalling #586

Closed

akshayjshah deleted the emcfarlane/marshallers branch September 15, 2023 23:51

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Separate Envelope writing from marshalling #546

Separate Envelope writing from marshalling #546

emcfarlane commented Jul 10, 2023 •

edited

Loading

emcfarlane Jul 10, 2023 •

edited by jhump

Loading

akshayjshah Jul 11, 2023

akshayjshah left a comment

akshayjshah Jul 11, 2023

akshayjshah Jul 11, 2023

akshayjshah Jul 11, 2023

akshayjshah Jul 11, 2023

emcfarlane Jul 12, 2023 •

edited

Loading

emcfarlane Jul 10, 2023 •

edited

Loading

emcfarlane Jul 12, 2023 •

edited

Loading

emcfarlane commented Jul 18, 2023

Separate Envelope writing from marshalling #546

Separate Envelope writing from marshalling #546

Conversation

emcfarlane commented Jul 10, 2023 • edited Loading

emcfarlane Jul 10, 2023 • edited by jhump Loading

Choose a reason for hiding this comment

akshayjshah Jul 11, 2023

Choose a reason for hiding this comment

akshayjshah left a comment

Choose a reason for hiding this comment

akshayjshah Jul 11, 2023

Choose a reason for hiding this comment

akshayjshah Jul 11, 2023

Choose a reason for hiding this comment

akshayjshah Jul 11, 2023

Choose a reason for hiding this comment

akshayjshah Jul 11, 2023

Choose a reason for hiding this comment

emcfarlane Jul 12, 2023 • edited Loading

Choose a reason for hiding this comment

emcfarlane Jul 10, 2023 • edited Loading

Choose a reason for hiding this comment

emcfarlane Jul 12, 2023 • edited Loading

Choose a reason for hiding this comment

emcfarlane commented Jul 18, 2023

emcfarlane commented Jul 10, 2023 •

edited

Loading

emcfarlane Jul 10, 2023 •

edited by jhump

Loading

emcfarlane Jul 12, 2023 •

edited

Loading

emcfarlane Jul 10, 2023 •

edited

Loading

emcfarlane Jul 12, 2023 •

edited

Loading