diff --git a/CHANGELOG.md b/CHANGELOG.md index 86dc777..5f8450b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ ## Unreleased - 新增 transport-neutral `metrics` 包，支持请求数、错误数、in-flight、累计耗时和状态分布，并可直接注册 `/metrics` 接口。 +- native 新增 `Server.Stats()`，提供 accepted connections、active connections 和 active requests 生命周期指标。 - native 请求体改为流式读取，Content-Length 与 chunked body 不再在进入 Handler 前整体复制到内存；保留大小限制、读超时、trailer 校验和 Keep-Alive 排空。 - 统一标准库与 native 的真实 TCP benchmark，增加固定并发档位、响应状态与读写错误检查。 - 新增独立进程并发与长连接内存测量工具，记录成功吞吐、P95/P99、服务端分配、空闲连接堆内存和回收情况；修正并行 `ns/op` 与客户端/服务端混合分配的解释。 diff --git a/README.md b/README.md index f3125e3..c13420d 100644 --- a/README.md +++ b/README.md @@ -124,6 +124,8 @@ if err := app.RunWith(ctx, native.New(nil), ":8080"); err != nil { `transport/native` 当前支持 HTTP/1.1、Keep-Alive、`Content-Length` 和 chunked 请求体、`http.Flusher` 流式响应、可选 TLS、读写超时和优雅关闭；不支持的传输编码会被拒绝。它会复用连接状态和响应对象，并可通过 `MaxConnections`、`MaxConcurrentRequests` 限制内存峰值。它仍通过标准 HTTP 请求/响应接口连接业务层，适合在现有 Handler / Service 上做网络层验证。 +自研网络层还提供 `server.Stats()`，可读取 accepted、active connections 和 active requests，与上面的业务请求 metrics 组合后能区分连接堆积和 Handler 慢请求。 + 业务代码可以直接使用 `git.yuntuo.hk/k2safe/go-omni/logx` 打结构化日志，不需要依赖具体网络库： ```go diff --git a/transport/README.md b/transport/README.md index 6d2249a..ba8a04b 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。 +native 提供 `server.Stats()` 读取已接受连接数、当前连接数和当前请求数，适合和 `metrics.Metrics` 的业务请求指标一起采集；读取不会阻塞连接生命周期。 + 业务 Handler 需要流式返回时可以断言 `http.Flusher` 并调用 `Flush()`；HTTP/1.1 会自动使用 chunked 响应，普通未 Flush 的响应仍自动生成 `Content-Length`。HTTP/1.0 没有 chunked framing，会退回到写完后关闭连接。 Handler 运行超过约 5 ms 时，native 启动一次阻塞读取监测；读到 EOF 或连接错误会取消 `Request.Context()`，流式写入失败也会触发取消。Handler 返回后先停止监测，再复用连接和 Reader；普通短请求不会启动监测 goroutine。 diff --git a/transport/native/server.go b/transport/native/server.go index 92f6a4e..8096c09 100644 --- a/transport/native/server.go +++ b/transport/native/server.go @@ -62,17 +62,42 @@ type Server struct { wg sync.WaitGroup connSlots chan struct{} requestSlots chan struct{} + accepted atomic.Uint64 + activeConns atomic.Int64 + activeRequests atomic.Int64 readerPool sync.Pool writerPool sync.Pool connectionPool sync.Pool } +// Stats is a point-in-time view of the native server lifecycle. Counters are +// monotonic unless their name starts with Active. +type Stats struct { + AcceptedConnections uint64 `json:"acceptedConnections"` + ActiveConnections int64 `json:"activeConnections"` + ActiveRequests int64 `json:"activeRequests"` +} + +// Stats returns native connection and request counters without taking the +// server lifecycle mutex, so it is safe to call from a metrics endpoint. +func (s *Server) Stats() Stats { + if s == nil { + return Stats{} + } + return Stats{ + AcceptedConnections: s.accepted.Load(), + ActiveConnections: s.activeConns.Load(), + ActiveRequests: s.activeRequests.Load(), + } +} + type connection struct { server *Server conn net.Conn active atomic.Bool closeWhenIdle atomic.Bool closed atomic.Bool + counted bool slotHeld bool closeDone chan struct{} } @@ -222,6 +247,7 @@ func (s *Server) Serve(ctx context.Context, listener net.Listener) error { serveErr = err break } + s.accepted.Add(1) if slots := s.connSlots; slots != nil { select { case slots <- struct{}{}: @@ -236,8 +262,10 @@ func (s *Server) Serve(ctx context.Context, listener net.Listener) error { state.active.Store(false) state.closeWhenIdle.Store(false) state.closed.Store(false) + state.counted = true state.slotHeld = s.connSlots != nil state.closeDone = make(chan struct{}) + s.activeConns.Add(1) s.mu.Lock() if s.closing { s.mu.Unlock() @@ -379,6 +407,7 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) { requestContext, cancel := context.WithCancel(baseCtx) request = request.WithContext(requestContext) state.active.Store(true) + s.activeRequests.Add(1) writer := s.writerPool.Get().(*responseWriter) writer.reset(request.Method) writer.configure(request, state.conn, closeAfter, s.WriteTimeout, cancel) @@ -392,6 +421,7 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) { writer.reset("") s.writerPool.Put(writer) state.active.Store(false) + s.activeRequests.Add(-1) return default: writer.Header().Set("Content-Type", "text/plain; charset=utf-8") @@ -403,6 +433,7 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) { writer.reset("") s.writerPool.Put(writer) state.active.Store(false) + s.activeRequests.Add(-1) if closeAfter || closing { return } @@ -439,6 +470,7 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) { writer.reset("") s.writerPool.Put(writer) state.active.Store(false) + s.activeRequests.Add(-1) return } // Request bodies are streamed into the handler, so a malformed @@ -460,6 +492,7 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) { writer.reset("") s.writerPool.Put(writer) state.active.Store(false) + s.activeRequests.Add(-1) return } if requestSlot { @@ -469,6 +502,7 @@ func (s *Server) serveConnection(baseCtx context.Context, state *connection) { writer.reset("") s.writerPool.Put(writer) state.active.Store(false) + s.activeRequests.Add(-1) if shouldClose { return } @@ -482,6 +516,10 @@ func (c *connection) close() { c.server.mu.Lock() delete(c.server.conns, c) c.server.mu.Unlock() + if c.counted { + c.server.activeConns.Add(-1) + c.counted = false + } } } diff --git a/transport/native/server_test.go b/transport/native/server_test.go index aa9a2e5..fc41590 100644 --- a/transport/native/server_test.go +++ b/transport/native/server_test.go @@ -789,6 +789,57 @@ func TestServerConnectionLimitReleasesAfterRequest(t *testing.T) { } } +func TestServerStatsTrackLifecycle(t *testing.T) { + started := make(chan struct{}) + release := make(chan struct{}) + application := omni.NewWithRouter(netrouter.New()) + application.GET("/hold", func(ctx *omni.Context) error { + close(started) + <-release + return ctx.OK("done") + }) + 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() + waitForStats := func(check func(native.Stats) bool) { + t.Helper() + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + if check(server.Stats()) { + return + } + time.Sleep(time.Millisecond) + } + t.Fatalf("stats condition not reached: %+v", server.Stats()) + } + waitForStats(func(stats native.Stats) bool { return stats.ActiveConnections == 1 && stats.AcceptedConnections >= 1 }) + if _, err := io.WriteString(conn, "GET /hold HTTP/1.1\r\nHost: test\r\nConnection: close\r\n\r\n"); err != nil { + t.Fatal(err) + } + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("hold handler did not start") + } + waitForStats(func(stats native.Stats) bool { return stats.ActiveRequests == 1 }) + close(release) + response, err := http.ReadResponse(bufio.NewReader(conn), &http.Request{Method: http.MethodGet}) + if err != nil { + t.Fatal(err) + } + response.Body.Close() + if response.StatusCode != http.StatusOK { + t.Fatalf("unexpected hold response: %d", response.StatusCode) + } + waitForStats(func(stats native.Stats) bool { return stats.ActiveRequests == 0 }) + waitForStats(func(stats native.Stats) bool { return stats.ActiveConnections == 0 }) +} + func TestServerConnectionChurnReusesStateSafely(t *testing.T) { application := omni.NewWithRouter(netrouter.New()) application.GET("/health", func(ctx *omni.Context) error { return ctx.OK("ok") })