diff --git a/CHANGELOG.md b/CHANGELOG.md index fb1aa4e..9349738 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ - 新增独立进程并发与长连接内存测量工具，记录成功吞吐、P95/P99、服务端分配、空闲连接堆内存和回收情况；修正并行 `ns/op` 与客户端/服务端混合分配的解释。 - 加固 native HTTP/1.1 解析边界：校验 Host、重复 Content-Length、Connection token、Transfer-Encoding、控制字符和响应头名称；补充 HTTP/1.0 Keep-Alive、HEAD、204 以及异常请求测试。 - native 支持带 trailers 的 chunked 请求体，并继续通过 `MaxBodyBytes`、读超时和头部上限控制解析内存与连接占用。 +- native 响应实现 `http.Flusher`：HTTP/1.1 流式响应自动使用 chunked framing，普通响应继续自动生成 `Content-Length`。 - 补充慢请求体超时、大请求体和同连接管线请求验收，确认请求体未完整读取时不会进入业务 Handler，连续请求不会串线。 - 补充 `MaxConnections` 满载拒绝与槽位释放验收，确认请求完成后连接容量可再次使用。 - 增加 8 并发、800 次建连/断连的状态池 churn 验收，覆盖连接对象回收与容量复用。 diff --git a/README.md b/README.md index 12ad7ef..a63db2c 100644 --- a/README.md +++ b/README.md @@ -121,7 +121,7 @@ if err := app.RunWith(ctx, native.New(nil), ":8080"); err != nil { } ``` -`transport/native` 当前支持 HTTP/1.1、Keep-Alive、`Content-Length` 和 chunked 请求体、读写超时和优雅关闭；不支持的传输编码会被拒绝。它会复用连接状态和响应对象，并可通过 `MaxConnections`、`MaxConcurrentRequests` 限制内存峰值。它仍通过标准 HTTP 请求/响应接口连接业务层，适合在现有 Handler / Service 上做网络层验证。 +`transport/native` 当前支持 HTTP/1.1、Keep-Alive、`Content-Length` 和 chunked 请求体、`http.Flusher` 流式响应、读写超时和优雅关闭；不支持的传输编码会被拒绝。它会复用连接状态和响应对象，并可通过 `MaxConnections`、`MaxConcurrentRequests` 限制内存峰值。它仍通过标准 HTTP 请求/响应接口连接业务层，适合在现有 Handler / Service 上做网络层验证。 业务代码可以直接使用 `git.yuntuo.hk/k2safe/go-omni/logx` 打结构化日志，不需要依赖具体网络库： diff --git a/ROADMAP.md b/ROADMAP.md index feccd06..ec0c4eb 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 响应已支持标准 `http.Flusher`；HTTP/1.1 流式响应使用 chunked framing，普通响应继续走 `Content-Length`，HTTP/1.0 回退到关闭连接分帧。 - 已补充慢请求体超时、大请求体、HTTP/1.0 Keep-Alive、HEAD/204 和管线请求验收；Linux、慢客户端规模、TLS 和持续运行压测仍待在目标环境验证。 - 已验证 `MaxConnections` 满载时快速拒绝新连接、请求完成后释放容量；后续仍需在目标部署环境观察文件描述符、内核 socket 缓冲和反向代理联动。 - 已用 8 个并发 worker 完成 800 次建连/断连 race 验收；真实部署仍需补文件描述符和跨平台 socket 行为观测。 diff --git a/transport/README.md b/transport/README.md index 649b2a0..85a5d44 100644 --- a/transport/README.md +++ b/transport/README.md @@ -16,6 +16,8 @@ registerRoutes(application) // 与 omni.New() 复用同一份业务注册代码 `transport/native` 是第一版自研 HTTP/1.1 服务端：它自己管理 TCP listener、连接、Keep-Alive、请求解析、响应写回和优雅关闭，再把请求交给现有 `http.Handler`。请求体支持 `Content-Length` 和 chunked，并统一受 `MaxBodyBytes` 限制；其他传输编码会被拒绝。它仍使用标准 `http.Request` / `http.ResponseWriter` 作为业务桥接，因此不会影响现有 Handler / Service。 +业务 Handler 需要流式返回时可以断言 `http.Flusher` 并调用 `Flush()`；HTTP/1.1 会自动使用 chunked 响应，普通未 Flush 的响应仍自动生成 `Content-Length`。HTTP/1.0 没有 chunked framing，会退回到写完后关闭连接。 + 服务端不是把连接交给一个全局串行 worker，而是每条连接独立处理请求；Keep-Alive 负责复用客户端连接，`Server` 内部的 `sync.Pool` 负责复用小型 Reader、响应对象和连接状态。Reader 默认只有 4 KiB，需要处理大请求头时按需增长。需要保护内存上限时，可以设置 `MaxConnections` 和 `MaxConcurrentRequests`；超出连接上限的连接会被关闭，超出请求并发上限的请求返回 503，不会无限排队。 这个设计借鉴了 Nginx 的资源边界，但没有照搬它的进程模型：Nginx 用少量 worker 进程和事件机制管理大量连接，并用 `worker_connections` 限制每个 worker 的同时连接数；Go runtime 自己已经用 epoll / kqueue 等网络轮询器唤醒阻塞的 goroutine，因此 native 采用“accept loop + 每连接轻量 goroutine + 请求并发闸门”的等价实现。这样业务 Handler 可以继续执行同步代码，慢请求不会阻塞其他连接；只有在真实压测证明 goroutine 调度或协议状态机成为瓶颈时，才需要继续下沉为手写事件循环。 diff --git a/transport/native/server.go b/transport/native/server.go index 82fde3b..7c5f9f6 100644 --- a/transport/native/server.go +++ b/transport/native/server.go @@ -334,6 +334,7 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) { state.active.Store(true) writer := s.writerPool.Get().(*responseWriter) writer.reset(request.Method) + writer.configure(request, state.conn, closeAfter, s.WriteTimeout) requestSlot := false if slots := s.requestSlots; slots != nil { select { @@ -378,10 +379,11 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) { if requestSlot { <-s.requestSlots } + shouldClose := closeAfter || writer.closeAfter || closing writer.reset("") s.writerPool.Put(writer) state.active.Store(false) - if closeAfter || closing { + if shouldClose { return } } @@ -719,11 +721,20 @@ func writeSimple(conn net.Conn, status int, body string, closeAfter bool, timeou } type responseWriter struct { - header http.Header - status int - wroteHeader bool - body bytes.Buffer - requestMethod string + header http.Header + status int + wroteHeader bool + body bytes.Buffer + requestMethod string + protoMajor int + protoMinor int + conn net.Conn + closeAfter bool + writeTimeout time.Duration + streaming bool + chunked bool + streamingClose bool + writeErr error } func newResponseWriter(method string) *responseWriter { @@ -742,6 +753,23 @@ func (w *responseWriter) reset(method string) { w.body.Reset() } w.requestMethod = method + w.protoMajor = 0 + w.protoMinor = 0 + w.conn = nil + w.closeAfter = false + w.writeTimeout = 0 + w.streaming = false + w.chunked = false + w.streamingClose = false + w.writeErr = nil +} + +func (w *responseWriter) configure(request *http.Request, conn net.Conn, closeAfter bool, timeout time.Duration) { + w.protoMajor = request.ProtoMajor + w.protoMinor = request.ProtoMinor + w.conn = conn + w.closeAfter = closeAfter + w.writeTimeout = timeout } func (w *responseWriter) Header() http.Header { return w.header } @@ -755,13 +783,121 @@ func (w *responseWriter) WriteHeader(status int) { } func (w *responseWriter) Write(data []byte) (int, error) { + if w.writeErr != nil { + return 0, w.writeErr + } if !w.wroteHeader { w.WriteHeader(http.StatusOK) } + if w.streaming { + written, err := w.writeStream(data) + if err != nil { + w.writeErr = err + } + return written, err + } return w.body.Write(data) } +// Flush implements http.Flusher. Once a response is flushed, HTTP/1.1 uses +// chunked framing unless the handler supplied Content-Length. HTTP/1.0 falls +// back to close-delimited output because it has no chunked framing. +func (w *responseWriter) Flush() { + if w.writeErr != nil || w.streaming || w.requestMethod == http.MethodHead || w.status == http.StatusNoContent || + w.status == http.StatusResetContent || w.status == http.StatusNotModified || (w.status >= 100 && w.status < 200) { + return + } + if err := w.beginStream(); err != nil { + w.writeErr = err + } +} + +func (w *responseWriter) beginStream() error { + if w.conn == nil { + return errors.New("go-omni/native: response stream is not connected") + } + if !w.wroteHeader { + w.status = http.StatusOK + w.wroteHeader = true + } + if w.requestMethod == http.MethodHead || w.status == http.StatusNoContent || w.status == http.StatusResetContent || + w.status == http.StatusNotModified || (w.status >= 100 && w.status < 200) { + return nil + } + contentLength := w.header.Get("Content-Length") + w.chunked = contentLength == "" && w.protoMajor == 1 && w.protoMinor == 1 + if w.chunked { + w.header.Set("Transfer-Encoding", "chunked") + } else if contentLength == "" { + w.streamingClose = true + } + closeAfter := w.closeAfter || w.streamingClose + if closeAfter { + w.header.Set("Connection", "close") + } + w.closeAfter = closeAfter + if err := w.writeHeaders(w.conn); err != nil { + return err + } + w.streaming = true + buffered := w.body.Bytes() + w.body.Reset() + if len(buffered) == 0 { + return nil + } + _, err := w.writeStream(buffered) + return err +} + +func (w *responseWriter) writeStream(data []byte) (int, error) { + if len(data) == 0 { + return 0, nil + } + if w.conn == nil { + return 0, errors.New("go-omni/native: response stream is not connected") + } + if w.chunked { + var prefix [32]byte + framed := strconv.AppendInt(prefix[:0], int64(len(data)), 16) + framed = append(framed, '\r', '\n') + if err := w.writeBytes(framed); err != nil { + return 0, err + } + if err := w.writeBytes(data); err != nil { + return 0, err + } + if err := w.writeBytes([]byte("\r\n")); err != nil { + return 0, err + } + return len(data), nil + } + if err := w.writeBytes(data); err != nil { + return 0, err + } + return len(data), nil +} + +func (w *responseWriter) writeBytes(data []byte) error { + if w.writeTimeout > 0 { + _ = w.conn.SetWriteDeadline(time.Now().Add(w.writeTimeout)) + } + return writeAll(w.conn, data) +} + func (w *responseWriter) writeTo(conn net.Conn, closeAfter bool, timeout time.Duration) error { + if w.streaming { + if w.writeErr != nil { + _ = conn.SetWriteDeadline(time.Time{}) + return w.writeErr + } + if w.chunked { + if err := w.writeBytes([]byte("0\r\n\r\n")); err != nil { + w.writeErr = err + } + } + _ = conn.SetWriteDeadline(time.Time{}) + return w.writeErr + } if timeout > 0 { _ = conn.SetWriteDeadline(time.Now().Add(timeout)) } @@ -775,7 +911,7 @@ func (w *responseWriter) writeTo(conn net.Conn, closeAfter bool, timeout time.Du if noBody { body = nil } - if w.header.Get("Content-Length") == "" && w.status >= 200 && w.status != http.StatusNoContent && w.status != http.StatusNotModified && w.status != http.StatusResetContent { + if w.header.Get("Content-Length") == "" && w.header.Get("Transfer-Encoding") == "" && w.status >= 200 && w.status != http.StatusNoContent && w.status != http.StatusNotModified && w.status != http.StatusResetContent { w.header.Set("Content-Length", strconv.Itoa(len(w.body.Bytes()))) } if closeAfter { @@ -808,7 +944,56 @@ func (w *responseWriter) writeTo(conn net.Conn, closeAfter bool, timeout time.Du } output.WriteString("\r\n") output.Write(body) - _, err := conn.Write(output.Bytes()) + err := writeAll(conn, output.Bytes()) _ = conn.SetWriteDeadline(time.Time{}) return err } + +func (w *responseWriter) writeHeaders(conn net.Conn) error { + if w.writeTimeout > 0 { + _ = conn.SetWriteDeadline(time.Now().Add(w.writeTimeout)) + } + keys := make([]string, 0, len(w.header)) + for key := range w.header { + keys = append(keys, key) + } + sort.Strings(keys) + var output bytes.Buffer + output.WriteString("HTTP/1.1 ") + output.WriteString(strconv.Itoa(w.status)) + output.WriteByte(' ') + output.WriteString(http.StatusText(w.status)) + output.WriteString("\r\n") + for _, key := range keys { + if !validToken(key) { + return errors.New("invalid response header name") + } + for _, value := range w.header.Values(key) { + if !validHeaderValue(value) { + return errors.New("invalid response header") + } + output.WriteString(key) + output.WriteString(": ") + output.WriteString(value) + output.WriteString("\r\n") + } + } + output.WriteString("\r\n") + return writeAll(conn, output.Bytes()) +} + +func writeAll(conn net.Conn, data []byte) error { + for len(data) > 0 { + written, err := conn.Write(data) + if written > 0 { + data = data[written:] + } + if err != nil { + return err + } + if written == 0 { + return io.ErrShortWrite + } + } + return nil +} diff --git a/transport/native/server_test.go b/transport/native/server_test.go index 9263748..efc0c02 100644 --- a/transport/native/server_test.go +++ b/transport/native/server_test.go @@ -225,6 +225,60 @@ func TestServerPipelinedRequestsAndLargeBody(t *testing.T) { } } +func TestServerFlushesChunkedResponses(t *testing.T) { + flushed := make(chan struct{}) + release := make(chan struct{}) + application := omni.NewWithRouter(netrouter.New()) + application.GET("/stream", func(ctx *omni.Context) error { + flusher, ok := ctx.Raw().(http.Flusher) + if !ok { + return fmt.Errorf("native response writer does not implement http.Flusher") + } + ctx.Raw().Header().Set("Content-Type", "text/plain") + _, _ = io.WriteString(ctx.Raw(), "one\n") + flusher.Flush() + close(flushed) + <-release + _, _ = io.WriteString(ctx.Raw(), "two\n") + return nil + }) + 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() + reader := bufio.NewReader(conn) + _, _ = io.WriteString(conn, "GET /stream HTTP/1.1\r\nHost: test\r\nConnection: close\r\n\r\n") + response, err := http.ReadResponse(reader, &http.Request{Method: http.MethodGet}) + if err != nil { + t.Fatal(err) + } + if response.StatusCode != http.StatusOK || response.TransferEncoding == nil || response.TransferEncoding[0] != "chunked" { + response.Body.Close() + t.Fatalf("expected chunked response, status=%d transfer=%v", response.StatusCode, response.TransferEncoding) + } + select { + case <-flushed: + case <-time.After(time.Second): + response.Body.Close() + t.Fatal("handler did not flush response") + } + first := make([]byte, len("one\n")) + if _, err := io.ReadFull(response.Body, first); err != nil || string(first) != "one\n" { + response.Body.Close() + t.Fatalf("unexpected first stream chunk: %q error=%v", first, err) + } + close(release) + rest, err := io.ReadAll(response.Body) + response.Body.Close() + if err != nil || string(rest) != "two\n" { + t.Fatalf("unexpected remaining stream body: %q error=%v", rest, err) + } +} + func TestServerProtocolBoundaries(t *testing.T) { application := omni.NewWithRouter(netrouter.New()) application.GET("/health", func(ctx *omni.Context) error { return ctx.OK("ok") })