github/gost-x
4
0
Fork 0
mirror of https://github.com/go-gost/x.git synced 2026-10-19 16:16:32 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(pht): eliminate authorize round-trip, fix close notification and race

Browse Source 
perf(pht): eliminate /authorize HTTP round-trip by self-generating CID
  Client now uses xid to generate connection IDs, removing one RTT per
  tunnel setup. Server lazily creates pipe pairs in handlePush/handlePull
  via getOrCreateConn helper, keeping backward compat for old clients
  that still call /authorize.

fix: send close notification before canceling context
  Close() was canceling c.ctx before calling write(nil). With a canceled
  context, http.Client.Do returns immediately without network I/O, so
  the close notification POST never reached the server. Swap the order.

fix: add Flush() after heartbeat write in handlePull
  w.Write() on http.ResponseWriter goes to a 2k bufio.Writer. Without a
  Flush(), the keepalive newline never reaches the wire, causing the
  client long-poll GET to time out.

fix: use LoadOrStore in createPipe to prevent concurrent push/pull race
  When push and pull HTTP requests arrive concurrently for a new CID,
  both saw !ok from Load() and both called createPipe(), overwriting
  the first conns entry. The losing pipe pair leaked and the listener
  accepted a phantom connection. LoadOrStore makes the first Store win;
  the second caller gets the existing pipe.

fix: close conn and delete entry on empty-data POST in handlePush
  A POST body of "\n" produced empty data after TrimSuffix, leaking the
  pipe pair and conns entry. Now always cleans up before returning.

fix: use caller ctx instead of context.Background() in Dial
  Dial(ctx, addr) was discarding the caller context, so deadlines and
  cancellation did not propagate to the HTTP long-poll requests.

fix: detect ListenAndServe failure in Init() and clean up goroutine
  Non-blocking select on errChan after starting the server goroutine
  surfaces immediate failures (port conflict, TLS error) instead of
  leaking a goroutine blocked on server.Accept().

chore: remove dead AuthorizePath from Client struct and dialer wiring
  Client no longer calls /authorize; the field and its metadata parsing
  in dialer/pht, dialer/http3 are dead code.

test: add unit tests for createPipe race, close notification,
  empty-data cleanup, and queue-full fallback
