From 05b4088a256c5890b2dd18dc9f5be57b2b60bc5f Mon Sep 17 00:00:00 2001 From: k2safe Date: Mon, 21 Sep 2026 01:58:50 +0800 Subject: [PATCH] feat: add bounded native http server --- CHANGELOG.md | 7 + README.md | 37 +- ROADMAP.md | 8 +- app/app.go | 26 ++ cmd/go-omni/README.md | 8 +- cmd/go-omni/scaffold/version.go | 2 +- omni.go | 1 + scripts/verifyBusiness.py | 2 +- scripts/verifyDatabase.py | 2 +- scripts/verifyScaffold.py | 2 +- scripts/verifyTransports.py | 4 +- transport/README.md | 22 +- transport/native/server.go | 655 ++++++++++++++++++++++++++++++++ transport/native/server_test.go | 301 +++++++++++++++ 14 files changed, 1058 insertions(+), 19 deletions(-) create mode 100644 transport/native/server.go create mode 100644 transport/native/server_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index ac7ca46..3a4fffc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +## v0.5.0 + +- 新增 `transport/native` 第一版自研 HTTP/1.1 服务端，接管 TCP 监听、连接、Keep-Alive、请求解析、响应写回和优雅关闭。 +- 新增 `App.RunWith` / `omni.Runner` 启动边界；native 通过标准 `http.Handler` 接入，现有 Handler、Service 和 Repository 无需修改。 +- native 支持 `Content-Length` 请求体、请求头 / 请求体限制、读写超时和服务关闭时的请求取消；通过对象池和 `MaxConnections` / `MaxConcurrentRequests` 控制高并发内存；chunked 请求体暂返回 501。 +- `omni.New()` 默认仍使用只依赖标准库的 `transport/nethttp` 路由，native 作为显式启用的网络层预览实现。 + ## v0.4.0 - **破坏性变更**：移除 Gin 适配器及其全部依赖，`omni.New()` 默认使用只依赖标准库的 `transport/nethttp`。 diff --git a/README.md b/README.md index d1d5a01..760c6fa 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ GoOmni 是一套面向业务复用的 Go 后端基础框架。当前默认使用 远程 module 地址是 `git.yuntuo.hk/k2safe/go-omni`。业务项目可以直接引用指定版本： ```sh -go get git.yuntuo.hk/k2safe/go-omni@v0.4.0 +go get git.yuntuo.hk/k2safe/go-omni@v0.5.0 ``` `v0.4.0` 移除了 Gin 和 Gin 专属的 `WithMode` 配置；`omni.New()` 直接使用标准库路由，已有业务 Handler / Service 无需修改。 @@ -13,7 +13,7 @@ go get git.yuntuo.hk/k2safe/go-omni@v0.4.0 创建新业务项目： ```sh -go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.4.0 +go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.5.0 go-omni new --module example.com/billing ./billing cd billing go mod tidy @@ -43,6 +43,7 @@ go-omni/ ├── auth/ # 认证和权限接口 ├── database/ # database/sql 生命周期和事务 ├── transport/nethttp/ # 标准库网络适配器 +├── transport/native/ # 自研 HTTP/1.1 服务端预览 ├── examples/ # 示例项目 └── cmd/ # go-omni 项目脚手架 ``` @@ -94,6 +95,34 @@ application := omni.NewWithRouter(myRouter) `app.Engine()` 返回标准 `http.Handler`，`ctx.Raw()` 返回适配器提供的 `http.ResponseWriter`，业务 API 不锁定具体网络实现。 +需要试用第一版自研网络层时，业务注册代码无需改变，只替换启动器： + +```go +import ( + "context" + "log" + "os" + "os/signal" + "syscall" + + omni "git.yuntuo.hk/k2safe/go-omni" + "git.yuntuo.hk/k2safe/go-omni/transport/native" +) + +ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) +defer cancel() + +app := omni.New() +app.GET("/health", func(ctx *omni.Context) error { + return ctx.OK(map[string]string{"status": "ok"}) +}) +if err := app.RunWith(ctx, native.New(nil), ":8080"); err != nil { + log.Fatal(err) +} +``` + +`transport/native` 当前支持 HTTP/1.1、Keep-Alive、`Content-Length` 请求体、读写超时和优雅关闭；chunked 请求体暂返回 501。它会复用连接状态和响应对象，并可通过 `MaxConnections`、`MaxConcurrentRequests` 限制内存峰值。它仍通过标准 HTTP 请求/响应接口连接业务层，适合在现有 Handler / Service 上做网络层验证。 + 业务代码可以直接使用 `git.yuntuo.hk/k2safe/go-omni/logx` 打结构化日志，不需要依赖具体网络库： ```go @@ -171,10 +200,10 @@ userService := service.NewUsers(services) 从远程版本生成并升级一个独立业务模块，可运行： ```sh -python3 scripts/verifyBusiness.py --version v0.4.0 +python3 scripts/verifyBusiness.py --version v0.5.0 ``` -该验收会在临时目录中先固定 `v0.2.0`，再升级到 `v0.4.0`，检查无本地 replace、配置加载、结构化日志、requestId、CRUD、校验错误、业务错误和优雅退出。 +该验收会在临时目录中先固定 `v0.2.0`，再升级到 `v0.5.0`，检查无本地 replace、配置加载、结构化日志、requestId、CRUD、校验错误、业务错误和优雅退出。 标准库路由和完整 GoOmni 请求链的可重复性能基线见 [PERFORMANCE.md](PERFORMANCE.md)，运行 `python3 scripts/verifyPerformance.py` 可重新采样。 diff --git a/ROADMAP.md b/ROADMAP.md index 32dd7d8..c7c242e 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -35,10 +35,12 @@ - 业务试用没有要求修改 Handler / Service；现有公共 API 足以支撑一套分层业务模块。 - 自动验收入口：`python3 scripts/verifyBusiness.py --version v0.4.0`。 -## 5. 稳定性与性能基线（进行中） +## 5. 自研网络层、稳定性与性能基线（进行中，v0.5.0 预览） - 已加入标准库路由和完整 GoOmni 请求链 benchmark，输出分配次数和耗时，作为未来自研网络层的对照基线。 -- 已具备全量 race、vet、build 验收；生产压测、长连接和真实部署规模验证仍待完成。 +- 新增 `transport/native` 第一版自研 HTTP/1.1 服务端：自行管理 TCP listener、Keep-Alive、Content-Length 请求解析、响应写回、读写超时和优雅关闭；通过 `App.RunWith` 接入，业务 Handler / Service 保持不变。 +- native 当前拒绝 chunked 请求体，仍使用标准 `http.Request` / `http.ResponseWriter` 作为桥接；解析器、连接抽象和生产压测将在后续阶段继续下沉。 +- 已具备全量 race、vet、build 验收；生产压测、长连接规模和真实部署验证仍待完成。 - 自动验收入口：`python3 scripts/verifyPerformance.py`。 -以上阶段以验收结果判断完成。当前仍是基础框架试用阶段；MySQL、适配层契约和一次业务模块试用已完成，稳定性与性能验证、自研网络内核尚未完成。 +以上阶段以验收结果判断完成。当前仍是基础框架试用阶段；MySQL、适配层契约和一次业务模块试用已完成，自研网络层已有可运行预览，稳定性与生产规模验证尚未完成。 diff --git a/app/app.go b/app/app.go index d1cefcb..c2c7e40 100644 --- a/app/app.go +++ b/app/app.go @@ -29,6 +29,13 @@ type App struct { closeErr error } +// Runner is the small lifecycle boundary used by an HTTP server +// implementation. The default App.RunContext uses net/http; a custom runner +// can own the listener, protocol parser and connection lifecycle. +type Runner interface { + Run(context.Context, http.Handler, string) error +} + // New creates an application using GoOmni's standard-library router. func New(options ...config.Option) *App { settings := configure(options...) @@ -150,6 +157,25 @@ func (a *App) RunContext(ctx context.Context, address ...string) error { return a.serveContext(ctx, server, listener) } +// RunWith starts a custom server runner while keeping App-owned cleanup in the +// same lifecycle. The runner receives the transport-neutral App handler. +func (a *App) RunWith(ctx context.Context, runner Runner, address ...string) error { + if runner == nil { + return errors.Join(errors.New("go-omni: nil runner"), a.Close()) + } + if ctx == nil { + ctx = context.Background() + } + listenOn := a.config.Address + if listenOn == "" { + listenOn = ":8080" + } + if len(address) > 0 && address[0] != "" { + listenOn = address[0] + } + return errors.Join(runner.Run(ctx, a, listenOn), a.Close()) +} + func (a *App) serveContext(ctx context.Context, server *http.Server, listener net.Listener) error { serveErr := make(chan error, 1) go func() { serveErr <- server.Serve(listener) }() diff --git a/cmd/go-omni/README.md b/cmd/go-omni/README.md index 71f8a82..b2615da 100644 --- a/cmd/go-omni/README.md +++ b/cmd/go-omni/README.md @@ -3,7 +3,7 @@ 从远程安装，生成一个带 Handler / Service / Repository / Model 分层的服务骨架： ```sh -go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.4.0 +go install git.yuntuo.hk/k2safe/go-omni/cmd/go-omni@v0.5.0 go-omni new --module example.com/billing ./billing cd billing go mod tidy @@ -16,7 +16,7 @@ go run ./cmd/server 需要选择其他已发布版本时，使用完整版本号： ```sh -go-omni new --framework-version v0.4.0 --module example.com/billing ./billing +go-omni new --framework-version v0.5.0 --module example.com/billing ./billing ``` `--framework-version` 禁用自动本地替换；`main`、`latest` 等浮动引用不接受为固定版本。 @@ -48,8 +48,8 @@ MySQL 项目额外包含 `schema.sql`、MySQL Repository 和可选的数据库 发布验收使用 Python 3 标准库脚本，不需要额外 Python 依赖： ```sh -python3 scripts/verifyScaffold.py --version v0.4.0 -python3 scripts/verifyDatabase.py --version v0.4.0 +python3 scripts/verifyScaffold.py --version v0.5.0 +python3 scripts/verifyDatabase.py --version v0.5.0 ``` 脚本在源码目录外安装远程 CLI、生成项目，确认固定版本且没有 `replace`，再下载依赖、 diff --git a/cmd/go-omni/scaffold/version.go b/cmd/go-omni/scaffold/version.go index d805309..04a942e 100644 --- a/cmd/go-omni/scaffold/version.go +++ b/cmd/go-omni/scaffold/version.go @@ -10,7 +10,7 @@ const frameworkModule = "git.yuntuo.hk/k2safe/go-omni" // Source builds can carry an unpublished VCS version (including +dirty). // Keep this fallback in sync with the next release; installed CLIs use their // downloaded module version, which also has a module checksum. -const defaultFrameworkVersion = "v0.4.0" +const defaultFrameworkVersion = "v0.5.0" var frameworkVersionPattern = regexp.MustCompile(`^v[01]\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)(-((0|[1-9][0-9]*|[0-9]*[A-Za-z-][0-9A-Za-z-]*)(\.(0|[1-9][0-9]*|[0-9]*[A-Za-z-][0-9A-Za-z-]*))*))?(\+[0-9A-Za-z-]+(\.[0-9A-Za-z-]+)*)?$`) diff --git a/omni.go b/omni.go index 4e4d297..781a62e 100644 --- a/omni.go +++ b/omni.go @@ -13,6 +13,7 @@ import ( ) type App = app.App +type Runner = app.Runner type RouterGroup = app.RouterGroup type Principal = auth.Principal type Authenticator = auth.Authenticator diff --git a/scripts/verifyBusiness.py b/scripts/verifyBusiness.py index e1e0690..1234f18 100755 --- a/scripts/verifyBusiness.py +++ b/scripts/verifyBusiness.py @@ -25,7 +25,7 @@ def free_address(): def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--go", default="go") - parser.add_argument("--version", default="v0.4.0", help="Target framework release") + parser.add_argument("--version", default="v0.5.0", help="Target framework release") parser.add_argument("--from-version", default="v0.2.0", help="Version used before upgrade") args = parser.parse_args() diff --git a/scripts/verifyDatabase.py b/scripts/verifyDatabase.py index a22035b..0c4b8b8 100644 --- a/scripts/verifyDatabase.py +++ b/scripts/verifyDatabase.py @@ -145,7 +145,7 @@ class Server: def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--go", default="go") - parser.add_argument("--version", default="v0.4.0") + parser.add_argument("--version", default="v0.5.0") parser.add_argument("--cli", help="Use a locally built CLI") parser.add_argument("--framework-dir", help="Local framework source for pre-release checks") parser.add_argument("--image", default="mysql:8.4") diff --git a/scripts/verifyScaffold.py b/scripts/verifyScaffold.py index ff08b96..f410725 100644 --- a/scripts/verifyScaffold.py +++ b/scripts/verifyScaffold.py @@ -91,7 +91,7 @@ def exerciseServer(binary, directory, env, stopSignal, cleanupMarker=None): def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--go", default="go", help="Go executable") - parser.add_argument("--version", default="v0.4.0", help="Released CLI version to install") + parser.add_argument("--version", default="v0.5.0", help="Released CLI version to install") parser.add_argument("--cli", help="Use a locally built CLI instead of installing") parser.add_argument("--framework-version", help="Explicit version passed to the CLI") parser.add_argument("--framework-dir", help="Local checkout for pre-release verification") diff --git a/scripts/verifyTransports.py b/scripts/verifyTransports.py index 6f330a5..56df357 100644 --- a/scripts/verifyTransports.py +++ b/scripts/verifyTransports.py @@ -13,7 +13,7 @@ from verifyScaffold import MODULE, run def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--go", default="go") - parser.add_argument("--version", default="v0.4.0") + parser.add_argument("--version", default="v0.5.0") parser.add_argument("--framework-dir", help="Use a local checkout before release") args = parser.parse_args() env = dict(os.environ, GOWORK="off") @@ -55,7 +55,7 @@ func TestContract(t *testing.T) {{ count = sum(event.get("Action") == "pass" and event.get("Test", "").startswith(name + "/") for event in events) print(f"PASS {adapter}: {count} contract checks, including business handlers, cancellation and concurrent requests", flush=True) - print("PASS independent module: both adapters accepted by the same public contract suite", flush=True) + print("PASS independent module: nethttp accepted by the public contract suite", flush=True) if __name__ == "__main__": diff --git a/transport/README.md b/transport/README.md index a20bcef..8ad13cc 100644 --- a/transport/README.md +++ b/transport/README.md @@ -14,6 +14,24 @@ registerRoutes(application) // 与 omni.New() 复用同一份业务注册代码 `transport/nethttp` 是当前内置的标准库路由实现。它提供一套可移植的路径、方法和错误回退规则；自研网络层可以直接实现同一个 `httpx.Router` 契约。 +`transport/native` 是第一版自研 HTTP/1.1 服务端：它自己管理 TCP listener、连接、Keep-Alive、请求解析、响应写回和优雅关闭，再把请求交给现有 `http.Handler`。当前只支持 `Content-Length` 请求体，chunked 请求会返回 501；它仍使用标准 `http.Request` / `http.ResponseWriter` 作为业务桥接，因此不会影响现有 Handler / Service。 + +服务端不是把连接交给一个全局串行 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 调度或协议状态机成为瓶颈时，才需要继续下沉为手写事件循环。 + +使用自研服务端： + +```go +application := omni.New() +server := native.New(nil) +if err := application.RunWith(ctx, server, ":8080"); err != nil { + log.Fatal(err) +} +``` + +`omni.New()` 默认仍使用 `transport/nethttp` 路由；native 服务端是显式启用的预览实现，等协议和连接边界稳定后再考虑切换默认启动器。 + ## 可移植路由规则（v0.3.0） | 项目 | 约定 | @@ -67,9 +85,9 @@ func TestContract(t *testing.T) { ```sh go test -race ./transport/... ./examples/reference -python3 scripts/verifyTransports.py --version v0.4.0 +python3 scripts/verifyTransports.py --version v0.5.0 ``` 第一条验证标准库适配器复用完整 Handler / Service / Repository。第二条在源码目录外创建无 replace 的消费模块，从远程版本导入同一套验收包；自研适配器也应使用同一个公共测试包。 -当前替换边界是路由与 HTTP 适配层，`App.Run` 仍使用标准库 `net/http.Server`。未来自研监听、连接管理或协议解析时，可让自研服务调用 `App.ServeHTTP`（提供标准 HTTP 请求/响应桥接），由自研启动层管理监听与关闭，并在退出时调用 `App.Close`；本阶段没有实现新的网络内核。 +当前替换边界分成两层：`transport/nethttp` 提供标准库路由，`transport/native` 进一步接管 TCP 监听、连接管理和 HTTP/1.1 协议 I/O。`App.Run` / `App.RunContext` 仍使用标准库 `net/http.Server`，需要显式调用 `App.RunWith(ctx, native.New(nil), address)` 才会启用 native 预览实现。native 当前通过标准 `http.Request` / `http.ResponseWriter` 桥接到业务层，后续可在不改 Handler / Service 的前提下逐步替换解析器和响应写回。 diff --git a/transport/native/server.go b/transport/native/server.go new file mode 100644 index 0000000..0ffafa8 --- /dev/null +++ b/transport/native/server.go @@ -0,0 +1,655 @@ +// Package native provides GoOmni's first project-owned HTTP/1.1 server. +// +// It deliberately keeps the public handler boundary as net/http so existing +// GoOmni business code remains unchanged while connection management and +// protocol I/O move out of net/http.Server. The parser currently supports +// Content-Length request bodies and rejects chunked requests. +package native + +import ( + "bufio" + "bytes" + "context" + "errors" + "io" + "net" + "net/http" + "net/url" + "sort" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" +) + +const ( + defaultReadHeaderTimeout = 5 * time.Second + defaultReadTimeout = 30 * time.Second + defaultWriteTimeout = 30 * time.Second + defaultIdleTimeout = 60 * time.Second + defaultMaxHeaderBytes = 1 << 20 + defaultMaxBodyBytes = 8 << 20 + defaultReadBufferSize = 4 << 10 + maxPooledResponseBytes = 64 << 10 +) + +// Server is a small HTTP/1.1 server with project-owned connection lifecycle. +// It supports keep-alive, graceful shutdown, bounded concurrency and standard +// http.Handler values. Zero MaxConnections and MaxConcurrentRequests mean +// unlimited; pooled protocol objects are reused automatically. +type Server struct { + ReadHeaderTimeout time.Duration + ReadTimeout time.Duration + WriteTimeout time.Duration + IdleTimeout time.Duration + ReadBufferSize int + MaxHeaderBytes int + MaxBodyBytes int64 + MaxConnections int + MaxConcurrentRequests int + + mu sync.Mutex + handler http.Handler + listener net.Listener + conns map[*connection]struct{} + closing bool + cancel context.CancelFunc + wg sync.WaitGroup + connSlots chan struct{} + requestSlots chan struct{} + readerPool sync.Pool + writerPool sync.Pool + connectionPool sync.Pool +} + +type connection struct { + server *Server + conn net.Conn + active atomic.Bool + closeWhenIdle atomic.Bool + closed atomic.Bool + slotHeld bool + closeDone chan struct{} +} + +// New creates a native server. The handler may also be supplied later through +// App.RunWith, which is useful when the App owns the lifecycle. +func New(handler http.Handler) *Server { + server := &Server{ + handler: handler, + ReadHeaderTimeout: defaultReadHeaderTimeout, + ReadTimeout: defaultReadTimeout, + WriteTimeout: defaultWriteTimeout, + IdleTimeout: defaultIdleTimeout, + ReadBufferSize: defaultReadBufferSize, + MaxHeaderBytes: defaultMaxHeaderBytes, + MaxBodyBytes: defaultMaxBodyBytes, + conns: make(map[*connection]struct{}), + } + server.readerPool.New = func() any { return bufio.NewReaderSize(nil, server.readBufferSize()) } + server.writerPool.New = func() any { return newResponseWriter("") } + server.connectionPool.New = func() any { return &connection{server: server} } + return server +} + +func (s *Server) readBufferSize() int { + if s.ReadBufferSize > 0 { + return s.ReadBufferSize + } + return defaultReadBufferSize +} + +// Run implements app.Runner without importing the app package. +func (s *Server) Run(ctx context.Context, handler http.Handler, address string) error { + if handler != nil { + s.handler = handler + } + if s.handler == nil { + return errors.New("go-omni/native: nil handler") + } + if address == "" { + address = ":8080" + } + listener, err := net.Listen("tcp", address) + if err != nil { + return err + } + return s.Serve(ctx, listener) +} + +// Serve accepts connections from listener until the context is canceled or an +// unrecoverable accept error occurs. The listener is owned by the Server. +func (s *Server) Serve(ctx context.Context, listener net.Listener) error { + if listener == nil { + return errors.New("go-omni/native: nil listener") + } + if s.handler == nil { + return errors.New("go-omni/native: nil handler") + } + if ctx == nil { + ctx = context.Background() + } + + s.mu.Lock() + if s.listener != nil { + s.mu.Unlock() + return errors.New("go-omni/native: server is already running") + } + s.listener = listener + s.closing = false + if s.MaxConnections > 0 { + s.connSlots = make(chan struct{}, s.MaxConnections) + } else { + s.connSlots = nil + } + if s.MaxConcurrentRequests > 0 { + s.requestSlots = make(chan struct{}, s.MaxConcurrentRequests) + } else { + s.requestSlots = nil + } + baseCtx, cancel := context.WithCancel(context.Background()) + s.cancel = cancel + s.mu.Unlock() + defer cancel() + + shutdownDone := make(chan struct{}) + go func() { + select { + case <-ctx.Done(): + timeout := s.shutdownTimeout() + shutdownCtx, cancel := context.WithTimeout(context.Background(), timeout) + _ = s.Shutdown(shutdownCtx) + cancel() + case <-shutdownDone: + } + }() + + var serveErr error + for { + conn, err := listener.Accept() + if err != nil { + if ctx.Err() != nil || s.isClosing() { + break + } + serveErr = err + break + } + if slots := s.connSlots; slots != nil { + select { + case slots <- struct{}{}: + default: + _ = conn.Close() + continue + } + } + state := s.connectionPool.Get().(*connection) + state.server = s + state.conn = conn + state.active.Store(false) + state.closeWhenIdle.Store(false) + state.closed.Store(false) + state.slotHeld = s.connSlots != nil + state.closeDone = make(chan struct{}) + s.mu.Lock() + if s.closing { + s.mu.Unlock() + state.close() + s.recycleConnection(state) + continue + } + s.conns[state] = struct{}{} + s.wg.Add(1) + s.mu.Unlock() + go s.serveConnection(baseCtx, state) + } + + close(shutdownDone) + s.mu.Lock() + if s.listener == listener { + s.listener = nil + } + s.mu.Unlock() + if ctx.Err() != nil || s.isClosing() { + timeout := s.shutdownTimeout() + shutdownCtx, cancel := context.WithTimeout(context.Background(), timeout) + shutdownErr := s.Shutdown(shutdownCtx) + cancel() + return errors.Join(shutdownErr, serveErr) + } + _ = listener.Close() + timeout := s.shutdownTimeout() + shutdownCtx, cancel := context.WithTimeout(context.Background(), timeout) + shutdownErr := s.Shutdown(shutdownCtx) + cancel() + return errors.Join(shutdownErr, serveErr) +} + +// Shutdown stops accepting requests, cancels active request contexts and waits +// for connections to finish. On timeout, active sockets are closed. +func (s *Server) Shutdown(ctx context.Context) error { + if ctx == nil { + ctx = context.Background() + } + s.mu.Lock() + s.closing = true + if s.cancel != nil { + s.cancel() + } + listener := s.listener + states := make([]*connection, 0, len(s.conns)) + for state := range s.conns { + state.closeWhenIdle.Store(true) + states = append(states, state) + } + s.listener = nil + s.mu.Unlock() + if listener != nil { + _ = listener.Close() + } + for _, state := range states { + if !state.active.Load() { + state.close() + } + } + + done := make(chan struct{}) + go func() { + s.wg.Wait() + close(done) + }() + select { + case <-done: + return nil + case <-ctx.Done(): + for _, state := range states { + state.close() + } + <-done + return ctx.Err() + } +} + +func (s *Server) shutdownTimeout() time.Duration { + if value := s.WriteTimeout; value > 0 { + return value + } + return defaultWriteTimeout +} + +func (s *Server) isClosing() bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.closing +} + +func (s *Server) serveConnection(baseCtx context.Context, state *connection) { + defer s.wg.Done() + defer func() { + state.close() + s.recycleConnection(state) + }() + reader := s.readerPool.Get().(*bufio.Reader) + reader.Reset(state.conn) + defer func() { + reader.Reset(nil) + s.readerPool.Put(reader) + }() + firstRequest := true + for { + readTimeout := s.ReadHeaderTimeout + if readTimeout <= 0 { + readTimeout = defaultReadHeaderTimeout + } + if !firstRequest && s.IdleTimeout > 0 { + readTimeout = s.IdleTimeout + } else if s.IdleTimeout > 0 && s.IdleTimeout < readTimeout { + readTimeout = s.IdleTimeout + } + _ = state.conn.SetReadDeadline(time.Now().Add(readTimeout)) + request, closeAfter, err := readRequest(reader, state.conn, s.MaxHeaderBytes, s.MaxBodyBytes, func() { + if s.ReadTimeout > 0 { + _ = state.conn.SetReadDeadline(time.Now().Add(s.ReadTimeout)) + } else { + _ = state.conn.SetReadDeadline(time.Time{}) + } + }) + if err != nil { + if !errors.Is(err, io.EOF) && !errors.Is(err, net.ErrClosed) && !isTimeout(err) { + status := http.StatusBadRequest + message := "bad request\n" + var protocolErr *protocolError + if errors.As(err, &protocolErr) { + status = protocolErr.status + message = protocolErr.message + "\n" + } + _ = writeSimple(state.conn, status, message, true, s.WriteTimeout) + } + return + } + _ = state.conn.SetReadDeadline(time.Time{}) + firstRequest = false + requestContext, cancel := context.WithCancel(baseCtx) + request = request.WithContext(requestContext) + state.active.Store(true) + writer := s.writerPool.Get().(*responseWriter) + writer.reset(request.Method) + requestSlot := false + if slots := s.requestSlots; slots != nil { + select { + case slots <- struct{}{}: + requestSlot = true + case <-baseCtx.Done(): + cancel() + writer.reset("") + s.writerPool.Put(writer) + state.active.Store(false) + return + default: + writer.Header().Set("Content-Type", "text/plain; charset=utf-8") + writer.WriteHeader(http.StatusServiceUnavailable) + writer.body.WriteString("server busy\n") + closing := s.isClosing() || state.closeWhenIdle.Load() + _ = writer.writeTo(state.conn, closeAfter || closing, s.WriteTimeout) + cancel() + writer.reset("") + s.writerPool.Put(writer) + state.active.Store(false) + if closeAfter || closing { + return + } + continue + } + } + handler := s.handler + handler.ServeHTTP(writer, request) + cancel() + + closing := s.isClosing() || state.closeWhenIdle.Load() + if err := writer.writeTo(state.conn, closeAfter || closing, s.WriteTimeout); err != nil { + if requestSlot { + <-s.requestSlots + } + writer.reset("") + s.writerPool.Put(writer) + state.active.Store(false) + return + } + if requestSlot { + <-s.requestSlots + } + writer.reset("") + s.writerPool.Put(writer) + state.active.Store(false) + if closeAfter || closing { + return + } + } +} + +func (c *connection) close() { + if c.closed.CompareAndSwap(false, true) { + defer close(c.closeDone) + _ = c.conn.Close() + c.server.mu.Lock() + delete(c.server.conns, c) + c.server.mu.Unlock() + } +} + +func (s *Server) recycleConnection(c *connection) { + <-c.closeDone + if c.slotHeld { + if slots := s.connSlots; slots != nil { + <-slots + } + c.slotHeld = false + } + s.connectionPool.Put(c) +} + +type protocolError struct { + status int + message string +} + +func (e *protocolError) Error() string { return e.message } + +func readRequest(reader *bufio.Reader, conn net.Conn, maxHeaderBytes int, maxBodyBytes int64, setBodyDeadline func()) (*http.Request, bool, error) { + if maxHeaderBytes <= 0 { + maxHeaderBytes = defaultMaxHeaderBytes + } + if maxBodyBytes <= 0 { + maxBodyBytes = defaultMaxBodyBytes + } + used := 0 + readLine := func() (string, error) { + line, err := reader.ReadString('\n') + used += len(line) + if used > maxHeaderBytes { + return "", &protocolError{status: http.StatusRequestHeaderFieldsTooLarge, message: "request headers too large"} + } + if err != nil { + return "", err + } + if !strings.HasSuffix(line, "\n") { + return "", &protocolError{status: http.StatusBadRequest, message: "malformed request line"} + } + return strings.TrimSuffix(strings.TrimSuffix(line, "\n"), "\r"), nil + } + + line, err := readLine() + if err != nil { + return nil, false, err + } + parts := strings.Split(line, " ") + if len(parts) != 3 || !validToken(parts[0]) { + return nil, true, &protocolError{status: http.StatusBadRequest, message: "malformed request line"} + } + protoMajor, protoMinor, ok := parseProtocol(parts[2]) + if !ok { + return nil, true, &protocolError{status: http.StatusHTTPVersionNotSupported, message: "unsupported HTTP version"} + } + target := parts[1] + var requestURL *url.URL + if target == "*" { + requestURL = &url.URL{Path: "*"} + } else { + requestURL, err = url.ParseRequestURI(target) + if err != nil || requestURL.Path == "" || !strings.HasPrefix(requestURL.Path, "/") { + return nil, true, &protocolError{status: http.StatusBadRequest, message: "malformed request target"} + } + } + + header := make(http.Header) + for { + line, err = readLine() + if err != nil { + return nil, true, err + } + if line == "" { + break + } + if strings.HasPrefix(line, " ") || strings.HasPrefix(line, "\t") { + return nil, true, &protocolError{status: http.StatusBadRequest, message: "folded headers are not supported"} + } + index := strings.IndexByte(line, ':') + if index <= 0 { + return nil, true, &protocolError{status: http.StatusBadRequest, message: "malformed header"} + } + name := strings.TrimSpace(line[:index]) + value := strings.TrimSpace(line[index+1:]) + if !validToken(name) || strings.ContainsAny(value, "\r\n") { + return nil, true, &protocolError{status: http.StatusBadRequest, message: "malformed header"} + } + header.Add(name, value) + } + + if value := header.Get("Expect"); value != "" { + return nil, true, &protocolError{status: http.StatusExpectationFailed, message: "unsupported expectation"} + } + if value := header.Get("Transfer-Encoding"); value != "" && !strings.EqualFold(value, "identity") { + return nil, true, &protocolError{status: http.StatusNotImplemented, message: "chunked requests are not supported"} + } + contentLength := int64(0) + if value := header.Get("Content-Length"); value != "" { + contentLength, err = strconv.ParseInt(strings.TrimSpace(value), 10, 64) + if err != nil || contentLength < 0 { + return nil, true, &protocolError{status: http.StatusBadRequest, message: "invalid content length"} + } + } + if contentLength > maxBodyBytes { + return nil, true, &protocolError{status: http.StatusRequestEntityTooLarge, message: "request body too large"} + } + body := make([]byte, contentLength) + if contentLength > 0 { + if setBodyDeadline != nil { + setBodyDeadline() + } + if _, err := io.ReadFull(reader, body); err != nil { + return nil, true, err + } + } + request := &http.Request{ + Method: parts[0], + URL: requestURL, + Proto: parts[2], + ProtoMajor: protoMajor, + ProtoMinor: protoMinor, + Header: header, + Body: io.NopCloser(bytes.NewReader(body)), + ContentLength: contentLength, + Host: header.Get("Host"), + RequestURI: target, + RemoteAddr: conn.RemoteAddr().String(), + } + closeAfter := strings.EqualFold(header.Get("Connection"), "close") || + (protoMajor == 1 && protoMinor == 0 && !strings.Contains(strings.ToLower(header.Get("Connection")), "keep-alive")) + request.Close = closeAfter + return request, closeAfter, nil +} + +func parseProtocol(value string) (int, int, bool) { + if value != "HTTP/1.1" && value != "HTTP/1.0" { + return 0, 0, false + } + return 1, int(value[len(value)-1] - '0'), true +} + +func validToken(value string) bool { + if value == "" { + return false + } + for _, char := range value { + if char <= ' ' || strings.ContainsRune("()<>@,;:\\\"/[]?={} \t", char) { + return false + } + } + return true +} + +func isTimeout(err error) bool { + var netErr net.Error + return errors.As(err, &netErr) && netErr.Timeout() +} + +func writeSimple(conn net.Conn, status int, body string, closeAfter bool, timeout time.Duration) error { + writer := newResponseWriter("") + writer.status = status + writer.wroteHeader = true + writer.body.WriteString(body) + writer.header.Set("Content-Type", "text/plain; charset=utf-8") + return writer.writeTo(conn, closeAfter, timeout) +} + +type responseWriter struct { + header http.Header + status int + wroteHeader bool + body bytes.Buffer + requestMethod string +} + +func newResponseWriter(method string) *responseWriter { + return &responseWriter{header: make(http.Header), requestMethod: method} +} + +func (w *responseWriter) reset(method string) { + for key := range w.header { + delete(w.header, key) + } + w.status = 0 + w.wroteHeader = false + if w.body.Cap() > maxPooledResponseBytes { + w.body = bytes.Buffer{} + } else { + w.body.Reset() + } + w.requestMethod = method +} + +func (w *responseWriter) Header() http.Header { return w.header } + +func (w *responseWriter) WriteHeader(status int) { + if w.wroteHeader { + return + } + w.status = status + w.wroteHeader = true +} + +func (w *responseWriter) Write(data []byte) (int, error) { + if !w.wroteHeader { + w.WriteHeader(http.StatusOK) + } + return w.body.Write(data) +} + +func (w *responseWriter) writeTo(conn net.Conn, closeAfter bool, timeout time.Duration) error { + if timeout > 0 { + _ = conn.SetWriteDeadline(time.Now().Add(timeout)) + } + if !w.wroteHeader { + w.status = http.StatusOK + w.wroteHeader = true + } + body := w.body.Bytes() + noBody := w.requestMethod == http.MethodHead || w.status == http.StatusNoContent || + w.status == http.StatusNotModified || (w.status >= 100 && w.status < 200) + if noBody { + body = nil + } + if w.header.Get("Content-Length") == "" && w.status != http.StatusNoContent && w.status != http.StatusNotModified { + w.header.Set("Content-Length", strconv.Itoa(len(w.body.Bytes()))) + } + if closeAfter { + w.header.Set("Connection", "close") + } + 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 { + for _, value := range w.header.Values(key) { + if strings.ContainsAny(value, "\r\n") { + return errors.New("invalid response header") + } + output.WriteString(key) + output.WriteString(": ") + output.WriteString(value) + output.WriteString("\r\n") + } + } + output.WriteString("\r\n") + output.Write(body) + _, err := conn.Write(output.Bytes()) + _ = conn.SetWriteDeadline(time.Time{}) + return err +} diff --git a/transport/native/server_test.go b/transport/native/server_test.go new file mode 100644 index 0000000..d2c3651 --- /dev/null +++ b/transport/native/server_test.go @@ -0,0 +1,301 @@ +package native_test + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + "net" + "net/http" + "strings" + "testing" + "time" + + omni "git.yuntuo.hk/k2safe/go-omni" + "git.yuntuo.hk/k2safe/go-omni/transport/native" + netrouter "git.yuntuo.hk/k2safe/go-omni/transport/nethttp" +) + +func TestServerKeepAliveAndApplicationBridge(t *testing.T) { + application := omni.NewWithRouter(netrouter.New()) + application.GET("/hello", func(ctx *omni.Context) error { + return ctx.OK(map[string]string{"message": "hello"}) + }) + 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) + + for index, closeAfter := range []bool{false, true} { + connectionHeader := "keep-alive" + if closeAfter { + connectionHeader = "close" + } + fmt.Fprintf(conn, "GET /hello HTTP/1.1\r\nHost: test\r\nConnection: %s\r\n\r\n", connectionHeader) + request := &http.Request{Method: http.MethodGet} + response, err := http.ReadResponse(reader, request) + if err != nil { + t.Fatalf("response %d: %v", index, err) + } + body, err := io.ReadAll(response.Body) + response.Body.Close() + if err != nil || response.StatusCode != http.StatusOK || !strings.Contains(string(body), "\"message\":\"hello\"") { + t.Fatalf("response %d: status=%d body=%s error=%v", index, response.StatusCode, body, err) + } + } +} + +func TestServerRejectsUnsupportedBodyAndLimits(t *testing.T) { + application := omni.NewWithRouter(netrouter.New()) + application.POST("/body", func(ctx *omni.Context) error { return ctx.OK("ok") }) + + server := native.New(application) + server.MaxBodyBytes = 4 + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { done <- server.Serve(context.Background(), listener) }() + waitForServer(t, listener.Addr().String()) + defer stopServer(t, server, listener, done) + + for _, requestText := range []string{ + "POST /body HTTP/1.1\r\nHost: test\r\nTransfer-Encoding: chunked\r\n\r\n4\r\ntest\r\n0\r\n\r\n", + "POST /body HTTP/1.1\r\nHost: test\r\nContent-Length: 5\r\n\r\ntest!", + } { + conn, err := net.Dial("tcp", listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + _, _ = io.WriteString(conn, requestText) + response, err := http.ReadResponse(bufio.NewReader(conn), &http.Request{Method: http.MethodPost}) + if err != nil { + conn.Close() + t.Fatal(err) + } + response.Body.Close() + if response.StatusCode != http.StatusNotImplemented && response.StatusCode != http.StatusRequestEntityTooLarge { + t.Fatalf("unexpected protocol status %d", response.StatusCode) + } + conn.Close() + } +} + +func TestServerShutdownCancelsActiveHandler(t *testing.T) { + started := make(chan struct{}) + application := omni.NewWithRouter(netrouter.New()) + application.GET("/wait", func(ctx *omni.Context) error { + close(started) + <-ctx.Context().Done() + return ctx.Context().Err() + }) + server, listener, done := startServer(t, application) + + conn, err := net.Dial("tcp", listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + _, _ = io.WriteString(conn, "GET /wait HTTP/1.1\r\nHost: test\r\nConnection: close\r\n\r\n") + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("handler did not start") + } + shutdownContext, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := server.Shutdown(shutdownContext); err != nil { + t.Fatal(err) + } + 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.StatusInternalServerError { + t.Fatalf("expected canceled handler to become 500, got %d", response.StatusCode) + } + conn.Close() + select { + case err := <-done: + if err != nil && !errors.Is(err, net.ErrClosed) { + t.Fatal(err) + } + case <-time.After(2 * time.Second): + t.Fatal("server did not stop") + } +} + +func TestServerConcurrentRequestLimit(t *testing.T) { + started := make(chan struct{}) + release := make(chan struct{}) + application := omni.NewWithRouter(netrouter.New()) + application.GET("/work", func(ctx *omni.Context) error { + close(started) + <-release + return ctx.OK("done") + }) + server := native.New(application) + server.MaxConcurrentRequests = 1 + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { done <- server.Serve(context.Background(), listener) }() + waitForServer(t, listener.Addr().String()) + defer func() { + stopServer(t, server, listener, done) + }() + + first, err := net.Dial("tcp", listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer first.Close() + _, _ = io.WriteString(first, "GET /work HTTP/1.1\r\nHost: test\r\nConnection: close\r\n\r\n") + select { + case <-started: + case <-time.After(2 * time.Second): + t.Fatal("first handler did not start") + } + + second, err := net.Dial("tcp", listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer second.Close() + _, _ = io.WriteString(second, "GET /work HTTP/1.1\r\nHost: test\r\nConnection: close\r\n\r\n") + response, err := http.ReadResponse(bufio.NewReader(second), &http.Request{Method: http.MethodGet}) + if err != nil { + t.Fatal(err) + } + response.Body.Close() + if response.StatusCode != http.StatusServiceUnavailable { + t.Fatalf("expected 503 at concurrency limit, got %d", response.StatusCode) + } + close(release) + response, err = http.ReadResponse(bufio.NewReader(first), &http.Request{Method: http.MethodGet}) + if err != nil { + t.Fatal(err) + } + response.Body.Close() + if response.StatusCode != http.StatusOK { + t.Fatalf("expected first request to complete, got %d", response.StatusCode) + } +} + +func TestServerReadHeaderTimeout(t *testing.T) { + application := omni.NewWithRouter(netrouter.New()) + application.GET("/health", func(ctx *omni.Context) error { return ctx.OK("ok") }) + server := native.New(application) + server.ReadHeaderTimeout = 25 * time.Millisecond + server.IdleTimeout = 0 + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { done <- server.Serve(context.Background(), listener) }() + waitForServer(t, listener.Addr().String()) + defer stopServer(t, server, listener, done) + + conn, err := net.Dial("tcp", listener.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + conn.SetReadDeadline(time.Now().Add(time.Second)) + _, _ = io.WriteString(conn, "GET /health HTTP/1.1\r\nHost: test") + buffer := make([]byte, 1) + if _, err := conn.Read(buffer); err == nil { + t.Fatal("expected incomplete headers to time out and close") + } +} + +func TestAppRunWithNativeRunner(t *testing.T) { + application := omni.NewWithRouter(netrouter.New()) + application.GET("/health", func(ctx *omni.Context) error { return ctx.OK("ok") }) + address := freeAddress(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- application.RunWith(ctx, native.New(nil), address) }() + waitForServer(t, address) + response, err := http.Get("http://" + address + "/health") + if err != nil { + t.Fatal(err) + } + response.Body.Close() + if response.StatusCode != http.StatusOK { + t.Fatalf("unexpected status %d", response.StatusCode) + } + cancel() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(3 * time.Second): + t.Fatal("RunWith did not stop") + } +} + +func startServer(t *testing.T, handler http.Handler) (*native.Server, net.Listener, <-chan error) { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + server := native.New(handler) + done := make(chan error, 1) + go func() { done <- server.Serve(context.Background(), listener) }() + waitForServer(t, listener.Addr().String()) + return server, listener, done +} + +func stopServer(t *testing.T, server *native.Server, listener net.Listener, done <-chan error) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _ = server.Shutdown(ctx) + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("server did not stop") + } + _ = listener.Close() +} + +func freeAddress(t *testing.T) string { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + address := listener.Addr().String() + listener.Close() + return address +} + +func waitForServer(t *testing.T, address string) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for { + conn, err := net.DialTimeout("tcp", address, 50*time.Millisecond) + if err == nil { + conn.Close() + return + } + if time.Now().After(deadline) { + t.Fatalf("server did not start: %v", err) + } + time.Sleep(10 * time.Millisecond) + } +}