diff --git a/CHANGELOG.md b/CHANGELOG.md index ac2a526..7d163f0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ## Unreleased +- native 请求体改为流式读取,Content-Length 与 chunked body 不再在进入 Handler 前整体复制到内存;保留大小限制、读超时、trailer 校验和 Keep-Alive 排空。 - 统一标准库与 native 的真实 TCP benchmark,增加固定并发档位、响应状态与读写错误检查。 - 新增独立进程并发与长连接内存测量工具,记录成功吞吐、P95/P99、服务端分配、空闲连接堆内存和回收情况;修正并行 `ns/op` 与客户端/服务端混合分配的解释。 - 加固 native HTTP/1.1 解析边界:校验 Host、重复 Content-Length、Connection token、Transfer-Encoding、控制字符和响应头名称;补充 HTTP/1.0 Keep-Alive、HEAD、204 以及异常请求测试。 @@ -10,13 +11,13 @@ - native 在长 Handler 运行期间监测客户端断开并取消请求 context,流式写失败同步取消;短请求延迟启动监测,管线数据保留给下一次请求。补充断开、监测退出和长连接复用验收。 - native 增加 `ServeTLS` / `RunTLS` 和 `TLSConfig` 启动入口,App.RunWith 可以直接启用 TLS;补充真实 TLS 客户端握手、请求和优雅退出验收。 - native 请求补齐 `Request.TLS` 连接状态;Linux/arm64 容器通过全量 race、vet、build,以及 TLS 并发、Keep-Alive、慢握手和连接上限回收验收。 -- 补充慢请求体超时、大请求体和同连接管线请求验收,确认请求体未完整读取时不会进入业务 Handler,连续请求不会串线。 +- 补充慢请求体超时、大请求体和同连接管线请求验收,确认流式 body 读取超时会取消请求并关闭连接,连续请求不会串线。 - 补充 `MaxConnections` 满载拒绝与槽位释放验收,确认请求完成后连接容量可再次使用。 - 增加高频建连/断连的状态池 churn 验收,覆盖连接对象回收与容量复用。 ## v0.6.0-preview -- native 网络层完成慢请求头、慢 Content-Length / chunked body、慢 TLS 握手和 TLS 请求元数据验收,未完整读取的请求不会进入业务 Handler。 +- native 网络层完成流式 Content-Length / chunked body、慢请求头、慢 body、慢 TLS 握手和 TLS 请求元数据验收;Handler 可在 body 未完全到达时开始读取,超时、畸形和超限 body 会被拒绝并关闭连接。 - 将 native 连接 churn race 验收扩展为 16 个 worker、3,200 次建连/断连,并覆盖明文与 TLS 的连接上限、Keep-Alive 和槽位回收。 - perfcheck 支持最长 30 分钟持续场景,新增 Linux 文件描述符采样;固定记录服务端堆、goroutine、连接数、accept 数和 FD 回收状态。 - 新增 `scripts/verifyLinux.py`,在 `golang:1.25` Linux 容器内统一执行 race、vet、build 和可选 TLS / 非 TLS 压测。 diff --git a/ROADMAP.md b/ROADMAP.md index b764c77..f684b61 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -42,6 +42,7 @@ - 标准库与 native 共用 TCP benchmark,固定 1 / 32 / 128 并发;相同应用、客户端、超时和响应检查,避免两个测试入口逐渐失去可比性。 - 新增独立服务进程压测 `internal/perfcheck`,记录成功吞吐、P50/P95/P99、服务端分配、采样堆峰值,以及 128 / 1,024 条空闲长连接的堆内存、goroutine 和断开后的回收情况。原始报告保存在 `performance/tcpBaseline.json`。 - native 已支持受 `MaxBodyBytes` 约束的 `Content-Length` / chunked 请求体,仍使用标准 `http.Request` / `http.ResponseWriter` 作为桥接;解析器、连接抽象和生产压测将在后续阶段继续下沉。 +- native 请求体已改为流式读取,Handler 可边到边处理,服务端只在 Handler 返回后排空剩余 body,超时、畸形和超限 body 会取消请求并关闭连接。 - native 响应已支持标准 `http.Flusher`;HTTP/1.1 流式响应使用 chunked framing,普通响应继续走 `Content-Length`,HTTP/1.0 回退到关闭连接分帧。 - native 长请求已补客户端断开及流式写失败取消,监测退出后再复用 Reader;已验收普通/流式断开、慢 Handler 后连续 Keep-Alive 和管线数据不被误消费。短请求延迟启动监测,已有管线数据时取消依赖后续写失败或 Handler 返回。 - 已补充慢请求头、慢 Content-Length body、慢 chunked body、慢 TLS 握手、大请求体、HTTP/1.0 Keep-Alive、HEAD/204 和管线请求验收;Linux/arm64 容器已通过全量 race、vet、build,以及 3 秒 TLS 并发/空闲连接矩阵。原始报告见 [tlsBaselineLinux.json](performance/tlsBaselineLinux.json)。 diff --git a/transport/native/server.go b/transport/native/server.go index 6526968..92f6a4e 100644 --- a/transport/native/server.go +++ b/transport/native/server.go @@ -410,12 +410,49 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) { } } handler := s.handler - disconnectWatch := newClientDisconnectWatch(reader, state.conn, cancel) + var disconnectWatch *clientDisconnectWatch + if body, ok := request.Body.(*requestBody); ok { + // A body reader and the disconnect watcher must never read the + // bufio.Reader concurrently. The body starts the watcher once it + // reaches EOF; body read errors cancel the request directly. + body.cancel = cancel + body.onComplete = func() { + disconnectWatch = newClientDisconnectWatch(reader, state.conn, cancel) + } + } else { + disconnectWatch = newClientDisconnectWatch(reader, state.conn, cancel) + } handler.ServeHTTP(writer, request) - disconnectWatch.stop() + bodyErr := request.Body.Close() + if disconnectWatch != nil { + disconnectWatch.stop() + } cancel() - closing := s.isClosing() || state.closeWhenIdle.Load() + closing := s.isClosing() || state.closeWhenIdle.Load() || bodyErr != nil + if bodyErr != nil { + var protocolErr *protocolError + if !errors.As(bodyErr, &protocolErr) { + if requestSlot { + <-s.requestSlots + } + writer.reset("") + s.writerPool.Put(writer) + state.active.Store(false) + return + } + // Request bodies are streamed into the handler, so a malformed + // chunk or an over-limit body can be discovered while draining + // after the handler returns. Replace the buffered response with + // the same protocol error that eager parsing used to return. + if !writer.streaming { + writer.reset(request.Method) + writer.configure(request, state.conn, true, s.WriteTimeout, cancel) + writer.Header().Set("Content-Type", "text/plain; charset=utf-8") + writer.WriteHeader(protocolErr.status) + writer.body.WriteString(protocolErr.message + "\n") + } + } if err := writer.writeTo(state.conn, closeAfter || closing, s.WriteTimeout); err != nil { if requestSlot { <-s.requestSlots @@ -577,27 +614,23 @@ func readRequest(reader *bufio.Reader, conn net.Conn, maxHeaderBytes int, maxBod if !chunked && contentLength > maxBodyBytes { return nil, true, &protocolError{status: http.StatusRequestEntityTooLarge, message: "request body too large"} } - var body []byte + var body io.ReadCloser = http.NoBody var trailer http.Header if chunked { - body, trailer, err = readChunkedBody(reader, maxBodyBytes, setBodyDeadline, readLine) - if err != nil { - return nil, true, err + trailer = make(http.Header) + body = &requestBody{ + reader: reader, maxBytes: maxBodyBytes, chunked: true, + trailer: trailer, setDeadline: setBodyDeadline, readLine: readLine, } // net/http uses -1 to indicate that the decoded length was not known - // from the request headers. The body is already buffered for the - // handler, but retaining this value keeps the request contract - // compatible with the standard server. + // from the request headers. contentLength = -1 } else { - body = make([]byte, contentLength) if contentLength > 0 { if setBodyDeadline != nil { setBodyDeadline() } - if _, err := io.ReadFull(reader, body); err != nil { - return nil, true, err - } + body = &requestBody{reader: reader, remaining: contentLength, setDeadline: setBodyDeadline} } } request := &http.Request{ @@ -607,7 +640,7 @@ func readRequest(reader *bufio.Reader, conn net.Conn, maxHeaderBytes int, maxBod ProtoMajor: protoMajor, ProtoMinor: protoMinor, Header: header, - Body: io.NopCloser(bytes.NewReader(body)), + Body: body, ContentLength: contentLength, Host: header.Get("Host"), RequestURI: target, @@ -627,78 +660,199 @@ func readRequest(reader *bufio.Reader, conn net.Conn, maxHeaderBytes int, maxBod return request, closeAfter, nil } -func readChunkedBody(reader *bufio.Reader, maxBodyBytes int64, setBodyDeadline func(), readLine func() (string, error)) ([]byte, http.Header, error) { - var body bytes.Buffer - initialCapacity := int64(64 << 10) - if maxBodyBytes < initialCapacity { - initialCapacity = maxBodyBytes - } - if initialCapacity > 0 && initialCapacity <= int64(int(^uint(0)>>1)) { - body.Grow(int(initialCapacity)) - } +type requestBody struct { + reader *bufio.Reader + remaining int64 + maxBytes int64 + readBytes int64 + chunked bool + chunkRemain int64 + needSuffix bool + trailer http.Header + readLine func() (string, error) + setDeadline func() + started bool + cancel context.CancelFunc + onComplete func() + done bool + closed bool + err error +} +func (b *requestBody) Read(p []byte) (int, error) { + if len(p) == 0 { + return 0, nil + } + if b.err != nil { + return 0, b.err + } + if b.done { + return 0, io.EOF + } + if !b.chunked { + return b.readContentLength(p) + } + return b.readChunked(p) +} + +func (b *requestBody) readContentLength(p []byte) (int, error) { + if b.remaining == 0 { + b.complete() + return 0, io.EOF + } + if !b.started { + b.started = true + if b.setDeadline != nil { + b.setDeadline() + } + } + if int64(len(p)) > b.remaining { + p = p[:b.remaining] + } + n, err := b.reader.Read(p) + b.remaining -= int64(n) + b.readBytes += int64(n) + if b.remaining == 0 { + b.complete() + if n > 0 { + return n, nil + } + return 0, io.EOF + } + if err != nil { + b.fail(err) + } + return n, err +} + +func (b *requestBody) readChunked(p []byte) (int, error) { for { - line, err := readLine() - if err != nil { - return nil, nil, err - } - chunkSizeText := line - if extension := strings.IndexByte(chunkSizeText, ';'); extension >= 0 { - chunkSizeText = chunkSizeText[:extension] - } - chunkSizeText = strings.TrimSpace(chunkSizeText) - if chunkSizeText == "" || !validHexNumber(chunkSizeText) { - return nil, nil, &protocolError{status: http.StatusBadRequest, message: "invalid chunk size"} - } - chunkSize, parseErr := strconv.ParseUint(chunkSizeText, 16, 63) - if parseErr != nil { - return nil, nil, &protocolError{status: http.StatusBadRequest, message: "invalid chunk size"} - } - if chunkSize == 0 { - trailer := make(http.Header) - for { - line, err = readLine() - if err != nil { - return nil, nil, err + if b.chunkRemain == 0 { + if b.needSuffix { + var suffix [2]byte + if _, err := io.ReadFull(b.reader, suffix[:]); err != nil { + b.fail(err) + return 0, err } - if line == "" { - return body.Bytes(), trailer, nil + if suffix[0] != '\r' || suffix[1] != '\n' { + err := &protocolError{status: http.StatusBadRequest, message: "malformed chunk terminator"} + b.fail(err) + return 0, err } - if strings.HasPrefix(line, " ") || strings.HasPrefix(line, "\t") { - return nil, nil, &protocolError{status: http.StatusBadRequest, message: "folded trailers are not supported"} - } - index := strings.IndexByte(line, ':') - if index <= 0 { - return nil, nil, &protocolError{status: http.StatusBadRequest, message: "malformed trailer"} - } - name := strings.TrimSpace(line[:index]) - value := strings.TrimSpace(line[index+1:]) - if !validToken(name) || !validHeaderValue(value) { - return nil, nil, &protocolError{status: http.StatusBadRequest, message: "malformed trailer"} - } - trailer.Add(name, value) + b.needSuffix = false + } + if err := b.beginChunk(); err != nil { + b.fail(err) + return 0, err + } + if b.done { + return 0, io.EOF } } - if int64(body.Len()) > maxBodyBytes-int64(chunkSize) { - return nil, nil, &protocolError{status: http.StatusRequestEntityTooLarge, message: "request body too large"} + if int64(len(p)) > b.chunkRemain { + p = p[:b.chunkRemain] } - if setBodyDeadline != nil { - setBodyDeadline() + n, err := b.reader.Read(p) + b.chunkRemain -= int64(n) + b.readBytes += int64(n) + if b.chunkRemain == 0 { + b.needSuffix = true } - if _, err := io.CopyN(&body, reader, int64(chunkSize)); err != nil { - return nil, nil, err + if n > 0 { + if err != nil && b.chunkRemain > 0 { + b.fail(err) + return n, err + } + return n, nil } - if setBodyDeadline != nil { - setBodyDeadline() + if err != nil { + b.fail(err) + return 0, err } - var suffix [2]byte - if _, err := io.ReadFull(reader, suffix[:]); err != nil { - return nil, nil, err - } - if suffix[0] != '\r' || suffix[1] != '\n' { - return nil, nil, &protocolError{status: http.StatusBadRequest, message: "malformed chunk terminator"} + } +} + +func (b *requestBody) beginChunk() error { + if b.setDeadline != nil { + b.setDeadline() + } + line, err := b.readLine() + if err != nil { + return err + } + chunkSizeText := line + if extension := strings.IndexByte(chunkSizeText, ';'); extension >= 0 { + chunkSizeText = chunkSizeText[:extension] + } + chunkSizeText = strings.TrimSpace(chunkSizeText) + if chunkSizeText == "" || !validHexNumber(chunkSizeText) { + return &protocolError{status: http.StatusBadRequest, message: "invalid chunk size"} + } + chunkSize, parseErr := strconv.ParseUint(chunkSizeText, 16, 63) + if parseErr != nil { + return &protocolError{status: http.StatusBadRequest, message: "invalid chunk size"} + } + if chunkSize == 0 { + for { + line, err = b.readLine() + if err != nil { + return err + } + if line == "" { + b.complete() + return nil + } + if strings.HasPrefix(line, " ") || strings.HasPrefix(line, "\t") { + return &protocolError{status: http.StatusBadRequest, message: "folded trailers are not supported"} + } + index := strings.IndexByte(line, ':') + if index <= 0 { + return &protocolError{status: http.StatusBadRequest, message: "malformed trailer"} + } + name := strings.TrimSpace(line[:index]) + value := strings.TrimSpace(line[index+1:]) + if !validToken(name) || !validHeaderValue(value) { + return &protocolError{status: http.StatusBadRequest, message: "malformed trailer"} + } + b.trailer.Add(name, value) } } + if int64(chunkSize) > b.maxBytes-b.readBytes { + return &protocolError{status: http.StatusRequestEntityTooLarge, message: "request body too large"} + } + b.chunkRemain = int64(chunkSize) + return nil +} + +func (b *requestBody) Close() error { + if b.closed { + return b.err + } + b.closed = true + if !b.done && b.err == nil { + _, _ = io.Copy(io.Discard, b) + } + return b.err +} + +func (b *requestBody) complete() { + if b.done { + return + } + b.done = true + if b.onComplete != nil { + b.onComplete() + } +} + +func (b *requestBody) fail(err error) { + if b.err == nil { + b.err = err + } + b.done = true + if b.cancel != nil { + b.cancel() + } } func validHexNumber(value string) bool { diff --git a/transport/native/server_test.go b/transport/native/server_test.go index 2ce3edd..aa9a2e5 100644 --- a/transport/native/server_test.go +++ b/transport/native/server_test.go @@ -107,6 +107,47 @@ func TestServerRejectsMalformedBodyAndLimits(t *testing.T) { } } +func TestServerStreamsContentLengthBodyToHandler(t *testing.T) { + entered := make(chan struct{}) + application := omni.NewWithRouter(netrouter.New()) + application.POST("/body", func(ctx *omni.Context) error { + close(entered) + body, err := io.ReadAll(ctx.Request().Body) + if err != nil { + return err + } + return ctx.OK(map[string]string{"body": string(body)}) + }) + server, listener, done := startServer(t, application) + defer stopServer(t, server, listener, done) + + conn, err := net.Dial("tcp", listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + if _, err := io.WriteString(conn, "POST /body HTTP/1.1\r\nHost: test\r\nContent-Length: 5\r\n\r\nhel"); err != nil { + t.Fatal(err) + } + select { + case <-entered: + case <-time.After(time.Second): + t.Fatal("handler did not start before the complete body arrived") + } + if _, err := io.WriteString(conn, "lo"); err != nil { + t.Fatal(err) + } + response, err := http.ReadResponse(bufio.NewReader(conn), &http.Request{Method: http.MethodPost}) + if err != nil { + t.Fatal(err) + } + body, err := io.ReadAll(response.Body) + response.Body.Close() + if err != nil || response.StatusCode != http.StatusOK || !strings.Contains(string(body), `"body":"hello"`) { + t.Fatalf("unexpected streamed body response: status=%d body=%s error=%v", response.StatusCode, body, err) + } +} + func TestServerChunkedBodyAndTrailers(t *testing.T) { application := omni.NewWithRouter(netrouter.New()) application.POST("/body", func(ctx *omni.Context) error { @@ -154,11 +195,18 @@ func TestServerChunkedBodyLimit(t *testing.T) { } } -func TestServerSlowBodyTimesOutBeforeHandler(t *testing.T) { +func TestServerSlowBodyTimesOutWithoutSuccessfulHandler(t *testing.T) { var handlerCalls atomic.Int32 application := omni.NewWithRouter(netrouter.New()) application.POST("/body", func(ctx *omni.Context) error { + body, err := io.ReadAll(ctx.Request().Body) + if err != nil { + return err + } handlerCalls.Add(1) + if len(body) != 4 { + return fmt.Errorf("unexpected body length %d", len(body)) + } return ctx.OK("ok") }) server := native.New(application) @@ -178,14 +226,10 @@ func TestServerSlowBodyTimesOutBeforeHandler(t *testing.T) { t.Fatal(err) } defer conn.Close() - conn.SetReadDeadline(time.Now().Add(time.Second)) _, _ = io.WriteString(conn, "POST /body HTTP/1.1\r\nHost: test\r\nContent-Length: 4\r\n\r\na") - buffer := make([]byte, 1) - if _, err := conn.Read(buffer); err == nil { - t.Fatal("expected slow request body to time out and close") - } + requirePeerClose(t, conn) if handlerCalls.Load() != 0 { - t.Fatal("handler ran before the request body was complete") + t.Fatal("handler reported success before the request body was complete") } } diff --git a/transport/native/slow_client_test.go b/transport/native/slow_client_test.go index 5ab88fa..e4b61b5 100644 --- a/transport/native/slow_client_test.go +++ b/transport/native/slow_client_test.go @@ -43,8 +43,10 @@ func TestServerSlowHeadersCloseBeforeHandler(t *testing.T) { func TestServerSlowChunkedBodyClosesBeforeHandler(t *testing.T) { var handlerCalls atomic.Int32 - server := native.New(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { - handlerCalls.Add(1) + server := native.New(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + if _, err := io.ReadAll(r.Body); err == nil { + handlerCalls.Add(1) + } })) server.ReadTimeout = 40 * time.Millisecond listener, err := net.Listen("tcp", "127.0.0.1:0")