diff --git a/README.md b/README.md index 24a5eb0c..b461b749 100644 --- a/README.md +++ b/README.md @@ -205,6 +205,8 @@ Easyss 通过检测配置文件自动区分模式： 在 `easyss` 所在目录下新建文本文件（如 `direct.txt`、`proxy.txt`），IP/CIDR/域名可混写，每行一条记录。然后在配置中指定路径： +> 配置中的相对路径（`direct_file`、`proxy_file`、`ca_path` 等）会先按当前工作目录查找，找不到时自动回退到 `easyss` 可执行文件所在目录（macOS 下为 `.app` 旁）。这样从 Finder 双击、开机自启（launchd）等方式启动时也能正常读取，不受启动目录影响。 + **简化模式：** ```json diff --git a/client/config/config.go b/client/config/config.go index 405a67d3..1f510e98 100644 --- a/client/config/config.go +++ b/client/config/config.go @@ -10,6 +10,7 @@ import ( utls "github.com/refraction-networking/utls" "github.com/nange/easyss/v3/config" + "github.com/nange/easyss/v3/util" ) // DirectDNSServers are the public DNS servers used for direct (non-proxied) DNS lookups. @@ -56,7 +57,7 @@ type TransportConfig struct { StreamThreshold int `json:"stream_threshold"` PrioritySlotRatio float64 `json:"priority_slot_ratio"` ConnLifetimeSec int `json:"conn_lifetime_sec"` // max connection lifetime in seconds, 0 uses default - ConnMaxBytes int64 `json:"conn_max_bytes"` // max bytes per connection, 0 uses default + ConnMaxBytes int64 `json:"conn_max_bytes"` // max bytes carried by a connection in either direction, 0 uses default } type ShaperConfig struct { @@ -222,6 +223,19 @@ func applyDefaults(c *ClientConfig) { } } +// ResolveFilePaths resolves relative file paths in the config against the +// executable directory when they cannot be found in the current working +// directory. On macOS the app is often launched by Finder/launchd with cwd=/, +// so relative paths like direct.txt, proxy.txt or ca_path would otherwise not +// be found even though the files sit next to the binary/.app bundle. +func (c *ClientConfig) ResolveFilePaths() { + c.Routing.DirectFile = util.ResolvePath(c.Routing.DirectFile) + c.Routing.ProxyFile = util.ResolvePath(c.Routing.ProxyFile) + for _, srv := range c.Servers { + srv.CAPath = util.ResolvePath(srv.CAPath) + } +} + func (c *ClientConfig) Clone() *ClientConfig { data, err := json.Marshal(c) if err != nil { diff --git a/client/config/config_test.go b/client/config/config_test.go index f5fca6c5..188aa80d 100644 --- a/client/config/config_test.go +++ b/client/config/config_test.go @@ -7,6 +7,7 @@ import ( "testing" "github.com/nange/easyss/v3/config" + "github.com/nange/easyss/v3/util" ) func TestDefaultConfig(t *testing.T) { @@ -640,3 +641,38 @@ func TestApplyDefaults(t *testing.T) { } }) } + +func TestResolveFilePaths(t *testing.T) { + relDirect := "direct.txt" + relCA := "ca.pem" + abs, err := filepath.Abs("proxy.txt") + if err != nil { + t.Fatal(err) + } + + cfg := &ClientConfig{ + Routing: RoutingConfig{ + DirectFile: relDirect, + ProxyFile: abs, + }, + Servers: []*ServerProfile{ + {CAPath: relCA}, + {CAPath: ""}, + }, + } + + cfg.ResolveFilePaths() + + if want := filepath.Join(util.CurrentDir(), relDirect); cfg.Routing.DirectFile != want { + t.Errorf("DirectFile = %q, want %q", cfg.Routing.DirectFile, want) + } + if cfg.Routing.ProxyFile != abs { + t.Errorf("ProxyFile = %q, want %q (absolute unchanged)", cfg.Routing.ProxyFile, abs) + } + if want := filepath.Join(util.CurrentDir(), relCA); cfg.Servers[0].CAPath != want { + t.Errorf("Servers[0].CAPath = %q, want %q", cfg.Servers[0].CAPath, want) + } + if cfg.Servers[1].CAPath != "" { + t.Errorf("Servers[1].CAPath = %q, want empty (unchanged)", cfg.Servers[1].CAPath) + } +} diff --git a/cmd/easyss-server/main.go b/cmd/easyss-server/main.go index 797d7de5..d316ad97 100644 --- a/cmd/easyss-server/main.go +++ b/cmd/easyss-server/main.go @@ -43,6 +43,11 @@ func main() { os.Exit(0) } + // On macOS the server is often launched by launchd with cwd=/, + // so a relative config path is first looked up in the cwd and then + // falls back to the executable directory. + configFile = util.ResolvePath(configFile) + data, err := os.ReadFile(configFile) if err != nil { log.Error("[EASYSS-SERVER-V3] read config", "err", err) @@ -55,6 +60,11 @@ func main() { log.Error("[EASYSS-SERVER-V3] parse config", "err", err) os.Exit(1) } + // Resolve relative file paths (cert_path/key_path/next_proxy_file) + // against the executable directory so that macOS launchd launches + // (cwd=/) can still find the files placed next to the binary. + // Must run before EffectiveServerConfig, which copies by value. + fileCfg.ResolveFilePaths() cfg = fileCfg.EffectiveServerConfig() if pprofEnabled { cfg.PprofEnabled = true diff --git a/cmd/easyss/main.go b/cmd/easyss/main.go index 08e98f9f..5eb3333c 100644 --- a/cmd/easyss/main.go +++ b/cmd/easyss/main.go @@ -88,16 +88,10 @@ func main() { os.Exit(runTunHelper(tunHTTPAddr, tunFDSocket, logFile, sc.LogLevel)) } - if !filepath.IsAbs(configFile) { - if _, err := os.Stat(configFile); os.IsNotExist(err) { - if dir := util.CurrentDir(); dir != "" { - altPath := filepath.Join(dir, configFile) - if _, err := os.Stat(altPath); err == nil { - configFile = altPath - } - } - } - } + // On macOS the app is often launched by Finder/launchd with cwd=/, + // so a relative config path is first looked up in the cwd and then + // falls back to the executable directory. + configFile = util.ResolvePath(configFile) cfg, err := config.LoadConfig(configFile) if err != nil { @@ -115,6 +109,11 @@ func main() { config.ApplySimpleOverrides(cfg, sc) } + // Resolve relative file paths (direct_file/proxy_file/ca_path) against + // the executable directory so that macOS Finder/launchd launches (cwd=/) + // can still find the files placed next to the binary/.app bundle. + cfg.ResolveFilePaths() + if cfg.Log.FilePath != "" && !filepath.IsAbs(cfg.Log.FilePath) { if dir := util.CurrentDir(); dir != "" { cfg.Log.FilePath = filepath.Join(dir, cfg.Log.FilePath) diff --git a/config/types.go b/config/types.go index 862df92f..6ac9a42d 100644 --- a/config/types.go +++ b/config/types.go @@ -64,11 +64,17 @@ const ( // either limit stops accepting new streams and its idle connection is // closed, so the next stream dials a fresh one (invisible to users). DefaultConnLifetimeSec = 900 // 15min - DefaultConnMaxBytes = 150 * 1024 * 1024 // 150MB + DefaultConnMaxBytes = 150 * 1024 * 1024 // 150MB，双向（上下行）累计 - HTTP2ServerMaxReadFrameSize = 1<<24 - 1 // 16MB-1，nginx/Cloudflare 主流值 - HTTP2ServerReceiveBufferPerConnection = 1 << 20 // 1MB，避免 64KB 瓶颈导致长期运行吞吐量下降 - HTTP2ServerReceiveBufferPerStream = 256 * 1024 // 256KB，流级别接收窗口 + // Upload flow control on the server side: the per-stream window bounds + // a single upload stream's in-flight data, capping its throughput at + // roughly window/RTT. 256KB would pin a single-stream upload to + // ~6.8Mbps on a 300ms link; 1MB (the stdlib server default) raises that + // to ~26Mbps, and the 4MB connection window keeps aggregate uploads + // from being constrained to ~1MB in flight per connection. + HTTP2ServerMaxReadFrameSize = 1<<24 - 1 // 16MB-1，nginx/Cloudflare 主流值 + HTTP2ServerReceiveBufferPerConnection = 4 << 20 // 4MB，连接级上行窗口 + HTTP2ServerReceiveBufferPerStream = 1 << 20 // 1MB，流级上行窗口（stdlib 服务端默认） HTTP2ClientMaxReadFrameSize = 1 * 1024 * 1024 // 1MB，Chrome MAX_FRAME_SIZE HTTP2ClientReceiveBufferPerConnection = 15 * 1024 * 1024 // ~15MB，Chrome 连接级窗口 diff --git a/server/config/config.go b/server/config/config.go index 56f13ce8..57502896 100644 --- a/server/config/config.go +++ b/server/config/config.go @@ -1,5 +1,7 @@ package config +import "github.com/nange/easyss/v3/util" + type LogConfig struct { Level string `json:"level"` FilePath string `json:"file_path"` @@ -51,6 +53,19 @@ func (fc *FileConfig) EffectiveServerConfig() ServerConfig { return cfg } +// ResolveFilePaths resolves relative file paths in the config against the +// executable directory when they cannot be found in the current working +// directory. Must be called before EffectiveServerConfig since the latter +// copies the Server/NextProxy structs by value. On macOS the server is often +// launched by launchd with cwd=/, so relative paths like cert_path/key_path +// or next_proxy_file would otherwise not be found even though the files sit +// next to the binary. +func (fc *FileConfig) ResolveFilePaths() { + fc.Server.CertPath = util.ResolvePath(fc.Server.CertPath) + fc.Server.KeyPath = util.ResolvePath(fc.Server.KeyPath) + fc.NextProxy.NextProxyFile = util.ResolvePath(fc.NextProxy.NextProxyFile) +} + func (c *ServerConfig) GetAllowedMethods() []string { if len(c.AllowedMethods) == 0 { return []string{"aes-256-gcm", "chacha20-poly1305"} diff --git a/server/config/config_test.go b/server/config/config_test.go index 7e568905..f3189dc8 100644 --- a/server/config/config_test.go +++ b/server/config/config_test.go @@ -2,9 +2,12 @@ package config import ( "encoding/json" + "path/filepath" "testing" "github.com/stretchr/testify/require" + + "github.com/nange/easyss/v3/util" ) func TestFileConfigEffectiveServerConfig(t *testing.T) { @@ -33,3 +36,69 @@ func TestFileConfigEffectiveServerConfig(t *testing.T) { require.Equal(t, "socks5://127.0.0.1:1080", cfg.NextProxy.URL) require.True(t, cfg.NextProxy.EnableUDP) } + +func TestResolveFilePaths(t *testing.T) { + relCert := "server.crt" + relNextProxy := "next_proxy.txt" + abs, err := filepath.Abs("server.key") + if err != nil { + t.Fatal(err) + } + + fc := &FileConfig{ + Server: ServerConfig{ + CertPath: relCert, + KeyPath: abs, + }, + NextProxy: NextProxyConfig{ + NextProxyFile: relNextProxy, + }, + } + + fc.ResolveFilePaths() + + if want := filepath.Join(util.CurrentDir(), relCert); fc.Server.CertPath != want { + t.Errorf("CertPath = %q, want %q", fc.Server.CertPath, want) + } + if fc.Server.KeyPath != abs { + t.Errorf("KeyPath = %q, want %q (absolute unchanged)", fc.Server.KeyPath, abs) + } + if want := filepath.Join(util.CurrentDir(), relNextProxy); fc.NextProxy.NextProxyFile != want { + t.Errorf("NextProxyFile = %q, want %q", fc.NextProxy.NextProxyFile, want) + } +} + +func TestResolveFilePathsEmpty(t *testing.T) { + fc := &FileConfig{} + fc.ResolveFilePaths() + + if fc.Server.CertPath != "" || fc.Server.KeyPath != "" || fc.NextProxy.NextProxyFile != "" { + t.Errorf("empty paths should stay empty, got %+v", fc) + } +} + +func TestEffectiveServerConfigCarriesResolvedPaths(t *testing.T) { + fc := &FileConfig{ + Server: ServerConfig{ + CertPath: "server.crt", + KeyPath: "server.key", + }, + NextProxy: NextProxyConfig{ + NextProxyFile: "next_proxy.txt", + }, + Timeout: 30, + } + + fc.ResolveFilePaths() + cfg := fc.EffectiveServerConfig() + + if cfg.CertPath != fc.Server.CertPath { + t.Errorf("effective CertPath = %q, want %q", cfg.CertPath, fc.Server.CertPath) + } + if cfg.NextProxy.NextProxyFile != fc.NextProxy.NextProxyFile { + t.Errorf("effective NextProxyFile = %q, want %q", cfg.NextProxy.NextProxyFile, fc.NextProxy.NextProxyFile) + } + if cfg.Timeout != 30 { + t.Errorf("Timeout = %d, want 30", cfg.Timeout) + } +} diff --git a/server/server.go b/server/server.go index d40d7add..ffdb1b4e 100644 --- a/server/server.go +++ b/server/server.go @@ -306,10 +306,23 @@ func (s *Server) Start() error { s.mux.Handle(sharedconfig.EndpointUDP, proxyHandler) s.mux.Handle(sharedconfig.EndpointICMP, proxyHandler) - s.httpServer = &http.Server{ - Addr: s.cfg.Listen, + s.httpServer = buildHTTPServer(cfg, tlsConfig, s.mux, timeout) + + log.Info("[SERVER] listening", "addr", s.cfg.Listen, "routes", []string{"/", sharedconfig.EndpointTCP, sharedconfig.EndpointUDP, sharedconfig.EndpointICMP}) + s.statsDone = make(chan struct{}) + go s.statsLoop() + return s.httpServer.ListenAndServeTLS("", "") +} + +// buildHTTPServer assembles the HTTP server with HTTP/2 flow-control windows +// sized for upload throughput: the per-stream receive window bounds a single +// upload stream's in-flight data (throughput ≈ window/RTT), so both windows +// must be generous enough for high-RTT links. +func buildHTTPServer(cfg *config.ServerConfig, tlsConfig *tls.Config, mux *http.ServeMux, timeout time.Duration) *http.Server { + srv := &http.Server{ + Addr: cfg.Listen, TLSConfig: tlsConfig, - Handler: s.mux, + Handler: mux, ErrorLog: stdErrorLog(), Protocols: &http.Protocols{}, HTTP2: &http.HTTP2Config{ @@ -320,13 +333,9 @@ func (s *Server) Start() error { IdleTimeout: 8 * timeout, ReadHeaderTimeout: min(timeout/2, 10*time.Second), } - s.httpServer.Protocols.SetHTTP1(true) - s.httpServer.Protocols.SetHTTP2(true) - - log.Info("[SERVER] listening", "addr", s.cfg.Listen, "routes", []string{"/", sharedconfig.EndpointTCP, sharedconfig.EndpointUDP, sharedconfig.EndpointICMP}) - s.statsDone = make(chan struct{}) - go s.statsLoop() - return s.httpServer.ListenAndServeTLS("", "") + srv.Protocols.SetHTTP1(true) + srv.Protocols.SetHTTP2(true) + return srv } func (s *Server) Shutdown(ctx context.Context) error { diff --git a/server/server_test.go b/server/server_test.go index 46a0d8ea..d5461d91 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -2,6 +2,11 @@ package server import ( "context" + "crypto/tls" + "encoding/binary" + "io" + "net/http" + "net/http/httptest" "os" "path/filepath" "regexp" @@ -9,6 +14,7 @@ import ( "time" "github.com/caddyserver/certmagic" + sharedconfig "github.com/nange/easyss/v3/config" "github.com/nange/easyss/v3/server/config" "github.com/stretchr/testify/require" ) @@ -137,3 +143,95 @@ func TestResolveEmail_Generated(t *testing.T) { require.NoError(t, err) require.True(t, matched, "unexpected generated email format: %s", s.cfg.Email) } + +// TestBuildHTTPServerUploadFlowControl pins the upload-side flow-control +// windows with literal lower bounds (not the constants themselves, so a +// regression that shrinks the constants also fails this test): a single +// upload stream is capped at roughly window/RTT, so the per-stream window +// must be at least 1MB (~26Mbps on a 300ms link) and the connection window +// at least 2MB. +func TestBuildHTTPServerUploadFlowControl(t *testing.T) { + srv := buildHTTPServer(&config.ServerConfig{Listen: ":443"}, nil, http.NewServeMux(), 30*time.Second) + require.NotNil(t, srv.HTTP2) + require.GreaterOrEqual(t, srv.HTTP2.MaxReceiveBufferPerStream, 1<<20, + "per-stream upload window must be >= 1MB") + require.GreaterOrEqual(t, srv.HTTP2.MaxReceiveBufferPerConnection, 2<<20, + "connection upload window must be >= 2MB") +} + +// TestServerUploadFlowControlWindowsOnWire verifies the windows the server +// actually advertises to a client: SETTINGS_INITIAL_WINDOW_SIZE carries the +// per-stream upload window, and a connection-level WINDOW_UPDATE grants the +// connection window beyond the RFC 7540 initial 65535 bytes. The stdlib +// server queues its SETTINGS and flushes them as soon as the client preface +// arrives, so the frames are immediately readable after the preface. +func TestServerUploadFlowControlWindowsOnWire(t *testing.T) { + ts := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + ts.EnableHTTP2 = true + ts.Config.Protocols = &http.Protocols{} + ts.Config.Protocols.SetHTTP2(true) + ts.Config.HTTP2 = &http.HTTP2Config{ + MaxReceiveBufferPerConnection: sharedconfig.HTTP2ServerReceiveBufferPerConnection, + MaxReceiveBufferPerStream: sharedconfig.HTTP2ServerReceiveBufferPerStream, + } + ts.StartTLS() + t.Cleanup(ts.Close) + + conn, err := tls.Dial("tcp", ts.Listener.Addr().String(), &tls.Config{ + InsecureSkipVerify: true, //nolint:gosec // test-only self-signed cert + NextProtos: []string{"h2"}, + }) + require.NoError(t, err) + defer conn.Close() //nolint:errcheck + require.Equal(t, "h2", conn.ConnectionState().NegotiatedProtocol) + + // Send the HTTP/2 client preface: the server flushes its SETTINGS and + // connection WINDOW_UPDATE once it reads it. + _, err = conn.Write([]byte("PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n")) + require.NoError(t, err) + require.NoError(t, conn.SetReadDeadline(time.Now().Add(5*time.Second))) + + const ( + frameSettings = 0x4 + frameWindowUpd = 0x8 + settingInitWin = 0x4 + initialConnWin = 65535 // RFC 7540 initial connection window + maxFramesToRead = 16 + ) + var streamWindow, connWindow = 0, initialConnWin + for i := 0; i < maxFramesToRead; i++ { + var hdr [9]byte + _, err := io.ReadFull(conn, hdr[:]) + require.NoError(t, err) + length := int(hdr[0])<<16 | int(hdr[1])<<8 | int(hdr[2]) + streamID := binary.BigEndian.Uint32(hdr[5:9]) & 0x7fffffff + + payload := make([]byte, length) + _, err = io.ReadFull(conn, payload) + require.NoError(t, err) + + switch hdr[3] { + case frameSettings: + for len(payload) >= 6 { + id := binary.BigEndian.Uint16(payload[:2]) + val := binary.BigEndian.Uint32(payload[2:6]) + payload = payload[6:] + if id == settingInitWin { + streamWindow = int(val) + } + } + case frameWindowUpd: + if streamID == 0 && len(payload) == 4 { + connWindow += int(binary.BigEndian.Uint32(payload)) + } + } + if streamWindow > 0 && connWindow > initialConnWin { + break + } + } + + require.GreaterOrEqual(t, streamWindow, 1<<20, + "advertised per-stream upload window must be >= 1MB") + require.GreaterOrEqual(t, connWindow, 2<<20, + "connection upload window must be >= 2MB") +} diff --git a/transport/http2/client.go b/transport/http2/client.go index b3191d0a..555f53e8 100644 --- a/transport/http2/client.go +++ b/transport/http2/client.go @@ -39,7 +39,7 @@ type Config struct { StreamThreshold int PrioritySlotRatio float64 ConnLifetime time.Duration // max age of a connection before rotation (0: default) - ConnMaxBytes int64 // max bytes per connection before rotation (0: default) + ConnMaxBytes int64 // max bytes carried by a connection in either direction before rotation (0: default) Timeout time.Duration DialContext func(ctx context.Context, network, addr string) (net.Conn, error) } diff --git a/transport/http2/client_test.go b/transport/http2/client_test.go index 795dc18d..f3fc00f6 100644 --- a/transport/http2/client_test.go +++ b/transport/http2/client_test.go @@ -162,8 +162,8 @@ func TestTrackReadMarksSlotHeavy(t *testing.T) { if slot.bytesRecv.Load() != int64(sharedconfig.HeavyStreamThresholdBytes+1) { t.Fatalf("bytesRecv not accumulated: %d", slot.bytesRecv.Load()) } - if slot.connBytesRecv.Load() != slot.bytesRecv.Load() { - t.Fatalf("connBytesRecv = %d, want %d", slot.connBytesRecv.Load(), slot.bytesRecv.Load()) + if slot.connBytes.Load() != slot.bytesRecv.Load() { + t.Fatalf("connBytes = %d, want %d", slot.connBytes.Load(), slot.bytesRecv.Load()) } // Further transfers must not double-mark. diff --git a/transport/http2/lifecycle.go b/transport/http2/lifecycle.go index ecae43d7..ee3c6db8 100644 --- a/transport/http2/lifecycle.go +++ b/transport/http2/lifecycle.go @@ -142,14 +142,15 @@ func (lc *slotLifecycle) evaluateRotation(idx int, s *transportSlot) { } // rotationDue reports whether the slot's connection exceeded the lifetime -// or bytes limit and should stop accepting new streams. +// or bytes limit and should stop accepting new streams. The byte limit +// counts traffic in both directions. func (lc *slotLifecycle) rotationDue(s *transportSlot, now time.Time) bool { if lc.connLifetime > 0 { if expireAt := s.expireAt.Load(); expireAt > 0 && now.UnixNano() >= expireAt { return true } } - if lc.connMaxBytes > 0 && s.connBytesRecv.Load() >= lc.connMaxBytes { + if lc.connMaxBytes > 0 && s.connBytes.Load() >= lc.connMaxBytes { return true } return false diff --git a/transport/http2/lifecycle_test.go b/transport/http2/lifecycle_test.go index bb3ec171..07dc52ff 100644 --- a/transport/http2/lifecycle_test.go +++ b/transport/http2/lifecycle_test.go @@ -132,7 +132,7 @@ func TestRotationDue(t *testing.T) { lc := &slotLifecycle{connLifetime: time.Hour, connMaxBytes: 1024} s := &transportSlot{} s.expireAt.Store(now.Add(time.Hour).UnixNano()) - s.connBytesRecv.Store(2048) + s.connBytes.Store(2048) if !lc.rotationDue(s, now) { t.Fatal("expected rotation due by bytes") } @@ -142,7 +142,7 @@ func TestRotationDue(t *testing.T) { lc := &slotLifecycle{connLifetime: time.Minute, connMaxBytes: 1024} s := &transportSlot{} s.expireAt.Store(now.Add(time.Minute).UnixNano()) - s.connBytesRecv.Store(512) + s.connBytes.Store(512) if lc.rotationDue(s, now) { t.Fatal("fresh connection must not rotate") } diff --git a/transport/http2/scheduler.go b/transport/http2/scheduler.go index 3b31df90..066897d9 100644 --- a/transport/http2/scheduler.go +++ b/transport/http2/scheduler.go @@ -105,17 +105,17 @@ func (s *slotScheduler) leastActiveInRange(start, end int) *transportSlot { return s.slots[0] } -// grow activates one more live slot (up to maxSlots) when every eligible -// slot that a new stream of this class would use is at or above the -// threshold. Uses double-checked locking. -func (s *slotScheduler) grow(highPriority bool) { - live := s.liveCount.Load() - if int(live) >= s.maxSlots { - return - } - - thresh := s.threshold - start, end := int32(0), live +// growRange returns the slot range a new stream of the given class would be +// scheduled onto, plus the saturation threshold used to decide whether one +// more slot is needed: priority streams prefer [0, prioritySlots), bulk +// streams prefer [prioritySlots, live). Like pick, both fall back to the +// whole live range when their own range is empty — in particular, bulk +// streams schedule over the entire pool while live < prioritySlots, so a +// pure bulk workload must be able to grow the pool instead of piling onto +// the initial connections forever. +func (s *slotScheduler) growRange(highPriority bool, live int32) (start, end, thresh int32) { + thresh = s.threshold + start, end = 0, live if highPriority && s.prioritySlots > 0 { end = int32(s.prioritySlots) if end > live { @@ -124,8 +124,26 @@ func (s *slotScheduler) grow(highPriority bool) { } else if s.prioritySlots > 0 { start = int32(s.prioritySlots) thresh = s.bulkThreshold + if start >= end { + // The bulk range is still empty (live <= prioritySlots): bulk + // streams fall back onto the whole live range, so growth must + // use that same range. + start = 0 + } } + return start, end, thresh +} +// grow activates one more live slot (up to maxSlots) when every eligible +// slot that a new stream of this class would use is at or above the +// threshold. Uses double-checked locking. +func (s *slotScheduler) grow(highPriority bool) { + live := s.liveCount.Load() + if int(live) >= s.maxSlots { + return + } + + start, end, thresh := s.growRange(highPriority, live) if live > 0 { if start >= end { return @@ -145,20 +163,12 @@ func (s *slotScheduler) grow(highPriority bool) { if int(live) >= s.maxSlots { return } - start2, end2 := int32(0), live - if highPriority && s.prioritySlots > 0 { - end2 = int32(s.prioritySlots) - if end2 > live { - end2 = live - } - } else if s.prioritySlots > 0 { - start2 = int32(s.prioritySlots) - } + start, end, thresh = s.growRange(highPriority, live) if live > 0 { - if start2 >= end2 { + if start >= end { return } - if !s.needsMore(start2, end2, thresh) { + if !s.needsMore(start, end, thresh) { return } } diff --git a/transport/http2/scheduler_test.go b/transport/http2/scheduler_test.go index 3b9d1fe9..bdf816b6 100644 --- a/transport/http2/scheduler_test.go +++ b/transport/http2/scheduler_test.go @@ -175,3 +175,88 @@ func TestRemoveShrinksLiveCount(t *testing.T) { t.Fatalf("busy slot removed: liveCount = %d", got) } } + +// newGrowTestScheduler builds a scheduler with maxSlots live slots where +// only `live` are active, threshold 4 and prioritySlots priority-class slots +// (bulk threshold 8). +func newGrowTestScheduler(maxSlots, prioritySlots, live int) *slotScheduler { + slots := make([]*transportSlot, maxSlots) + for i := range slots { + slots[i] = &transportSlot{} + } + sch := newScheduler(maxSlots, slots, 4, prioritySlots) + sch.liveCount.Store(int32(live)) + return sch +} + +// TestGrowBulkOnlyWorkloadGrowsPool guards the bulk-range fallback in grow: +// while live < prioritySlots the bulk range is empty, and a pure bulk +// workload (no priority streams to drive growth) must still grow the pool +// past the initial connections instead of piling onto them forever. +func TestGrowBulkOnlyWorkloadGrowsPool(t *testing.T) { + sch := newGrowTestScheduler(10, 5, 2) + + // Slots below the bulk threshold (8): no growth yet. + sch.slots[0].active.Store(4) + sch.slots[1].active.Store(7) + sch.grow(false) + if got := sch.liveCount.Load(); got != 2 { + t.Fatalf("liveCount = %d, want 2 while slots still have capacity", got) + } + + // Every live slot at or above the bulk threshold: growth must fire even + // though the bulk range [prioritySlots, live) is empty. + sch.slots[0].active.Store(8) + sch.slots[1].active.Store(9) + sch.grow(false) + if got := sch.liveCount.Load(); got != 3 { + t.Fatalf("liveCount = %d, want 3 (bulk workload must grow the pool)", got) + } + + // The fallback keeps working until the bulk range becomes non-empty: + // the newly activated slot starts idle, so saturation of all live slots + // keeps growing the pool. + sch.slots[2].active.Store(8) + sch.grow(false) + if got := sch.liveCount.Load(); got != 4 { + t.Fatalf("liveCount = %d, want 4", got) + } +} + +func TestGrowBulkUsesBulkRangeOnceLive(t *testing.T) { + sch := newGrowTestScheduler(10, 5, 6) // slots 0-4 priority, slot 5 bulk + + // Bulk slot below threshold: no growth. + sch.slots[5].active.Store(7) + sch.grow(false) + if got := sch.liveCount.Load(); got != 6 { + t.Fatalf("liveCount = %d, want 6 while bulk slot has capacity", got) + } + + // Bulk slot saturated: grow. + sch.slots[5].active.Store(8) + sch.grow(false) + if got := sch.liveCount.Load(); got != 7 { + t.Fatalf("liveCount = %d, want 7", got) + } +} + +func TestGrowPriorityUsesPriorityRange(t *testing.T) { + sch := newGrowTestScheduler(10, 5, 3) + + // Priority slot below threshold (4): no growth. + sch.slots[0].active.Store(4) + sch.slots[1].active.Store(4) + sch.slots[2].active.Store(3) + sch.grow(true) + if got := sch.liveCount.Load(); got != 3 { + t.Fatalf("liveCount = %d, want 3 while priority slot has capacity", got) + } + + // All live priority slots at threshold: grow. + sch.slots[2].active.Store(4) + sch.grow(true) + if got := sch.liveCount.Load(); got != 4 { + t.Fatalf("liveCount = %d, want 4", got) + } +} diff --git a/transport/http2/slot.go b/transport/http2/slot.go index c5e7f3d3..882c8867 100644 --- a/transport/http2/slot.go +++ b/transport/http2/slot.go @@ -22,8 +22,8 @@ type transportSlot struct { degraded atomic.Bool // Connection rotation state. - expireAt atomic.Int64 // unix nano deadline of the current connection (dial time + lifetime + jitter) - connBytesRecv atomic.Int64 // bytes downloaded over the current connection + expireAt atomic.Int64 // unix nano deadline of the current connection (dial time + lifetime + jitter) + connBytes atomic.Int64 // bytes carried over the current connection in either direction // expiring marks a slot whose connection exceeded the lifetime or bytes // limit; new streams avoid it and its idle connection is closed so the // next stream dials a fresh one. Cleared when a new connection is @@ -58,6 +58,6 @@ func (s *transportSlot) eligible(skipHeavy, skipDegraded, skipExpiring bool) boo // carried and the expiring mark all start fresh. func (s *transportSlot) resetConn(connLifetime time.Duration) { s.expireAt.Store(time.Now().Add(rotationLifetime(connLifetime)).UnixNano()) - s.connBytesRecv.Store(0) + s.connBytes.Store(0) s.expiring.Store(false) } diff --git a/transport/http2/stream.go b/transport/http2/stream.go index 64c73b2c..80cecbfe 100644 --- a/transport/http2/stream.go +++ b/transport/http2/stream.go @@ -63,12 +63,19 @@ func (s *HTTP2Stream) trackRead(n int) { return } s.slot.bytesRecv.Add(int64(n)) - s.slot.connBytesRecv.Add(int64(n)) + s.slot.connBytes.Add(int64(n)) s.accumulate(n) } -// trackWrite accumulates uploaded bytes into the heavy-stream detector. +// trackWrite accumulates uploaded bytes: into the connection rotation +// counter (rotation must trigger for upload-heavy connections too, since +// middleboxes throttle by total bytes in either direction) and the +// heavy-stream detector. func (s *HTTP2Stream) trackWrite(n int) { + if s.slot == nil || n <= 0 { + return + } + s.slot.connBytes.Add(int64(n)) s.accumulate(n) } diff --git a/transport/http2/stream_test.go b/transport/http2/stream_test.go index b5cf599e..72ed3454 100644 --- a/transport/http2/stream_test.go +++ b/transport/http2/stream_test.go @@ -5,6 +5,7 @@ import ( "io" "sync" "testing" + "time" ) func newTestStream() (*HTTP2Stream, *io.PipeReader) { @@ -111,3 +112,42 @@ func TestSetRoundTripErr_Concurrent(t *testing.T) { } s.rtErrMu.Unlock() } + +// TestHTTP2Stream_ConnBytesCountsBothDirections verifies the connection +// rotation byte counter accumulates uploaded and downloaded bytes alike: +// rotation against conn_max_bytes must trigger for upload-only traffic too, +// since middleboxes throttle by total bytes in either direction. +func TestHTTP2Stream_ConnBytesCountsBothDirections(t *testing.T) { + slot := &transportSlot{} + s, pr := newTestStream() + defer pr.Close() + defer s.Close() + s.slot = slot + s.startTime = time.Now() + + s.trackRead(1000) + s.trackWrite(2000) + + if got := slot.connBytes.Load(); got != 3000 { + t.Fatalf("connBytes = %d, want 3000 (both directions)", got) + } + // The throughput health sample stays download-only: bytesRecv must not + // include uploaded bytes. + if got := slot.bytesRecv.Load(); got != 1000 { + t.Fatalf("bytesRecv = %d, want 1000 (download only)", got) + } +} + +// TestHTTP2Stream_TrackWriteNilSlotNoOp guards the nil-slot fast path in +// trackWrite after it started counting connection bytes. +func TestHTTP2Stream_TrackWriteNilSlotNoOp(t *testing.T) { + s, pr := newTestStream() + defer pr.Close() + defer s.Close() + + s.trackWrite(1 << 20) // must not panic + + if s.heavyState.Load() != heavyIdle { + t.Fatal("nil slot must not mark heavy") + } +} diff --git a/util/file.go b/util/file.go index babf8387..71b9fb10 100644 --- a/util/file.go +++ b/util/file.go @@ -23,6 +23,29 @@ func FileExists(path string) (bool, error) { return false, err } +// ResolvePath resolves a possibly-relative file path to an absolute one when +// it cannot be found in the current working directory. On macOS the app is +// often launched by Finder or launchd with cwd=/ (LaunchAgent plists have no +// WorkingDirectory), so relative paths in config files (direct.txt, proxy.txt, +// ca_path, cert files, ...) would otherwise never be found even though they +// sit next to the binary/.app bundle. +// +// Empty strings and absolute paths are returned unchanged; relative paths that +// exist in the cwd are kept as-is for backward compatibility; anything else is +// joined with the executable directory (see CurrentDir). +func ResolvePath(p string) string { + if p == "" || filepath.IsAbs(p) { + return p + } + if _, err := os.Stat(p); err == nil { + return p + } + if dir := CurrentDir(); dir != "" { + return filepath.Join(dir, p) + } + return p +} + func CurrentDir() string { path, err := os.Executable() if err != nil { diff --git a/util/file_test.go b/util/file_test.go index 279e0a78..6df2042e 100644 --- a/util/file_test.go +++ b/util/file_test.go @@ -1,6 +1,7 @@ package util import ( + "path/filepath" "testing" "github.com/stretchr/testify/assert" @@ -42,3 +43,26 @@ func TestReadFileLinesMap(t *testing.T) { _, ok := m["Ni hao!"] assert.True(t, ok) } + +func TestResolvePath(t *testing.T) { + // Empty string is returned unchanged. + assert.Equal(t, "", ResolvePath("")) + + // Absolute paths are returned unchanged. + abs, err := filepath.Abs("direct.txt") + assert.Nil(t, err) + assert.True(t, filepath.IsAbs(abs)) + assert.Equal(t, abs, ResolvePath(abs)) + + // Relative paths that exist in the cwd are kept as-is. + relExisting := "file.go" + e, err := FileExists(relExisting) + assert.Nil(t, err) + assert.True(t, e) + assert.Equal(t, relExisting, ResolvePath(relExisting)) + + // Relative paths missing from the cwd fall back to the executable dir. + relMissing := "direct.txt" + want := filepath.Join(CurrentDir(), relMissing) + assert.Equal(t, want, ResolvePath(relMissing)) +}