diff --git a/caddytest/integration/caddyfile_adapt/reverse_proxy_stream_buffer_size.caddyfiletest b/caddytest/integration/caddyfile_adapt/reverse_proxy_stream_buffer_size.caddyfiletest new file mode 100644 index 000000000..5320b0529 --- /dev/null +++ b/caddytest/integration/caddyfile_adapt/reverse_proxy_stream_buffer_size.caddyfiletest @@ -0,0 +1,56 @@ +https://example.com { + reverse_proxy https://localhost:54321 { + stream_buffer_size 8KB + } +} + +---------- +{ + "apps": { + "http": { + "servers": { + "srv0": { + "listen": [ + ":443" + ], + "routes": [ + { + "match": [ + { + "host": [ + "example.com" + ] + } + ], + "handle": [ + { + "handler": "subroute", + "routes": [ + { + "handle": [ + { + "handler": "reverse_proxy", + "stream_buffer_size": 8000, + "transport": { + "protocol": "http", + "tls": {} + }, + "upstreams": [ + { + "dial": "localhost:54321" + } + ] + } + ] + } + ] + } + ], + "terminal": true + } + ] + } + } + } + } +} diff --git a/modules/caddyhttp/reverseproxy/caddyfile.go b/modules/caddyhttp/reverseproxy/caddyfile.go index 777bc06ac..a370a2873 100644 --- a/modules/caddyhttp/reverseproxy/caddyfile.go +++ b/modules/caddyhttp/reverseproxy/caddyfile.go @@ -96,6 +96,7 @@ func parseCaddyfile(h httpcaddyfile.Helper) (caddyhttp.MiddlewareHandler, error) // flush_interval // request_buffers // response_buffers +// stream_buffer_size // stream_timeout // stream_close_delay // verbose_logs @@ -646,7 +647,7 @@ func (h *Handler) UnmarshalCaddyfile(d *caddyfile.Dispenser) error { h.FlushInterval = caddy.Duration(dur) } - case "request_buffers", "response_buffers": + case "request_buffers", "response_buffers", "stream_buffer_size": subdir := d.Val() if !d.NextArg() { return d.ArgErr() @@ -670,6 +671,8 @@ func (h *Handler) UnmarshalCaddyfile(d *caddyfile.Dispenser) error { h.RequestBuffers = size case "response_buffers": h.ResponseBuffers = size + case "stream_buffer_size": + h.StreamBufferSize = int(size) } case "stream_timeout": diff --git a/modules/caddyhttp/reverseproxy/reverseproxy.go b/modules/caddyhttp/reverseproxy/reverseproxy.go index 2169d1717..3b9b56a05 100644 --- a/modules/caddyhttp/reverseproxy/reverseproxy.go +++ b/modules/caddyhttp/reverseproxy/reverseproxy.go @@ -171,6 +171,12 @@ type Handler struct { // forcibly closed at the end of the timeout. Default: no timeout. StreamTimeout caddy.Duration `json:"stream_timeout,omitempty"` + // The size of the buffer used for each direction of streaming + // requests such as WebSockets. If zero, the default size is 32 KiB. + // This only affects upgraded bidirectional streams, not normal + // request or response buffering. + StreamBufferSize int `json:"stream_buffer_size,omitempty"` + // If nonzero, streaming requests such as WebSockets will not be // closed when the proxy config is unloaded, and instead the stream // will remain open until the delay is complete. In other words, diff --git a/modules/caddyhttp/reverseproxy/streaming.go b/modules/caddyhttp/reverseproxy/streaming.go index 64b6d39d1..e454ee655 100644 --- a/modules/caddyhttp/reverseproxy/streaming.go +++ b/modules/caddyhttp/reverseproxy/streaming.go @@ -204,7 +204,12 @@ func (h *Handler) handleUpgradeResponse(logger *zap.Logger, wg *sync.WaitGroup, defer deleteFrontConn() defer deleteBackConn() - spc := switchProtocolCopier{user: conn, backend: backConn, wg: wg} + spc := switchProtocolCopier{ + user: conn, + backend: backConn, + wg: wg, + bufferSize: h.StreamBufferSize, + } // setup the timeout if requested var timeoutc <-chan time.Time @@ -636,20 +641,29 @@ func (m *maxLatencyWriter) stop() { type switchProtocolCopier struct { user, backend io.ReadWriteCloser wg *sync.WaitGroup + bufferSize int } func (c switchProtocolCopier) copyFromBackend(errc chan<- error) { - _, err := io.Copy(c.user, c.backend) + _, err := io.CopyBuffer(c.user, c.backend, c.buffer()) errc <- err c.wg.Done() } func (c switchProtocolCopier) copyToBackend(errc chan<- error) { - _, err := io.Copy(c.backend, c.user) + _, err := io.CopyBuffer(c.backend, c.user, c.buffer()) errc <- err c.wg.Done() } +func (c switchProtocolCopier) buffer() []byte { + size := c.bufferSize + if size <= 0 { + size = defaultBufferSize + } + return make([]byte, size) +} + var streamingBufPool = sync.Pool{ New: func() any { // The Pool's New function should generally only return pointer diff --git a/modules/caddyhttp/reverseproxy/streaming_test.go b/modules/caddyhttp/reverseproxy/streaming_test.go index 3f6da2ffa..ce0db65a0 100644 --- a/modules/caddyhttp/reverseproxy/streaming_test.go +++ b/modules/caddyhttp/reverseproxy/streaming_test.go @@ -2,8 +2,10 @@ package reverseproxy import ( "bytes" + "io" "net/http/httptest" "strings" + "sync" "testing" "github.com/caddyserver/caddy/v2" @@ -34,3 +36,47 @@ func TestHandlerCopyResponse(t *testing.T) { } } } + +func TestSwitchProtocolCopierBufferSize(t *testing.T) { + var wg sync.WaitGroup + var errc = make(chan error, 1) + var dst bytes.Buffer + + copier := switchProtocolCopier{ + user: nopReadWriteCloser{Reader: strings.NewReader("hello")}, + backend: nopReadWriteCloser{Writer: &dst}, + wg: &wg, + bufferSize: 7, + } + + buf := copier.buffer() + if got := len(buf); got != 7 { + t.Fatalf("buffer len = %d, want 7", got) + } + + wg.Add(1) + go copier.copyToBackend(errc) + wg.Wait() + + if err := <-errc; err != nil { + t.Fatalf("copyToBackend() error = %v", err) + } + if got := dst.String(); got != "hello" { + t.Fatalf("copied data = %q, want %q", got, "hello") + } +} + +func TestSwitchProtocolCopierDefaultBufferSize(t *testing.T) { + copier := switchProtocolCopier{} + buf := copier.buffer() + if got := len(buf); got != defaultBufferSize { + t.Fatalf("buffer len = %d, want %d", got, defaultBufferSize) + } +} + +type nopReadWriteCloser struct { + io.Reader + io.Writer +} + +func (nopReadWriteCloser) Close() error { return nil }