diff --git a/examples/http-linux/main-httplinux.go b/examples/http-linux/main-httplinux.go index 830b292..481d0f0 100644 --- a/examples/http-linux/main-httplinux.go +++ b/examples/http-linux/main-httplinux.go @@ -100,5 +100,5 @@ func homepage(exch *httphi.Exchange) { exch.StageHeader("Content-Type", "text/html") exch.StageHeaderInt("Content-Length", int64(n), 10) exch.WriteHeader(int(httphi.StatusOK)) - exch.Write(page[:n]) + exch.WriteBody(page[:n]) } diff --git a/http/httphi/bench_test.go b/http/httphi/bench_test.go index 256bf83..0d9996e 100644 --- a/http/httphi/bench_test.go +++ b/http/httphi/bench_test.go @@ -67,7 +67,7 @@ func BenchmarkHandle(b *testing.B) { if !present || !internal.BytesEqual(data, expect) { panic("invalid result") } - ex.Write(benchBody) + ex.WriteBody(benchBody) }, }, { diff --git a/http/httphi/exchange.go b/http/httphi/exchange.go index 35bf8ee..99c9561 100644 --- a/http/httphi/exchange.go +++ b/http/httphi/exchange.go @@ -1,6 +1,7 @@ package httphi import ( + "net" "slices" "strconv" "sync/atomic" @@ -16,7 +17,7 @@ const maxStatusLine = len("HTTP/1.1 ") + 3 + 1 + len("Network Authentication Req // Exchange is a single request-response cycle over a connection, playing the // part of both http.Request and http.ResponseWriter: Request* methods read the -// request, [Exchange.StageHeader] and [Exchange.Write] produce the response. +// request, [Exchange.StageHeader] and [Exchange.WriteBody] produce the response. // A [Router] owns a fixed pool of them, which is what bounds its memory. // // Request and response share one buffer, the response header being written over @@ -24,6 +25,7 @@ const maxStatusLine = len("HTTP/1.1 ") + 3 + 1 + len("Network Authentication Req // [Exchange.ReadBody] before setting response headers. type Exchange struct { used atomic.Bool + gen atomic.Uint32 respTopBuf [maxStatusLine]byte respTopWritten uint8 @@ -94,6 +96,7 @@ func (exch *Exchange) Acquire(conn conn) bool { if !exch.used.CompareAndSwap(false, true) { return false } + exch.gen.Add(1) exch.readErr = nil exch.respErr = nil exch.hijacked = false @@ -116,10 +119,12 @@ func (exch *Exchange) Release() { exch.rw.Close() } exch.rw = nil + exch.gen.Add(1) exch.used.Store(false) } -// UnsafeRawBuffer returns the contiguous buffer being used for the request and response. +// UnsafeRawBuffer returns the contiguous buffer owned by [Exchange] being used for the request and response. +// // Writing to it will mangle the entire request header+body and/or any staged response headers. // Does not return the buffer used for the response first line so can be safely // written to and used without modifying the staged response first line. @@ -127,12 +132,13 @@ func (exch *Exchange) Release() { // Staging headers will write to this buffer so use mindfully. // To access only the request header buffer portion use [httpraw.Header.BufferRaw] limited // to [httpraw.Header.BufferParsed] as returned by [Exchange.RequestHeaderRaw]. +// Writing to this section will not change the contents read by [Exchange.ReadBody]. // // In [Router] context, the size of this buffer is influenced directly by [RouterConfig] HeaderBufferSize fields. func (exch *Exchange) UnsafeRawBuffer() []byte { return exch.rawbuf } // StageHeader stages a response header field, written on the first -// [Exchange.FlushHeader], [Exchange.WriteHeader] or [Exchange.Write]. +// [Exchange.FlushHeader], [Exchange.WriteHeader] or [Exchange.WriteBody]. // Returns false and drops the field if the response buffer cannot fit it. // Has no effect once the header has been written. func (exch *Exchange) StageHeader(key, value string) (enoughMemory bool) { @@ -247,11 +253,71 @@ func (exch *Exchange) FlushHeader() (int, error) { return ng + ng2, err } +// ExchangeRW is an [io.ReadWriteCloser] view of an [Exchange] wrapping +// [Exchange.ReadBody] and [Exchange.WriteBody] methods. +// +// Exchanges are pooled and reused, so a handle records the exchange generation +// it was taken at and refuses to touch the connection once that exchange moves +// on to another request. Obtain one with [Exchange.ReadWriter]. +type ExchangeRW struct { + gen uint32 + exch *Exchange +} + +// IsValid returns true while the handle still refers to the request it was +// taken from, i.e: false once the exchange was released or hijacked away. +func (rw *ExchangeRW) IsValid() bool { + return rw.gen == rw.exch.gen.Load() && rw.exch.used.Load() +} + +func (rw *ExchangeRW) validate() error { + if !rw.IsValid() { + return net.ErrClosed + } + return nil +} + +// Write writes response body bytes. See [Exchange.WriteBody]. +// Fails with [net.ErrClosed] once the handle is no longer valid. +func (rw *ExchangeRW) Write(buf []byte) (int, error) { + if err := rw.validate(); err != nil { + return 0, err + } + return rw.exch.WriteBody(buf) +} + +// Read reads request body bytes. See [Exchange.ReadBody]. +// Fails with [net.ErrClosed] once the handle is no longer valid. +func (rw *ExchangeRW) Read(buf []byte) (int, error) { + if err := rw.validate(); err != nil { + return 0, err + } + return rw.exch.ReadBody(buf) +} + +// Close invalidates this handle so later reads and writes fail. It does not +// close the connection nor end the exchange, both of which the [Router] owns. +func (rw *ExchangeRW) Close() error { + if err := rw.validate(); err != nil { + return err + } + rw.gen-- + return nil +} + +// ReadWriter fills dst with a stream view of the exchange, valid until the +// exchange is released. The caller owns dst, so a handler may keep one and +// refill it every request without allocating. +func (exch *Exchange) ReadWriter(dst *ExchangeRW) { + dst.gen = exch.gen.Load() + dst.exch = exch +} + // Write writes response body bytes, flushing the header first if the handler // has not written it yet. Once a write to the connection fails the response is // unrecoverable and every later write returns that same error, so a body never // reaches the wire without its header. -func (exch *Exchange) Write(buf []byte) (int, error) { +func (exch *Exchange) WriteBody(buf []byte) (int, error) { if exch.respErr != nil { return 0, exch.respErr } else if !exch.headerWritten { diff --git a/http/httphi/exchange_test.go b/http/httphi/exchange_test.go index 9cd0225..12267a4 100644 --- a/http/httphi/exchange_test.go +++ b/http/httphi/exchange_test.go @@ -3,6 +3,7 @@ package httphi import ( "context" "errors" + "io" "strings" "testing" @@ -79,7 +80,7 @@ func TestExchangeWriteFlushesHeader(t *testing.T) { const body = "hello" conn := newConn("") exch := newExchange(t, conn, 128, false) - n, err := exch.Write([]byte(body)) + n, err := exch.WriteBody([]byte(body)) if err != nil { t.Fatal(err) } @@ -266,6 +267,56 @@ func TestHandleSilentHandler(t *testing.T) { } // Body bytes arriving in the same segment as the header must be readable. +var _ io.ReadWriteCloser = (*ExchangeRW)(nil) + +// ExchangeRW writes the response body and reads the request body, so it may be +// handed to code that wants an io.ReadWriter. +func TestExchangeRW(t *testing.T) { + const body = "hello" + conn := newConn("") + exch := newExchange(t, conn, 128, false) + var rw ExchangeRW + exch.ReadWriter(&rw) + + n, err := io.WriteString(&rw, body) + if err != nil { + t.Fatal(err) + } + if n != len(body) { + t.Errorf("want %d bytes written, got %d", len(body), n) + } + const want = "HTTP/1.1 200 OK\r\n\r\n" + body + if got := conn.ViewWritten(); got != want { + t.Errorf("want %q, got %q", want, got) + } +} + +// The exchange is pooled and reused: a handle kept past the request it was +// taken from must fail instead of reaching the next request's connection. +func TestExchangeRWOutlivesExchange(t *testing.T) { + conn := newConn("") + exch := newExchange(t, conn, 128, false) + var rw ExchangeRW + exch.ReadWriter(&rw) + if !rw.IsValid() { + t.Fatal("want a fresh handle to be valid") + } + exch.Release() + + if rw.IsValid() { + t.Error("want handle invalidated by release") + } + if _, err := rw.Write([]byte("late")); err == nil { + t.Error("want error writing through a released exchange, got nil") + } + if _, err := rw.Read(make([]byte, 4)); err == nil { + t.Error("want error reading through a released exchange, got nil") + } + if got := conn.ViewWritten(); strings.Contains(got, "late") { + t.Errorf("late write reached the connection: %q", got) + } +} + func TestExchangeReadBody(t *testing.T) { const body = "message body" var got string @@ -406,7 +457,7 @@ func TestExchangeWriteHeaderFlushFails(t *testing.T) { exch := newExchange(t, conn, 128, false) conn.FailWrites(1) // Status line write fails, body write would succeed. - n, err := exch.Write([]byte(body)) + n, err := exch.WriteBody([]byte(body)) if err == nil { t.Error("want error when header flush fails, got nil") } @@ -418,7 +469,7 @@ func TestExchangeWriteHeaderFlushFails(t *testing.T) { } // Writes after a failed header stay failed: the response is unrecoverable, // a body without its header would corrupt the stream. - if _, err = exch.Write([]byte(body)); err == nil { + if _, err = exch.WriteBody([]byte(body)); err == nil { t.Error("want error on write after failed header flush, got nil") } if got := conn.ViewWritten(); got != "" { diff --git a/http/httphi/router_test.go b/http/httphi/router_test.go index 2aef612..76989bb 100644 --- a/http/httphi/router_test.go +++ b/http/httphi/router_test.go @@ -296,7 +296,9 @@ func TestRouterSplitRequest(t *testing.T) { func staticPage(t *testing.T, page string) HandlerFunc { return func(ex *Exchange) { - n, err := io.WriteString(ex, page) + var rw ExchangeRW // Streaming APIs take the ReadWriter view. + ex.ReadWriter(&rw) + n, err := io.WriteString(&rw, page) // Handler runs on the router goroutine: Error, never Fatal. if err != nil { t.Error(err)