This commit is contained in:
ginuerzh
2026-07-06 22:03:34 +08:00
parent ca767cc102
commit a2e2810669
11 changed files with 348 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dialer/http3/dialer.go
+4 -5
View File
@@ -97,11 +97,10 @@ func (d *http3Dialer) Dial(ctx context.Context, addr string, opts ...dialer.Dial
},
},
},
AuthorizePath: d.md.authorizePath,
PushPath: d.md.pushPath,
PullPath: d.md.pullPath,
TLSEnabled: true,
Logger: d.options.Logger,
PushPath: d.md.pushPath,
PullPath: d.md.pullPath,
TLSEnabled: true,
Logger: d.options.Logger,
}
d.clients[addr] = client
dialer/http3/metadata.go
+5 -11
View File
@@ -9,16 +9,14 @@ import (
)
const (
defaultAuthorizePath = "/authorize"
defaultPushPath = "/push"
defaultPullPath = "/pull"
defaultPushPath = "/push"
defaultPullPath = "/pull"
)
type metadata struct {
authorizePath string
pushPath string
pullPath string
host string
pushPath string
pullPath string
host string
// QUIC config options
keepAlivePeriod time.Duration
@@ -36,10 +34,6 @@ func (d *http3Dialer) parseMetadata(md mdata.Metadata) (err error) {
maxStreams = "maxStreams"
)
d.md.authorizePath = mdutil.GetString(md, "pht.authorizePath", "authorizePath")
if !strings.HasPrefix(d.md.authorizePath, "/") {
d.md.authorizePath = defaultAuthorizePath
}
d.md.pushPath = mdutil.GetString(md, "pht.pushPath", "pushPath")
if !strings.HasPrefix(d.md.pushPath, "/") {
d.md.pushPath = defaultPushPath
dialer/pht/dialer.go
+5 -6
View File
@@ -105,12 +105,11 @@ func (d *phtDialer) Dial(ctx context.Context, addr string, opts ...dialer.DialOp
// Timeout: 60 * time.Second,
Transport: tr,
},
AuthorizePath: d.md.authorizePath,
PushPath: d.md.pushPath,
PullPath: d.md.pullPath,
TLSEnabled: d.tlsEnabled,
Header: d.md.header,
Logger: d.options.Logger,
PushPath: d.md.pushPath,
PullPath: d.md.pullPath,
TLSEnabled: d.tlsEnabled,
Header: d.md.header,
Logger: d.options.Logger,
}
d.clients[addr] = client
}
dialer/pht/metadata.go
+10 -18
View File
@@ -9,33 +9,25 @@ import (
)
const (
dialTimeout = "dialTimeout"
defaultAuthorizePath = "/authorize"
defaultPushPath = "/push"
defaultPullPath = "/pull"
defaultPushPath = "/push"
defaultPullPath = "/pull"
)
type metadata struct {
authorizePath string
pushPath string
pullPath string
host string
header http.Header
pushPath string
pullPath string
host string
header http.Header
}
func (d *phtDialer) parseMetadata(md mdata.Metadata) (err error) {
const (
authorizePath = "authorizePath"
pushPath = "pushPath"
pullPath = "pullPath"
host = "host"
header = "header"
pushPath = "pushPath"
pullPath = "pullPath"
host = "host"
header = "header"
)
d.md.authorizePath = mdutil.GetString(md, authorizePath)
if !strings.HasPrefix(d.md.authorizePath, "/") {
d.md.authorizePath = defaultAuthorizePath
}
d.md.pushPath = mdutil.GetString(md, pushPath)
if !strings.HasPrefix(d.md.pushPath, "/") {
d.md.pushPath = defaultPushPath
go.mod
+1 -1
View File
@@ -13,7 +13,7 @@ require (
github.com/go-gost/gosocks4 v0.1.0
github.com/go-gost/gosocks5 v0.5.0
github.com/go-gost/plugin v0.5.0
github.com/go-gost/relay v0.6.1
github.com/go-gost/relay v0.6.2
github.com/go-gost/tls-dissector v0.2.0
github.com/go-redis/redis/v8 v8.11.5
github.com/gobwas/glob v0.2.3
go.sum
+2
View File
@@ -63,6 +63,8 @@ github.com/go-gost/plugin v0.5.0 h1:zK3Ezcv+z6bz6aUhaFZcZiJc42w44ohybzrj5tnWzcM=
github.com/go-gost/plugin v0.5.0/go.mod h1:oN23l+yGDCIP9G3KnDl/I/0zVGOobZUDCB2Z5yYYXts=
github.com/go-gost/relay v0.6.1 h1:7SqnHFbY8x/DzvjpK03a5zcVH9+TbJAcnW/RT6s1ecc=
github.com/go-gost/relay v0.6.1/go.mod h1:Dku0f5sfjOClrZFiDmQUrYYJ4uof7rnkCUBfsl0PSAI=
github.com/go-gost/relay v0.6.2 h1:0Rgm1TOZvrilAKj6zn+UMvAGWtS0H08e2wBX03Y6g+I=
github.com/go-gost/relay v0.6.2/go.mod h1:Dku0f5sfjOClrZFiDmQUrYYJ4uof7rnkCUBfsl0PSAI=
github.com/go-gost/tls-dissector v0.2.0 h1:9tE6WOzzpurATTBWn60DU4R8gibpGNY8/qVcc1SicVg=
github.com/go-gost/tls-dissector v0.2.0/go.mod h1:/9QfdewqmHdaE362Hv5nDaSWLx3pCmtD870d6GaquXs=
github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
internal/util/pht/client.go
+14 -73
View File
@@ -2,27 +2,23 @@ package pht
import (
"context"
"errors"
"fmt"
"io"
"net"
"net/http"
"net/http/httputil"
"strconv"
"strings"
"github.com/go-gost/core/logger"
"github.com/rs/xid"
)
type Client struct {
Host string
Client *http.Client
AuthorizePath string
PushPath string
PullPath string
TLSEnabled bool
Header http.Header
Logger logger.Logger
Host string
Client *http.Client
PushPath string
PullPath string
TLSEnabled bool
Header http.Header
Logger logger.Logger
}
func (c *Client) Dial(ctx context.Context, addr string) (net.Conn, error) {
@@ -36,17 +32,15 @@ func (c *Client) Dial(ctx context.Context, addr string) (net.Conn, error) {
addr = net.JoinHostPort(c.Host, strconv.Itoa(raddr.Port))
}
token, err := c.authorize(ctx, addr)
if err != nil {
c.Logger.Error(err)
return nil, err
}
connCtx, cancel := context.WithCancel(ctx)
cid := xid.New().String()
cn := &clientConn{
client: c.Client,
header: c.Header,
rxc: make(chan []byte, 128),
closed: make(chan struct{}),
ctx: connCtx,
cancel: cancel,
localAddr: &net.TCPAddr{},
remoteAddr: raddr,
logger: c.Logger,
@@ -56,63 +50,10 @@ func (c *Client) Dial(ctx context.Context, addr string) (net.Conn, error) {
if c.TLSEnabled {
scheme = "https"
}
cn.pushURL = fmt.Sprintf("%s://%s%s?token=%s", scheme, addr, c.PushPath, token)
cn.pullURL = fmt.Sprintf("%s://%s%s?token=%s", scheme, addr, c.PullPath, token)
cn.pushURL = fmt.Sprintf("%s://%s%s?token=%s", scheme, addr, c.PushPath, cid)
cn.pullURL = fmt.Sprintf("%s://%s%s?token=%s", scheme, addr, c.PullPath, cid)
go cn.readLoop()
return cn, nil
}
func (c *Client) authorize(ctx context.Context, addr string) (token string, err error) {
var url string
if c.TLSEnabled {
url = fmt.Sprintf("https://%s%s", addr, c.AuthorizePath)
} else {
url = fmt.Sprintf("http://%s%s", addr, c.AuthorizePath)
}
r, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
return
}
// Inject custom headers
if c.Header != nil {
for k, values := range c.Header {
for _, v := range values {
r.Header.Add(k, v)
}
}
}
if c.Logger.IsLevelEnabled(logger.TraceLevel) {
dump, _ := httputil.DumpRequest(r, false)
c.Logger.Trace(string(dump))
} else if c.Logger.IsLevelEnabled(logger.DebugLevel) {
c.Logger.Debugf("%s %s", r.Method, r.URL)
}
resp, err := c.Client.Do(r)
if err != nil {
return
}
defer resp.Body.Close()
if c.Logger.IsLevelEnabled(logger.TraceLevel) {
dump, _ := httputil.DumpResponse(resp, false)
c.Logger.Trace(string(dump))
}
data, err := io.ReadAll(resp.Body)
if err != nil {
return
}
if strings.HasPrefix(string(data), "token=") {
token = strings.TrimPrefix(string(data), "token=")
}
if token == "" {
err = errors.New("authorize failed")
}
return
}
internal/util/pht/conn.go
+21 -10
View File
@@ -3,6 +3,7 @@ package pht
import (
"bufio"
"bytes"
"context"
"encoding/base64"
"errors"
"io"
@@ -24,6 +25,8 @@ type clientConn struct {
rxc chan []byte
closed chan struct{}
mu sync.Mutex
ctx context.Context
cancel context.CancelFunc
localAddr net.Addr
remoteAddr net.Addr
logger logger.Logger
@@ -33,6 +36,10 @@ func (c *clientConn) Read(b []byte) (n int, err error) {
if len(c.buf) == 0 {
select {
case c.buf = <-c.rxc:
if c.buf == nil {
err = io.EOF
return
}
case <-c.closed:
err = io.ErrClosedPipe
return
@@ -53,11 +60,6 @@ func (c *clientConn) Write(b []byte) (n int, err error) {
}
func (c *clientConn) write(b []byte) (n int, err error) {
if c.isClosed() {
err = io.ErrClosedPipe
return
}
var r io.Reader
if len(b) > 0 {
buf := bytes.NewBufferString(base64.StdEncoding.EncodeToString(b))
@@ -65,7 +67,7 @@ func (c *clientConn) write(b []byte) (n int, err error) {
r = buf
}
req, err := http.NewRequest(http.MethodPost, c.pushURL, r)
req, err := http.NewRequestWithContext(c.ctx, http.MethodPost, c.pushURL, r)
if err != nil {
return
}
@@ -107,6 +109,8 @@ func (c *clientConn) write(b []byte) (n int, err error) {
}
func (c *clientConn) readLoop() {
defer close(c.rxc)
for {
if c.isClosed() {
return
@@ -114,7 +118,7 @@ func (c *clientConn) readLoop() {
done := true
err := func() error {
r, err := http.NewRequest(http.MethodGet, c.pullURL, nil)
r, err := http.NewRequestWithContext(c.ctx, http.MethodGet, c.pullURL, nil)
if err != nil {
return err
}
@@ -171,7 +175,9 @@ func (c *clientConn) readLoop() {
}()
if err != nil {
c.Close()
// Don't call Close() here — it would close(c.closed) again
// and the two callers (Dial and Close) already own that.
// Just stop the read loop; the caller holds the close.
return
}
@@ -202,9 +208,14 @@ func (c *clientConn) Close() error {
c.mu.Unlock()
_, err := c.write(nil)
// Send empty POST to notify server the tunnel is closed.
_, _ = c.write(nil)
return err
// Cancel readLoop after sending the close notification so the
// HTTP GET is canceled once the notification is delivered.
c.cancel()
return nil
}
func (c *clientConn) isClosed() bool {
internal/util/pht/server.go
+62 -26
View File
@@ -252,9 +252,22 @@ func (s *Server) handleAuthorize(w http.ResponseWriter, r *http.Request) {
raddr = &net.TCPAddr{}
}
// connection id
cid := xid.New().String()
if _, err := s.createPipe(cid, raddr); err != nil {
s.options.logger.Warnf("connection queue is full, client %s discarded", r.RemoteAddr)
w.WriteHeader(http.StatusTooManyRequests)
return
}
w.Write([]byte(fmt.Sprintf("token=%s", cid)))
}
// createPipe creates a net.Pipe pair, enqueues the client side to the Accept
// queue, and stores the server side in conns under cid. It returns the server
// side (c2). On a full accept queue it closes both ends and returns an error.
// Uses LoadOrStore to avoid a second pipe pair when concurrent push/pull
// requests arrive for the same CID before either stores into conns.
func (s *Server) createPipe(cid string, raddr net.Addr) (net.Conn, error) {
c1, c2 := net.Pipe()
c := &serverConn{
Conn: c1,
@@ -262,17 +275,38 @@ func (s *Server) handleAuthorize(w http.ResponseWriter, r *http.Request) {
remoteAddr: raddr,
}
existing, loaded := s.conns.LoadOrStore(cid, c2)
if loaded {
// Another goroutine already created the pipe for this CID.
c.Close()
c2.Close()
return existing.(net.Conn), nil
}
select {
case s.cqueue <- c:
default:
s.conns.Delete(cid)
c.Close()
s.options.logger.Warnf("connection queue is full, client %s discarded", r.RemoteAddr)
w.WriteHeader(http.StatusTooManyRequests)
c2.Close()
return nil, errors.New("connection queue full")
}
return c2, nil
}
// getOrCreateConn returns the pipe for cid, creating it lazily if needed.
func (s *Server) getOrCreateConn(cid string, r *http.Request) (conn net.Conn, err error) {
v, ok := s.conns.Load(cid)
if ok {
conn = v.(net.Conn)
return
}
w.Write([]byte(fmt.Sprintf("token=%s", cid)))
s.conns.Store(cid, c2)
raddr, _ := net.ResolveTCPAddr("tcp", r.RemoteAddr)
if raddr == nil {
raddr = &net.TCPAddr{}
}
conn, err = s.createPipe(cid, raddr)
return
}
func (s *Server) handlePush(w http.ResponseWriter, r *http.Request) {
@@ -294,12 +328,11 @@ func (s *Server) handlePush(w http.ResponseWriter, r *http.Request) {
}
cid := r.Form.Get("token")
v, ok := s.conns.Load(cid)
if !ok {
w.WriteHeader(http.StatusForbidden)
conn, err := s.getOrCreateConn(cid, r)
if err != nil {
w.WriteHeader(http.StatusTooManyRequests)
return
}
conn := v.(net.Conn)
br := bufio.NewReader(r.Body)
data, err := br.ReadString('\n')
@@ -315,6 +348,8 @@ func (s *Server) handlePush(w http.ResponseWriter, r *http.Request) {
data = strings.TrimSuffix(data, "\n")
if len(data) == 0 {
conn.Close()
s.conns.Delete(cid)
return
}
@@ -357,14 +392,12 @@ func (s *Server) handlePull(w http.ResponseWriter, r *http.Request) {
}
cid := r.Form.Get("token")
v, ok := s.conns.Load(cid)
if !ok {
w.WriteHeader(http.StatusForbidden)
conn, err := s.getOrCreateConn(cid, r)
if err != nil {
w.WriteHeader(http.StatusTooManyRequests)
return
}
conn := v.(net.Conn)
// Disable per-response buffering so streaming proxies forward each flushed
// chunk immediately. Without this, Nginx with proxy_buffering on (the
// default) holds the tiny streamed chunks in its buffer and the response
@@ -386,7 +419,7 @@ func (s *Server) handlePull(w http.ResponseWriter, r *http.Request) {
bw.WriteString(base64.StdEncoding.EncodeToString(b[:n]))
bw.WriteString("\n")
if err := bw.Flush(); err != nil {
return
break
}
if fw, ok := w.(http.Flusher); ok {
fw.Flush()
@@ -394,18 +427,21 @@ func (s *Server) handlePull(w http.ResponseWriter, r *http.Request) {
}
if err != nil {
if errors.Is(err, os.ErrDeadlineExceeded) {
b[0] = '\n' // no data
w.Write(b[:1])
} else if errors.Is(err, io.EOF) {
// server connection closed
} else {
if !errors.Is(err, io.ErrClosedPipe) {
s.options.logger.Error(err)
if _, werr := w.Write([]byte{'\n'}); werr != nil {
break // heartbeat write failed, client disconnected
}
s.conns.Delete(cid)
conn.Close()
if fw, ok := w.(http.Flusher); ok {
fw.Flush()
}
continue
}
return
if !errors.Is(err, io.EOF) && !errors.Is(err, io.ErrClosedPipe) {
s.options.logger.Error(err)
}
break
}
}
s.conns.Delete(cid)
conn.Close()
}
internal/util/pht/server_test.go
+166
View File
@@ -0,0 +1,166 @@
package pht
import (
"net"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"github.com/go-gost/core/logger"
)
func TestCreatePipeConcurrent(t *testing.T) {
s := NewServer(":0",
PathServerOption("/authorize", "/push", "/pull"),
LoggerServerOption(nopLogger{}),
)
// Drain the accept queue so we don't block on send.
go func() {
for range s.cqueue {
}
}()
cid := "test-cid"
raddr := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 12345}
const n = 20
var wg sync.WaitGroup
wg.Add(n)
conns := make([]net.Conn, n)
for i := 0; i < n; i++ {
go func(idx int) {
defer wg.Done()
conn, err := s.createPipe(cid, raddr)
if err != nil {
t.Errorf("createPipe %d failed: %v", idx, err)
return
}
conns[idx] = conn
}(i)
}
wg.Wait()
// All goroutines must reference the same pipe.
var first net.Conn
for i := 0; i < n; i++ {
if conns[i] == nil {
t.Fatalf("goroutine %d returned nil conn", i)
}
if first == nil {
first = conns[i]
} else if conns[i] != first {
t.Fatalf("goroutine 0 and %d returned different pipes — LoadOrStore race", i)
}
}
// Verify only one entry in conns after cleanup.
first.Close()
s.conns.Delete(cid)
}
func TestCreatePipeQueueFull(t *testing.T) {
s := NewServer(":0",
BacklogServerOption(0),
PathServerOption("/authorize", "/push", "/pull"),
LoggerServerOption(nopLogger{}),
)
// Force the cqueue channel buffer to 0 so every send blocks.
s.cqueue = make(chan net.Conn, 0)
_, err := s.createPipe("cid", &net.TCPAddr{})
if err == nil || err.Error() != "connection queue full" {
t.Fatalf("expected 'connection queue full', got %v", err)
}
}
func TestCloseNotification(t *testing.T) {
// Verify that LoadOrStore correctly returns the existing pipe on collision.
s := NewServer(":0",
PathServerOption("/authorize", "/push", "/pull"),
LoggerServerOption(nopLogger{}),
)
go func() {
for range s.cqueue {
}
}()
cid := "close-test"
raddr := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 1}
conn1, err := s.createPipe(cid, raddr)
if err != nil {
t.Fatal(err)
}
// Second createPipe for same CID returns the existing conn.
conn2, err := s.createPipe(cid, raddr)
if err != nil {
t.Fatal(err)
}
if conn1 != conn2 {
t.Fatal("createPipe returned different conn for same CID")
}
// Clean up.
conn1.Close()
s.conns.Delete(cid)
}
func TestHandlePushEmptyData(t *testing.T) {
s := NewServer(":0",
PathServerOption("/authorize", "/push", "/pull"),
LoggerServerOption(nopLogger{}),
)
go func() {
for range s.cqueue {
}
}()
// Pre-create a pipe entry for the CID.
cid := "empty-data-test"
raddr := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 1}
conn, err := s.createPipe(cid, raddr)
if err != nil {
t.Fatal(err)
}
// Send POST with body "\n" (empty data after trim).
body := strings.NewReader("\n")
req := httptest.NewRequest(http.MethodPost, "/push?token="+cid, body)
w := httptest.NewRecorder()
s.handlePush(w, req)
// Verify no 4xx/5xx status (should not crash/leak).
if w.Code >= 400 {
t.Fatalf("unexpected status %d", w.Code)
}
// The pipe and conns entry should be cleaned up by the empty-data path.
// The pipe is closed, so writing should fail.
_, err = conn.Write([]byte("test"))
if err == nil {
t.Fatal("expected write to closed pipe to fail")
}
}
// nopLogger implements logger.Logger with no-ops.
type nopLogger struct{}
func (l nopLogger) WithFields(map[string]any) logger.Logger { return l }
func (nopLogger) Debug(args ...any) {}
func (nopLogger) Debugf(format string, args ...any) {}
func (nopLogger) Info(args ...any) {}
func (nopLogger) Infof(format string, args ...any) {}
func (nopLogger) Warn(args ...any) {}
func (nopLogger) Warnf(format string, args ...any) {}
func (nopLogger) Error(args ...any) {}
func (nopLogger) Errorf(format string, args ...any) {}
func (nopLogger) Fatal(args ...any) {}
func (nopLogger) Fatalf(format string, args ...any) {}
func (nopLogger) Trace(args ...any) {}
func (nopLogger) Tracef(format string, args ...any) {}
func (nopLogger) GetLevel() logger.LogLevel { return logger.InfoLevel }
func (nopLogger) IsLevelEnabled(level logger.LogLevel) bool { return false }
listener/pht/listener.go
+58 -16
View File
@@ -3,6 +3,7 @@
package pht
import (
"errors"
"net"
"github.com/go-gost/core/limiter"
@@ -12,6 +13,7 @@ import (
admission "github.com/go-gost/x/admission/wrapper"
xnet "github.com/go-gost/x/internal/net"
pht_util "github.com/go-gost/x/internal/util/pht"
climiter "github.com/go-gost/x/limiter/conn/wrapper"
limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper"
metrics "github.com/go-gost/x/metrics/wrapper"
stats "github.com/go-gost/x/observer/stats/wrapper"
@@ -27,6 +29,8 @@ type phtListener struct {
addr net.Addr
tlsEnabled bool
server *pht_util.Server
cqueue chan net.Conn
errChan chan error
logger logger.Logger
md metadata
options listener.Options
@@ -79,32 +83,70 @@ func (l *phtListener) Init(md md.Metadata) (err error) {
pht_util.MPTCPServerOption(l.md.mptcp),
)
l.cqueue = make(chan net.Conn, l.md.backlog)
l.errChan = make(chan error, 1)
go func() {
if err := l.server.ListenAndServe(); err != nil {
l.logger.Error(err)
l.errChan <- err
}
close(l.errChan)
}()
go func() {
for {
conn, err := l.server.Accept()
if err != nil {
return // server closed; ListenAndServe goroutine handles errChan
}
l.cqueue <- conn
}
}()
// Verify the server started successfully. If ListenAndServe returned
// immediately with an error (e.g., port already in use), surface it
// now and clean up the bridging goroutine.
select {
case err = <-l.errChan:
l.server.Close()
return
default:
}
return
}
func (l *phtListener) Accept() (conn net.Conn, err error) {
conn, err = l.server.Accept()
if err != nil {
return
var ok bool
select {
case conn = <-l.cqueue:
if l.options.ConnLimiter != nil {
host, _, _ := net.SplitHostPort(conn.RemoteAddr().String())
if lim := l.options.ConnLimiter.Limiter(host); lim != nil {
if !lim.Allow(1) {
conn.Close()
return nil, errors.New("connection limit exceeded")
}
conn = climiter.WrapConn(lim, conn)
}
}
conn = metrics.WrapConn(l.options.Service, conn)
conn = stats.WrapConn(conn, l.options.Stats)
conn = admission.WrapConn(l.options.Admission, conn)
conn = limiter_wrapper.WrapConn(
conn,
l.options.TrafficLimiter,
conn.RemoteAddr().String(),
limiter.ScopeOption(limiter.ScopeConn),
limiter.ServiceOption(l.options.Service),
limiter.NetworkOption(conn.LocalAddr().Network()),
limiter.SrcOption(conn.RemoteAddr().String()),
)
case err, ok = <-l.errChan:
if !ok {
err = listener.ErrClosed
}
}
conn = metrics.WrapConn(l.options.Service, conn)
conn = stats.WrapConn(conn, l.options.Stats)
conn = admission.WrapConn(l.options.Admission, conn)
conn = limiter_wrapper.WrapConn(
conn,
l.options.TrafficLimiter,
conn.RemoteAddr().String(),
limiter.ScopeOption(limiter.ScopeConn),
limiter.ServiceOption(l.options.Service),
limiter.NetworkOption(conn.LocalAddr().Network()),
limiter.SrcOption(conn.RemoteAddr().String()),
)
return
}