Skip to content
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

Jump to bottom

[http_util/bufWriter] fast-fail on error returned from flushKeepBuffer() #7394

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
+50 −9
Open

[http_util/bufWriter] fast-fail on error returned from flushKeepBuffer() #7394

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
0edfecd
[http_util/bufWriter] fast-fail on error returned from flushKeepBuffer()
veshij Jul 5, 2024
ac51609
address comments
veshij Jul 9, 2024
f43d124
return nil if success
veshij Jul 9, 2024
17d2342
address PR comments
veshij Jul 10, 2024
446bee1
correctly return total bytes written from Write() call
veshij Jul 19, 2024
93993a0
remove blankline
veshij Jul 19, 2024
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
23 changes: 14 additions & 9 deletions internal/transport/http_util.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -329,20 +329,24 @@ func (w *bufWriter) Write(b []byte) (n int, err error) {
b := w.pool.Get().(*[]byte)
w.buf = *b
}
var bytesWritten int
for len(b) > 0 {
nn := copy(w.buf[w.offset:], b)
b = b[nn:]
w.offset += nn
n += nn
if w.offset >= w.batchSize {
err = w.flushKeepBuffer()
veshij marked this conversation as resolved.
Show resolved Hide resolved
bytesWritten, err = w.flushKeepBuffer()
n += bytesWritten
if err != nil {
return n, err
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The tests are failing because with these new changes, n is not updated when the condition if w.offset >= w.batchSize it not true.

I think we can rewrite with variable names that are more descriptive (and without named return parameters) as follows:

func (w *bufWriter) Write(b []byte) (int, error) {
	if w.err != nil {
		return 0, w.err
	}
	if w.batchSize == 0 { // Buffer has been disabled.
		n, err := w.conn.Write(b)
		return n, toIOError(err)
	}
	if w.buf == nil {
		b := w.pool.Get().(*[]byte)
		w.buf = *b
	}
	written := 0
	for len(b) > 0 {
		copied := copy(w.buf[w.offset:], b)
		b = b[copied:]
		w.offset += copied
		if w.offset < w.batchSize {
			written += copied
			continue
		}
		flushed, err := w.flushKeepBuffer()
		if err != nil {
			return written + flushed, err
		}
		written += flushed
	}
	return written, nil
}

}
}
return n, err
return n, nil
}

func (w *bufWriter) Flush() error {
err := w.flushKeepBuffer()
_, err := w.flushKeepBuffer()
// Only release the buffer if we are in a "shared" mode
if w.buf != nil && w.pool != nil {
b := w.buf
Expand All @@ -352,17 +356,18 @@ func (w *bufWriter) Flush() error {
return err
}

func (w *bufWriter) flushKeepBuffer() error {
func (w *bufWriter) flushKeepBuffer() (int, error) {
if w.err != nil {
return w.err
return 0, w.err
}
if w.offset == 0 {
return nil
return 0, nil
}
_, w.err = w.conn.Write(w.buf[:w.offset])
var n int
n, w.err = w.conn.Write(w.buf[:w.offset])
w.err = toIOError(w.err)
w.offset = 0
return w.err
return n, w.err
Comment on lines +366 to +370
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This block of code could be simplified as:

	n, err := w.conn.Write(w.buf[:w.offset])
	w.err = toIOError(err)
	w.offset = 0
	return n, w.err

}

type ioError struct {
Expand Down
36 changes: 36 additions & 0 deletions internal/transport/http_util_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,10 @@
package transport

import (
"errors"
"fmt"
"io"
"net"
"reflect"
"testing"
"time"
Expand Down Expand Up @@ -215,6 +218,39 @@ func (s) TestParseDialTarget(t *testing.T) {
}
}

type badNetworkConn struct {
net.Conn
}

func (c *badNetworkConn) Write([]byte) (int, error) {
return 0, io.EOF
}

// This test ensures Write() on a broken network connection does not lead to
// an infinite loop. See https://github.com/grpc/grpc-go/issues/7389 for more details.
func (s) TestWriteBadConnection(t *testing.T) {
data := []byte("test_data")
veshij marked this conversation as resolved.
Show resolved Hide resolved
// Configure the bufWriter with a batchsize that results in data being flushed
// to the underlying conn, midway through Write().
writeBufferSize := (len(data) - 1) / 2
writer := newBufWriter(&badNetworkConn{}, writeBufferSize, getWriteBufferPool(writeBufferSize))
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We probably can add a test table, where we create the bufWriter with and without a shared pool and ensure that the behavior works across both cases.


errCh := make(chan error, 1)
go func() {
_, err := writer.Write(data)
errCh <- err
}()

select {
case <-time.After(time.Second):
t.Fatalf("Write() did not return in time")
case err := <-errCh:
if !errors.Is(err, io.EOF) {
t.Fatalf("Write() = %v, want error presence = %v", err, io.EOF)
}
}
}

func BenchmarkDecodeGrpcMessage(b *testing.B) {
input := "Hello, %E4%B8%96%E7%95%8C"
want := "Hello, 世界"
Expand Down
Loading