mirror of
https://github.com/zap-proto/http.git
synced 2026-10-04 03:07:19 +00:00
feat(client): connection pool
Adds a free-list pool of idle TCP connections with TCP keepalive + NoDelay. RoundTrip pulls a hot conn or dials new; release-on-Close returns it. Pool caps at MaxIdleConns (default 32). Enables HTTP/1.1-keepalive-style request reuse for ZAP-HTTP, closing the sequential-throughput gap with net/http measured in zap-proto/bench.
This commit is contained in:
@@ -10,37 +10,51 @@
|
||||
// to pin the peer) and req.URL.Scheme. Existing http.Client
|
||||
// machinery — redirects, cookies, retries — works unchanged.
|
||||
//
|
||||
// Connection management today is one connection per RoundTrip call.
|
||||
// v0.2 will introduce a connection pool with keep-alive once the
|
||||
// underlying ZAP transport ships handshake amortization.
|
||||
// Connection management uses a free-list pool of idle TCP
|
||||
// connections. RoundTrip pulls an idle conn (or dials a new one),
|
||||
// uses it for one exchange, and returns it to the pool. The pool
|
||||
// caps at MaxIdleConns; surplus conns close on return.
|
||||
|
||||
package zaphttp
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// pooledConn is one idle connection. The bufio.Reader is held with
|
||||
// the conn so the read-side amortizes its own buffer alloc across
|
||||
// requests.
|
||||
type pooledConn struct {
|
||||
c net.Conn
|
||||
br *bufio.Reader
|
||||
}
|
||||
|
||||
// Transport implements http.RoundTripper over a ZAP-HTTP TCP
|
||||
// connection. Field-zero is invalid; use NewTransport.
|
||||
type Transport struct {
|
||||
addr string
|
||||
dialTimeout time.Duration
|
||||
readTimeout time.Duration
|
||||
addr string
|
||||
dialTimeout time.Duration
|
||||
readTimeout time.Duration
|
||||
maxIdleConns int
|
||||
|
||||
mu sync.Mutex
|
||||
mu sync.Mutex
|
||||
idle []*pooledConn // LIFO; most-recently-returned is hottest
|
||||
}
|
||||
|
||||
// NewTransport returns a Transport connecting to the given host:port.
|
||||
func NewTransport(addr string) *Transport {
|
||||
return &Transport{
|
||||
addr: addr,
|
||||
dialTimeout: 10 * time.Second,
|
||||
readTimeout: 30 * time.Second,
|
||||
addr: addr,
|
||||
dialTimeout: 10 * time.Second,
|
||||
readTimeout: 30 * time.Second,
|
||||
maxIdleConns: 32,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -50,56 +64,125 @@ func (t *Transport) SetDialTimeout(d time.Duration) { t.dialTimeout = d }
|
||||
// SetReadTimeout overrides the default 30s response-read timeout.
|
||||
func (t *Transport) SetReadTimeout(d time.Duration) { t.readTimeout = d }
|
||||
|
||||
// SetMaxIdleConns caps the number of idle conns held in the pool.
|
||||
// Surplus conns close on return.
|
||||
func (t *Transport) SetMaxIdleConns(n int) {
|
||||
if n < 0 {
|
||||
n = 0
|
||||
}
|
||||
t.maxIdleConns = n
|
||||
}
|
||||
|
||||
// RoundTrip executes a single request/response exchange.
|
||||
func (t *Transport) RoundTrip(req *http.Request) (*http.Response, error) {
|
||||
if t.addr == "" {
|
||||
return nil, fmt.Errorf("zaphttp: Transport.addr is empty (use NewTransport)")
|
||||
}
|
||||
|
||||
conn, err := net.DialTimeout("tcp", t.addr, t.dialTimeout)
|
||||
pc, err := t.acquireConn()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("zaphttp: dial %s: %w", t.addr, err)
|
||||
}
|
||||
closeOnFail := true
|
||||
defer func() {
|
||||
if closeOnFail {
|
||||
conn.Close()
|
||||
}
|
||||
}()
|
||||
conn, br := pc.c, pc.br
|
||||
|
||||
// Marshal + send request.
|
||||
frame, err := MarshalRequest(req)
|
||||
if err != nil {
|
||||
conn.Close()
|
||||
return nil, err
|
||||
}
|
||||
if err := writeFrame(conn, frame); err != nil {
|
||||
conn.Close()
|
||||
return nil, fmt.Errorf("zaphttp: write request: %w", err)
|
||||
}
|
||||
|
||||
// Read response.
|
||||
if t.readTimeout > 0 {
|
||||
_ = conn.SetReadDeadline(time.Now().Add(t.readTimeout))
|
||||
}
|
||||
br := bufio.NewReader(conn)
|
||||
respFrame, err := readFrame(br)
|
||||
if err != nil {
|
||||
conn.Close()
|
||||
return nil, fmt.Errorf("zaphttp: read response: %w", err)
|
||||
}
|
||||
resp, err := UnmarshalResponse(respFrame)
|
||||
if err != nil {
|
||||
conn.Close()
|
||||
return nil, fmt.Errorf("zaphttp: unmarshal response: %w", err)
|
||||
}
|
||||
|
||||
// Tie the response Body to the connection so it gets closed when
|
||||
// the caller closes the body. The current single-frame design
|
||||
// means the body is fully buffered in memory, so we can close the
|
||||
// connection immediately and leave the buffer in resp.Body. v0.2
|
||||
// will defer connection close until body close.
|
||||
closeOnFail = false
|
||||
conn.Close()
|
||||
// Reset the read deadline before returning the conn to the pool.
|
||||
_ = conn.SetReadDeadline(time.Time{})
|
||||
|
||||
// The body is already buffered in memory by the unmarshal step,
|
||||
// so the conn is reusable as soon as the response is decoded.
|
||||
// Wrap Body so the caller can still call Close() without
|
||||
// affecting the conn — pool ownership is the transport's
|
||||
// responsibility, not the caller's.
|
||||
resp.Body = bodyCloser{ReadCloser: resp.Body, release: func() { t.releaseConn(pc) }}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// bodyCloser wraps the response body so closing it releases the
|
||||
// underlying TCP connection to the pool.
|
||||
type bodyCloser struct {
|
||||
io.ReadCloser
|
||||
release func()
|
||||
once sync.Once
|
||||
}
|
||||
|
||||
func (b bodyCloser) Close() error {
|
||||
err := b.ReadCloser.Close()
|
||||
b.once.Do(b.release)
|
||||
return err
|
||||
}
|
||||
|
||||
// acquireConn pops an idle conn or dials a new one.
|
||||
func (t *Transport) acquireConn() (*pooledConn, error) {
|
||||
t.mu.Lock()
|
||||
if n := len(t.idle); n > 0 {
|
||||
pc := t.idle[n-1]
|
||||
t.idle = t.idle[:n-1]
|
||||
t.mu.Unlock()
|
||||
return pc, nil
|
||||
}
|
||||
t.mu.Unlock()
|
||||
|
||||
c, err := net.DialTimeout("tcp", t.addr, t.dialTimeout)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if tc, ok := c.(*net.TCPConn); ok {
|
||||
_ = tc.SetKeepAlive(true)
|
||||
_ = tc.SetKeepAlivePeriod(30 * time.Second)
|
||||
_ = tc.SetNoDelay(true)
|
||||
}
|
||||
return &pooledConn{c: c, br: bufio.NewReader(c)}, nil
|
||||
}
|
||||
|
||||
// releaseConn returns a conn to the pool, or closes it if the pool
|
||||
// is full.
|
||||
func (t *Transport) releaseConn(pc *pooledConn) {
|
||||
t.mu.Lock()
|
||||
if len(t.idle) >= t.maxIdleConns {
|
||||
t.mu.Unlock()
|
||||
_ = pc.c.Close()
|
||||
return
|
||||
}
|
||||
t.idle = append(t.idle, pc)
|
||||
t.mu.Unlock()
|
||||
}
|
||||
|
||||
// CloseIdleConnections closes every idle conn in the pool. Active
|
||||
// requests are unaffected. Useful in tests + on shutdown.
|
||||
func (t *Transport) CloseIdleConnections() {
|
||||
t.mu.Lock()
|
||||
idle := t.idle
|
||||
t.idle = nil
|
||||
t.mu.Unlock()
|
||||
for _, pc := range idle {
|
||||
_ = pc.c.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// Get is a convenience for transports that don't need an http.Client
|
||||
// wrapper (e.g. service-to-service calls inside a cluster).
|
||||
func Get(addr, path string) (*http.Response, error) {
|
||||
@@ -110,3 +193,14 @@ func Get(addr, path string) (*http.Response, error) {
|
||||
}
|
||||
return t.RoundTrip(req)
|
||||
}
|
||||
|
||||
// Compile-time guard that the bodyCloser remains a valid io.ReadCloser
|
||||
// even if the wrapped Body returns a never-closed type.
|
||||
var _ io.ReadCloser = bodyCloser{}
|
||||
|
||||
// errInvalidConn is returned by acquireConn when the pool yields a
|
||||
// stale connection (currently unused; placeholder for future
|
||||
// liveness checks).
|
||||
var errInvalidConn = errors.New("zaphttp: pooled connection invalid")
|
||||
|
||||
var _ = errInvalidConn
|
||||
|
||||
Reference in New Issue
Block a user