GetBody on client Requests

[emcfarlane]

Enable client request retries by setting request.GetBody() allowing retryable errors to retry. We create a new temporary buffer for the request writer. On write the stream and buffer are written to. If the buffer becomes too large, or the response is returned, the buffer is discarded. Afterwards writes ignore the buffer but continue to pass through to the underlying stream.

Tested by sending two requests over the same connection and triggering GOAWAY by setting "Connection" to"close". The streams after the GOAWAY get retried.

Measure overhead with the connect benchmark. Need more testing (too small a sample, but pipe rewrite saved some allocations).

$ go test -v -run=^$ -bench=BenchmarkConnect -benchmem -benchtime=5s
- BenchmarkConnect/unary-8                     477           2174951 ns/op         5709248 B/op        233 allocs/op
+ BenchmarkConnect/unary-8                     597           1983015 ns/op         5610329 B/op        230 allocs/op

Fixes https://github.com/bufbuild/connect-go/issues/541

[jhump]

I think this approach has a couple of fundamental flaws:

1. We don't know message boundaries at the point where we are buffering. This matter because ideally we would always save at least the first message for any RPC. If we can save without making a copy, then we aren't actually using any more memory (we're just keeping the first message(s) pinned to the heap a little longer).
2. There is more copying and re-allocating buffers than is necessary. It looks like this is done (1) to continue using io.Pipe instead of an alternate mechanism to deliver data, and (2) to preserve the current pattern where the writer is what releases a buffer back to the pool instead of the reader. If we change those two assumptions, I think a much more efficient solution is possible.

[emcfarlane]

@jhump thanks for reviewing.

1. I was assuming the copies would be short lived and hopefully small. Like a copy of the buffer when compressing. It's only on the clients request. For any retry logic we still need to limit the memory, coping the buffer effectively halves the usable retry limit.
2. I agree this is a nicer solution.

[jhump]

the copies would be short lived and hopefully small

Unfortunately, Go's GC is not generational, so it being short-lived doesn't help much with GC pressure. As to whether they are small, that depends on the message sizes. This approach effectively double-allocates the bytes for the requests of every unary RPC, which feels kinda bad. While requests may usually be small, they could be quite large. So RPCs with large requests pay a proportionally bigger penalty with this approach.

[emcfarlane]

Closing in favour of the message queue.

[emcfarlane]

@jhump I think the initial memory analysis of this implementation was wrong. Currently for each send a message is written and returned to the pool for reuse. For the queue implementation each buffer not returned to the pool accumulates. Buffering will be equivalent to accumulating buffers minus the cost of the copy to the buffer, which is small. Having a single *bytes.Buffer for the buffer instead of []*bytes.Buffer for the message queue also reduces the number of alive buffers. I can use this implementation to benchmark the two. Adding the queue will have a very similar interface but Write([]byte) (int, error) will be replaced with WriteMessage(*bytes.Buffer) (int, error) and need the logic for conditionally freeing buffers.

For the GOAWAY condition I think we will only need a small buffer to manage the race between writing to a Conn and receiving a GOAWAY frame from an earlier stream on the same Conn. This needs testing. But either implementation has to have a limit based on bytes. I don't think a time limit makes sense as this would grow the heap excessively for a problem I think only requires a small amount of buffering.

Going to add the following improvements to this PR:

1. Add a method Rewind() bool to reset the reader and avoid the copy on error for GetBody
2. Avoid buffering on Read for unary client requests by blocking on establishing the connection.
3. Use a custom Pipe implementation to keep reads serialized with buffer writes using a single lock.

[jhump]

Having a single *bytes.Buffer for the buffer instead of []*bytes.Buffer for the message queue also reduces the number of alive buffers.

This isn't really true. The vast majority of RPCs have are unary, in which case we create a single buffer that goes into the queue. So with both approaches, we're keeping a single extra buffer, that can't be released until later.

minus the cost of the copy to the buffer, which is small

This is an assumption that may not always hold. For example, we have several RPCs in the buf CLI that can send many megabytes in a unary request. If this situation were to occur in a high-volume server, we'd suddenly be adding lots of wasted copying as well as extra memory pressure of having to duplicate the request bytes in memory for every operation.

I don't think a time limit makes sense as this would grow the heap excessively

I don't follow. The time limit wouldn't grow the heap excessively, because there's a size limit already. The time limit is strictly about decreasing the time for which a buffer is pinned to the heap: it allows us to reclaim the buffer sooner for cases where the server takes a very long time to reply with a status code and headers.

[mattrobenolt]

@emcfarlane jumping in here late and haven't fully read through the implementation, but would splitting the path for unary requests and bidirectional requests simplify a solution here?

The abstraction of a pipe and a writable buffer sorta aren't needed to the same extent for a unary request, and I think that abstraction is what's making it a bit more complicated to support the GetBody call.

I was sorta tinkering with a separate unary_http_call.go implementation rather than this duplex variant to simplify and speed up the unary path, which is also likely a more common path.

I suspect splitting the behavior on this boundary would help since we should easily have the entire request buffered already.

I also apologize since I haven't taken the time to fully read everything, just wanted to toss out this 2 cents for an alternative instead of trying to shove it into the current duplex behavior which has extra complexities to handle bidi streams.

[emcfarlane]

Hey @mattrobenolt, it would be great to see the tinkering on the unary client implementation! The unary solution would avoid the need for maintaining the buffers between send/recvs. I'm hoping an optimised solution for the streaming case won't add any additional overhead for unary calls. If that's the case we can avoid special pathing it, but that might not be realistic.

[jhump]

@mattrobenolt, by coincidence, @emcfarlane and I were discussing something similar: the approach we're looking at here would improve the robustness for all kinds of calls, but it is more intrusive and thus higher risk. So perhaps a better tradeoff for the short-term is to make a more surgical fix just for unary calls.

[mattrobenolt]

I'm hoping an optimised solution for the streaming case won't add any additional overhead for unary calls. If that's the case we can avoid special pathing it, but that might not be realistic.

That's kinda why I was going down this path. The overhead of the io.Pipe and using a goroutine for the request is kinda the biggest bottleneck for a unary request, and only needed to support bidi. If anything, I'd personally prefer swapping the idea to being slightly less optimal bidi, and being hyper focused on client unary.

Also arguably, bidi is by nature much generally long running and the allocations or setup required to setup the stream should be much less significant compared to clients doing thousands of unary RPC's per second. You probably aren't doing thousands or tens of thousands of bidi streams per second, that'd be weird.

[emcfarlane]

Putting this PR on hold. Next step is to move away from the Write interface to a WriteMessage one which would hand over ownership of the message buffer and allow us to avoid needing to copy buffers.