diff --git a/adapter/outbound.go b/adapter/outbound.go index f688f164..73d93ddb 100644 --- a/adapter/outbound.go +++ b/adapter/outbound.go @@ -51,7 +51,8 @@ type OutboundManager interface { Create(ctx context.Context, router Router, logger log.ContextLogger, tag string, outboundType string, options any) error } -type IdleConnectionCloser interface { +type IdleConnectionKeeper interface { + SetKeepIdleConnections(keep bool) CloseIdleConnections() } diff --git a/adapter/session.go b/adapter/session.go new file mode 100644 index 00000000..4a5f3fdf --- /dev/null +++ b/adapter/session.go @@ -0,0 +1,14 @@ +package adapter + +import "context" + +type keepSessionKey struct{} + +func ContextWithKeepSession(ctx context.Context) context.Context { + return context.WithValue(ctx, (*keepSessionKey)(nil), true) +} + +func KeepSessionFromContext(ctx context.Context) bool { + keep, _ := ctx.Value((*keepSessionKey)(nil)).(bool) + return keep +} diff --git a/adapter/v2ray.go b/adapter/v2ray.go index d9370807..2dfa4102 100644 --- a/adapter/v2ray.go +++ b/adapter/v2ray.go @@ -22,3 +22,8 @@ type V2RayClientTransport interface { DialContext(ctx context.Context) (net.Conn, error) Close() error } + +type V2RayMultiplexClientTransport interface { + V2RayClientTransport + MultiplexEnabled() bool +} diff --git a/box.go b/box.go index 35037766..f1dbc3bf 100644 --- a/box.go +++ b/box.go @@ -60,6 +60,7 @@ type Box struct { dnsRouter *dns.Router connection *route.ConnectionManager router *route.Router + referenceManager *route.ReferenceManager httpClientService adapter.LifecycleService internalService []adapter.LifecycleService done chan struct{} @@ -257,7 +258,8 @@ func New(options Options) (*Box, error) { service.MustRegisterPtr(ctx, clashMode) internalServices = append(internalServices, clashMode) } - internalServices = append(internalServices, route.NewReferenceManager(ctx, logFactory.NewLogger("reference"), options.Options)) + referenceManager := route.NewReferenceManager(ctx, logFactory.NewLogger("reference"), options.Options) + internalServices = append(internalServices, referenceManager) ntpOptions := common.PtrValueOrDefault(options.NTP) var timeService *tls.TimeServiceWrapper if ntpOptions.Enabled { @@ -480,6 +482,7 @@ func New(options Options) (*Box, error) { dnsRouter: dnsRouter, connection: connectionManager, router: router, + referenceManager: referenceManager, httpClientService: httpClientService, createdAt: createdAt, debugOptions: debugOptions, @@ -682,6 +685,14 @@ func (s *Box) Endpoint() adapter.EndpointManager { return s.endpoint } +func (s *Box) CreatedAt() time.Time { + return s.createdAt +} + +func (s *Box) CloseIdleConnections() { + s.referenceManager.CloseIdleConnections() +} + func (s *Box) LogFactory() log.Factory { return s.logFactory } diff --git a/common/dialer/default.go b/common/dialer/default.go index 154835dc..25b264ff 100644 --- a/common/dialer/default.go +++ b/common/dialer/default.go @@ -433,6 +433,8 @@ func (d *DefaultDialer) trackConn(ctx context.Context, destination M.Socksaddr, attribution := d.dialAttribution(ctx, destination) outboundCounter := recorder.TrafficCounter(powerreport.TrafficOutbound, attribution.Outbound) dnsCounter := recorder.TrafficCounter(powerreport.TrafficDNSTransport, attribution.DNS) + outboundCounter.CountDial() + dnsCounter.CountDial() conn = bufio.NewCounterConn(conn, []N.CountFunc{func(n int64) { outboundCounter.CountIn(n) dnsCounter.CountIn(n) @@ -461,6 +463,8 @@ func (d *DefaultDialer) trackPacketConn(ctx context.Context, destination M.Socks attribution := d.dialAttribution(ctx, destination) outboundCounter := recorder.TrafficCounter(powerreport.TrafficOutbound, attribution.Outbound) dnsCounter := recorder.TrafficCounter(powerreport.TrafficDNSTransport, attribution.DNS) + outboundCounter.CountDial() + dnsCounter.CountDial() conn = bufio.NewNetPacketConn(bufio.NewCounterPacketConn(bufio.NewPacketConn(conn), []N.CountFunc{func(n int64) { outboundCounter.CountIn(n) dnsCounter.CountIn(n) diff --git a/common/urltest/keep_session_quic.go b/common/urltest/keep_session_quic.go new file mode 100644 index 00000000..6aebd111 --- /dev/null +++ b/common/urltest/keep_session_quic.go @@ -0,0 +1,13 @@ +//go:build with_quic + +package urltest + +import ( + "context" + + "github.com/sagernet/sing-quic" +) + +func contextWithQUICKeepSession(ctx context.Context) context.Context { + return qtls.ContextWithKeepSession(ctx) +} diff --git a/common/urltest/keep_session_quic_stub.go b/common/urltest/keep_session_quic_stub.go new file mode 100644 index 00000000..d402f906 --- /dev/null +++ b/common/urltest/keep_session_quic_stub.go @@ -0,0 +1,9 @@ +//go:build !with_quic + +package urltest + +import "context" + +func contextWithQUICKeepSession(ctx context.Context) context.Context { + return ctx +} diff --git a/common/urltest/urltest.go b/common/urltest/urltest.go index 36d44957..26efe39c 100644 --- a/common/urltest/urltest.go +++ b/common/urltest/urltest.go @@ -11,6 +11,8 @@ import ( "github.com/sagernet/sing-box/adapter" C "github.com/sagernet/sing-box/constant" + "github.com/sagernet/sing-mux" + "github.com/sagernet/sing-snell" "github.com/sagernet/sing/common" M "github.com/sagernet/sing/common/metadata" N "github.com/sagernet/sing/common/network" @@ -81,7 +83,11 @@ func (s *HistoryStorage) Close() error { func URLTest(ctx context.Context, link string, detour N.Dialer) (uint16, error) { multiplexOutbound, isMultiplexOutbound := common.Cast[adapter.OutboundWithMultiplex](detour) if isMultiplexOutbound && multiplexOutbound.MultiplexEnabled() { - _, err := urlTest(ctx, link, detour) + warmContext := adapter.ContextWithKeepSession(ctx) + warmContext = mux.ContextWithKeepSession(warmContext) + warmContext = contextWithQUICKeepSession(warmContext) + warmContext = snell.ContextWithKeepSession(warmContext) + _, err := urlTest(warmContext, link, detour) if err != nil { return 0, err } diff --git a/daemon/instance.go b/daemon/instance.go index 4040693f..e60701dc 100644 --- a/daemon/instance.go +++ b/daemon/instance.go @@ -138,7 +138,7 @@ func (s *StartedService) newInstance(ctx context.Context, profileContent string, i.clashMode = service.PtrFromContext[clashmode.Manager](ctx) i.trafficManager = service.PtrFromContext[trafficcontrol.Manager](ctx) i.pauseManager = service.FromContext[pause.Manager](ctx) - i.registerPauseCallback(ctx) + i.registerPowerReport(ctx) i.cacheFile = service.FromContext[adapter.CacheFile](ctx) i.outboundManager = service.FromContext[adapter.OutboundManager](ctx) i.endpointManager = service.FromContext[adapter.EndpointManager](ctx) @@ -176,9 +176,16 @@ func (i *Instance) Close() error { return i.instance.Close() } -func (i *Instance) registerPauseCallback(ctx context.Context) { +func (i *Instance) registerPowerReport(ctx context.Context) { powerManager := service.FromContext[*powerreport.Manager](ctx) - if i.pauseManager == nil || powerManager == nil { + if powerManager == nil { + return + } + recorder := powerManager.Recorder() + if recorder != nil { + recorder.RecordServiceStart(i.instance.CreatedAt()) + } + if i.pauseManager == nil { return } i.pauseCallback = i.pauseManager.RegisterCallback(func(event int) { diff --git a/dns/transport/conn_pool.go b/dns/transport/conn_pool.go index 0e7a20a8..1069ed7d 100644 --- a/dns/transport/conn_pool.go +++ b/dns/transport/conn_pool.go @@ -47,6 +47,8 @@ type connPoolState[T comparable] struct { shared T hasShared bool sharedClaimed bool + sharedUsers int + sharedWaiters int sharedCtx context.Context sharedCancel context.CancelCauseFunc @@ -122,6 +124,9 @@ func (p *ConnPool[T]) Release(conn T, reuse bool) { p.options.Close(conn, net.ErrClosed) return } + if p.options.Mode == ConnPoolSingle && state.hasShared && state.shared == conn && state.sharedUsers > 0 { + state.sharedUsers-- + } if p.options.Mode == ConnPoolOrdered { if _, idle := state.idleElements[conn]; !idle { state.idleElements[conn] = state.idle.PushBack(conn) @@ -182,6 +187,7 @@ func (p *ConnPool[T]) removeConn(state *connPoolState[T], conn T, cause error) { state.shared = zero state.hasShared = false state.sharedClaimed = false + state.sharedUsers = 0 state.sharedCtx = nil if state.sharedCancel != nil { state.sharedCancel(cause) @@ -196,6 +202,23 @@ func (p *ConnPool[T]) removeConn(state *connPoolState[T], conn T, cause error) { } } +func (p *ConnPool[T]) CloseIdle() { + p.access.Lock() + if p.closed { + p.access.Unlock() + return + } + state := p.state + if p.options.Mode != ConnPoolSingle || !state.hasShared || state.sharedUsers > 0 || state.sharedWaiters > 0 { + p.access.Unlock() + return + } + conn := state.shared + p.removeConn(state, conn, net.ErrClosed) + p.access.Unlock() + p.options.Close(conn, net.ErrClosed) +} + func (p *ConnPool[T]) Reset() { p.access.Lock() if p.closed { @@ -309,6 +332,7 @@ func (p *ConnPool[T]) acquireShared(ctx context.Context, dial func(context.Conte if p.options.IsAlive(conn) { created := !current.sharedClaimed current.sharedClaimed = true + current.sharedUsers++ connCtx := current.sharedCtx p.access.Unlock() return conn, connCtx, created, nil @@ -324,6 +348,7 @@ func (p *ConnPool[T]) acquireShared(ctx context.Context, dial func(context.Conte current.connecting = &connPoolConnect[T]{done: make(chan struct{})} } state := current.connecting + current.sharedWaiters++ p.access.Unlock() if startDial { @@ -338,9 +363,13 @@ func (p *ConnPool[T]) acquireShared(ctx context.Context, dial func(context.Conte } return conn, connCtx, created, err case <-ctx.Done(): + p.access.Lock() + current.sharedWaiters-- + p.access.Unlock() return zero, nil, false, ctx.Err() case <-current.ctx.Done(): p.access.Lock() + current.sharedWaiters-- closed := p.closed p.access.Unlock() if closed { @@ -400,6 +429,7 @@ func (p *ConnPool[T]) collectShared(current *connPoolState[T], state *connPoolCo var zero T p.access.Lock() + current.sharedWaiters-- if state.err != nil { err := state.err p.access.Unlock() @@ -431,6 +461,7 @@ func (p *ConnPool[T]) collectShared(current *connPoolState[T], state *connPoolCo created := !current.sharedClaimed current.sharedClaimed = true + current.sharedUsers++ connCtx := current.sharedCtx p.access.Unlock() return conn, connCtx, created, false, nil diff --git a/dns/transport/https.go b/dns/transport/https.go index 05e9f8c5..582677ec 100644 --- a/dns/transport/https.go +++ b/dns/transport/https.go @@ -34,7 +34,10 @@ import ( const MimeType = "application/dns-message" -var _ adapter.DNSTransport = (*HTTPSTransport)(nil) +var ( + _ adapter.DNSTransport = (*HTTPSTransport)(nil) + _ adapter.IdleConnectionKeeper = (*HTTPSTransport)(nil) +) func RegisterHTTPS(registry *dns.TransportRegistry) { dns.RegisterTransport[option.RemoteHTTPSDNSServerOptions](registry, C.DNSTypeHTTPS, NewHTTPS) @@ -48,6 +51,7 @@ type HTTPSTransport struct { headers http.Header serverAddr M.Socksaddr fallback *atomic.Bool + keepIdle atomic.Bool transportAccess sync.Mutex transport *HTTPSTransportWrapper transportResetAt time.Time @@ -131,7 +135,7 @@ func NewHTTPSRaw( // plain HTTP DoH used by Tailscale fallback.Store(true) } - return &HTTPSTransport{ + transport := &HTTPSTransport{ TransportAdapter: adapter, logger: logger, dialer: dialer, @@ -141,6 +145,8 @@ func NewHTTPSRaw( fallback: fallback, transport: NewHTTPSTransportWrapper(dialer, serverAddr, fallback), } + transport.keepIdle.Store(true) + return transport } func (t *HTTPSTransport) Start(stage adapter.StartStage) error { @@ -161,6 +167,19 @@ func (t *HTTPSTransport) Reset() { t.resetTransportLocked() } +func (t *HTTPSTransport) SetKeepIdleConnections(keep bool) { + t.keepIdle.Store(keep) + if !keep { + t.CloseIdleConnections() + } +} + +func (t *HTTPSTransport) CloseIdleConnections() { + t.transportAccess.Lock() + defer t.transportAccess.Unlock() + t.transport.CloseIdleConnections() +} + func (t *HTTPSTransport) resetTransportLocked() { oldTransport := t.transport t.transport = NewHTTPSTransportWrapper(t.dialer, t.serverAddr, t.fallback) @@ -171,6 +190,9 @@ func (t *HTTPSTransport) resetTransportLocked() { func (t *HTTPSTransport) Exchange(ctx context.Context, message *mDNS.Msg) (*mDNS.Msg, error) { startAt := time.Now() response, err := t.exchange(ctx, message) + if !t.keepIdle.Load() { + t.CloseIdleConnections() + } if err != nil { if errors.Is(err, context.DeadlineExceeded) { t.transportAccess.Lock() diff --git a/dns/transport/https_transport.go b/dns/transport/https_transport.go index c4165535..73fafa56 100644 --- a/dns/transport/https_transport.go +++ b/dns/transport/https_transport.go @@ -111,6 +111,10 @@ func (h *HTTPSTransportWrapper) Close() { for _, trackedConn := range connections { trackedConn.Conn.Close() } + h.CloseIdleConnections() +} + +func (h *HTTPSTransportWrapper) CloseIdleConnections() { h.http2Transport.CloseIdleConnections() h.httpTransport.CloseIdleConnections() } diff --git a/dns/transport/multiplexer.go b/dns/transport/multiplexer.go index 6b275ae9..1b6e88c8 100644 --- a/dns/transport/multiplexer.go +++ b/dns/transport/multiplexer.go @@ -47,6 +47,7 @@ type queryMultiplexer struct { reuseState atomic.Int32 demoteFailures atomic.Int32 + keepIdle atomic.Bool probeAccess sync.Mutex probeEpoch uint32 @@ -81,7 +82,7 @@ type pendingQuery struct { } func newQueryMultiplexer(options queryMultiplexerOptions) *queryMultiplexer { - return &queryMultiplexer{ + multiplexer := &queryMultiplexer{ options: options, queries: make(map[uint16]*pendingQuery), connection: NewConnPool(ConnPoolOptions[*multiplexConn]{ @@ -94,6 +95,23 @@ func newQueryMultiplexer(options queryMultiplexerOptions) *queryMultiplexer { }, }), } + multiplexer.keepIdle.Store(true) + return multiplexer +} + +func (m *queryMultiplexer) SetKeepIdleConnections(keep bool) { + m.keepIdle.Store(keep) + m.closeIdleConnection() +} + +func (m *queryMultiplexer) CloseIdleConnections() { + m.connection.CloseIdle() +} + +func (m *queryMultiplexer) closeIdleConnection() { + if !m.keepIdle.Load() { + m.connection.CloseIdle() + } } func (m *queryMultiplexer) Close() error { @@ -307,6 +325,7 @@ func (m *queryMultiplexer) exchangeAsync(ctx context.Context, message *mDNS.Msg, queryId, err := m.register(ctx, connCtx, conn, message, callback, retryReadError && m.options.retryReadError && !created) if err != nil { m.connection.Release(conn, true) + m.closeIdleConnection() callback(nil, err) return } @@ -408,6 +427,7 @@ func (m *queryMultiplexer) complete(queryId uint16, response *mDNS.Msg, err erro response.Id = pending.message.Id } pending.callback(response, err) + m.closeIdleConnection() } func (m *queryMultiplexer) completeContextDone(queryId uint16, ctx context.Context) { @@ -422,6 +442,7 @@ func (m *queryMultiplexer) completeContextDone(queryId uint16, ctx context.Conte m.connection.Release(pending.conn, true) } pending.callback(nil, err) + m.closeIdleConnection() } func (m *queryMultiplexer) recvLoop(conn *multiplexConn) { diff --git a/dns/transport/tcp.go b/dns/transport/tcp.go index d406aeb2..a4b72724 100644 --- a/dns/transport/tcp.go +++ b/dns/transport/tcp.go @@ -22,7 +22,10 @@ import ( mDNS "github.com/miekg/dns" ) -var _ adapter.DNSTransport = (*TCPTransport)(nil) +var ( + _ adapter.DNSTransport = (*TCPTransport)(nil) + _ adapter.IdleConnectionKeeper = (*TCPTransport)(nil) +) func RegisterTCP(registry *dns.TransportRegistry) { dns.RegisterTransport[option.RemoteDNSServerOptions](registry, C.DNSTypeTCP, NewTCP) @@ -91,6 +94,14 @@ func (t *TCPTransport) Reset() { t.multiplexer.Reset() } +func (t *TCPTransport) SetKeepIdleConnections(keep bool) { + t.multiplexer.SetKeepIdleConnections(keep) +} + +func (t *TCPTransport) CloseIdleConnections() { + t.multiplexer.CloseIdleConnections() +} + func (t *TCPTransport) Exchange(ctx context.Context, message *mDNS.Msg) (*mDNS.Msg, error) { return t.multiplexer.Exchange(ctx, message) } diff --git a/dns/transport/tls.go b/dns/transport/tls.go index da1db319..36cd8dd3 100644 --- a/dns/transport/tls.go +++ b/dns/transport/tls.go @@ -20,7 +20,10 @@ import ( mDNS "github.com/miekg/dns" ) -var _ adapter.DNSTransport = (*TLSTransport)(nil) +var ( + _ adapter.DNSTransport = (*TLSTransport)(nil) + _ adapter.IdleConnectionKeeper = (*TLSTransport)(nil) +) func RegisterTLS(registry *dns.TransportRegistry) { dns.RegisterTransport[option.RemoteTLSDNSServerOptions](registry, C.DNSTypeTLS, NewTLS) @@ -97,6 +100,14 @@ func (t *TLSTransport) Reset() { t.multiplexer.Reset() } +func (t *TLSTransport) SetKeepIdleConnections(keep bool) { + t.multiplexer.SetKeepIdleConnections(keep) +} + +func (t *TLSTransport) CloseIdleConnections() { + t.multiplexer.CloseIdleConnections() +} + func (t *TLSTransport) Exchange(ctx context.Context, message *mDNS.Msg) (*mDNS.Msg, error) { return t.multiplexer.Exchange(ctx, message) } diff --git a/dns/transport/udp.go b/dns/transport/udp.go index 4d5370c4..ea6deb16 100644 --- a/dns/transport/udp.go +++ b/dns/transport/udp.go @@ -22,7 +22,10 @@ import ( mDNS "github.com/miekg/dns" ) -var _ adapter.DNSTransport = (*UDPTransport)(nil) +var ( + _ adapter.DNSTransport = (*UDPTransport)(nil) + _ adapter.IdleConnectionKeeper = (*UDPTransport)(nil) +) func RegisterUDP(registry *dns.TransportRegistry) { dns.RegisterTransport[option.RemoteDNSServerOptions](registry, C.DNSTypeUDP, NewUDP) @@ -91,6 +94,14 @@ func (t *UDPTransport) Reset() { t.multiplexer.Reset() } +func (t *UDPTransport) SetKeepIdleConnections(keep bool) { + t.multiplexer.SetKeepIdleConnections(keep) +} + +func (t *UDPTransport) CloseIdleConnections() { + t.multiplexer.CloseIdleConnections() +} + func (t *UDPTransport) Exchange(ctx context.Context, message *mDNS.Msg) (*mDNS.Msg, error) { t.updateUDPSize(message) response, err := t.multiplexer.Exchange(ctx, message) diff --git a/experimental/boxdd/server.go b/experimental/boxdd/server.go index ad9df8a2..cb195e01 100644 --- a/experimental/boxdd/server.go +++ b/experimental/boxdd/server.go @@ -223,7 +223,7 @@ func (d *Daemon) startServiceLocked(ctx context.Context, ownerUserID string, con }) d.startedService.SetOOMKillerOptions(options.OOMKillerEnabled, options.OOMKillerDisabled, uint64(options.OOMMemoryLimit)) if options.PowerReportEnabled { - err = d.powerManager.Start(libbox.PowerReportOptions(d.startedService)) + err = d.powerManager.Start(libbox.PowerReportOptions(d.startedService, nil)) if err != nil { d.logger.Warn("start power report recorder: ", err) } diff --git a/experimental/libbox/command_server.go b/experimental/libbox/command_server.go index 5f62d03d..45470b79 100644 --- a/experimental/libbox/command_server.go +++ b/experimental/libbox/command_server.go @@ -94,7 +94,7 @@ func NewCommandServer(handler CommandServerHandler, platformInterface PlatformIn OOMRecorder: oomRecorder, }) if sPowerReportEnabled { - err := powerManager.Start(PowerReportOptions(server.StartedService)) + err := powerManager.Start(PowerReportOptions(server.StartedService, platformInterface)) if err != nil { log.StdLogger().Error(E.Cause(err, "start power report recorder")) } @@ -258,14 +258,19 @@ func (s *CommandServer) NeedFindProcess() bool { func (s *CommandServer) Pause() { recorder := s.powerManager.Recorder() if recorder != nil { - recorder.RecordPlatformEvent("ne-sleep") + recorder.RecordDeviceSleep() } instance := s.StartedService.Instance() if instance == nil || instance.PauseManager() == nil { return } + instance.Box().CloseIdleConnections() instance.PauseManager().DevicePause() if C.IsIos { + // iOS calls wake within seconds of sleep while the device stays locked, so wake is + // ignored and the pause ends one minute after the last sleep instead. Go timers on + // darwin run on CLOCK_UPTIME_RAW, which does not advance while the device sleeps, + // so the minute counts awake time only and never expires inside a sleep. if s.endPauseTimer == nil { s.endPauseTimer = time.AfterFunc(time.Minute, s.endDevicePause) } else { @@ -285,7 +290,7 @@ func (s *CommandServer) endDevicePause() { func (s *CommandServer) Wake() { recorder := s.powerManager.Recorder() if recorder != nil { - recorder.RecordPlatformEvent("ne-wake") + recorder.RecordDeviceWake() } instance := s.StartedService.Instance() if instance == nil || instance.PauseManager() == nil { diff --git a/experimental/libbox/power_report.go b/experimental/libbox/power_report.go index a31afd2d..cad04136 100644 --- a/experimental/libbox/power_report.go +++ b/experimental/libbox/power_report.go @@ -13,17 +13,23 @@ import ( type powerReportMetadata struct { reportMetadata - StartedAt string `json:"startedAt"` + StartedAt string `json:"startedAt"` + IncludeAllNetworks *bool `json:"includeAllNetworks,omitempty"` } -func PowerReportOptions(startedService *daemon.StartedService) powerreport.Options { +func PowerReportOptions(startedService *daemon.StartedService, platformInterface PlatformInterface) powerreport.Options { + metadata := powerReportMetadata{ + reportMetadata: baseReportMetadata(), + StartedAt: time.Now().UTC().Format(time.RFC3339), + } + if platformInterface != nil && platformInterface.UnderNetworkExtension() { + includeAllNetworks := platformInterface.IncludeAllNetworks() + metadata.IncludeAllNetworks = &includeAllNetworks + } return powerreport.Options{ - BasePath: sWorkingPath, - Logger: log.StdLogger(), - Metadata: powerReportMetadata{ - reportMetadata: baseReportMetadata(), - StartedAt: time.Now().UTC().Format(time.RFC3339), - }, + BasePath: sWorkingPath, + Logger: log.StdLogger(), + Metadata: metadata, OwnerCallback: chownReport, LogCallback: func() []byte { return formatLogEntries(startedService.SavedLog()) diff --git a/protocol/hysteria/outbound.go b/protocol/hysteria/outbound.go index 61760d98..a5b7b7c2 100644 --- a/protocol/hysteria/outbound.go +++ b/protocol/hysteria/outbound.go @@ -13,7 +13,6 @@ import ( C "github.com/sagernet/sing-box/constant" "github.com/sagernet/sing-box/log" "github.com/sagernet/sing-box/option" - "github.com/sagernet/sing-box/protocol/tuic" "github.com/sagernet/sing-quic/hysteria" "github.com/sagernet/sing/common" "github.com/sagernet/sing/common/bufio" @@ -28,9 +27,10 @@ func RegisterOutbound(registry *outbound.Registry) { } var ( - _ adapter.Outbound = (*tuic.Outbound)(nil) - _ adapter.InterfaceUpdateListener = (*tuic.Outbound)(nil) - _ adapter.IdleConnectionCloser = (*Outbound)(nil) + _ adapter.Outbound = (*Outbound)(nil) + _ adapter.InterfaceUpdateListener = (*Outbound)(nil) + _ adapter.IdleConnectionKeeper = (*Outbound)(nil) + _ adapter.OutboundWithMultiplex = (*Outbound)(nil) ) type Outbound struct { @@ -119,6 +119,14 @@ func (h *Outbound) InterfaceUpdated(ctx context.Context) { h.client.CloseWithError(E.New("network changed")) } +func (h *Outbound) MultiplexEnabled() bool { + return true +} + +func (h *Outbound) SetKeepIdleConnections(keep bool) { + h.client.SetKeepIdleConnections(keep) +} + func (h *Outbound) CloseIdleConnections() { h.client.CloseIdleConnections() } diff --git a/protocol/hysteria2/outbound.go b/protocol/hysteria2/outbound.go index fd142e5d..398ee66b 100644 --- a/protocol/hysteria2/outbound.go +++ b/protocol/hysteria2/outbound.go @@ -16,7 +16,6 @@ import ( C "github.com/sagernet/sing-box/constant" "github.com/sagernet/sing-box/log" "github.com/sagernet/sing-box/option" - "github.com/sagernet/sing-box/protocol/tuic" qtls "github.com/sagernet/sing-quic" "github.com/sagernet/sing-quic/hysteria" "github.com/sagernet/sing-quic/hysteria2" @@ -35,9 +34,10 @@ func RegisterOutbound(registry *outbound.Registry) { } var ( - _ adapter.Outbound = (*tuic.Outbound)(nil) - _ adapter.InterfaceUpdateListener = (*tuic.Outbound)(nil) - _ adapter.IdleConnectionCloser = (*Outbound)(nil) + _ adapter.Outbound = (*Outbound)(nil) + _ adapter.InterfaceUpdateListener = (*Outbound)(nil) + _ adapter.IdleConnectionKeeper = (*Outbound)(nil) + _ adapter.OutboundWithMultiplex = (*Outbound)(nil) ) type Outbound struct { @@ -211,6 +211,14 @@ func (h *Outbound) InterfaceUpdated(ctx context.Context) { h.client.CloseWithError(E.New("network changed")) } +func (h *Outbound) MultiplexEnabled() bool { + return true +} + +func (h *Outbound) SetKeepIdleConnections(keep bool) { + h.client.SetKeepIdleConnections(keep) +} + func (h *Outbound) CloseIdleConnections() { h.client.CloseIdleConnections() } diff --git a/protocol/shadowsocks/outbound.go b/protocol/shadowsocks/outbound.go index 33532538..2ee20031 100644 --- a/protocol/shadowsocks/outbound.go +++ b/protocol/shadowsocks/outbound.go @@ -29,7 +29,7 @@ func RegisterOutbound(registry *outbound.Registry) { var ( _ adapter.OutboundWithMultiplex = (*Outbound)(nil) _ adapter.InterfaceUpdateListener = (*Outbound)(nil) - _ adapter.IdleConnectionCloser = (*Outbound)(nil) + _ adapter.IdleConnectionKeeper = (*Outbound)(nil) ) type Outbound struct { @@ -140,6 +140,12 @@ func (h *Outbound) InterfaceUpdated(ctx context.Context) { } } +func (h *Outbound) SetKeepIdleConnections(keep bool) { + if h.multiplexDialer != nil { + h.multiplexDialer.SetKeepIdleConnections(keep) + } +} + func (h *Outbound) CloseIdleConnections() { if h.multiplexDialer != nil { h.multiplexDialer.CloseIdleConnections() diff --git a/protocol/snell/outbound.go b/protocol/snell/outbound.go index 90ca6469..71ab1219 100644 --- a/protocol/snell/outbound.go +++ b/protocol/snell/outbound.go @@ -30,17 +30,20 @@ type Outbound struct { dialer N.Dialer client snellClient serverAddr M.Socksaddr + reuse bool } var ( _ adapter.InterfaceUpdateListener = (*Outbound)(nil) - _ adapter.IdleConnectionCloser = (*Outbound)(nil) + _ adapter.IdleConnectionKeeper = (*Outbound)(nil) + _ adapter.OutboundWithMultiplex = (*Outbound)(nil) ) type snellClient interface { snellprotocol.Method DialContext(ctx context.Context, destination M.Socksaddr) (net.Conn, error) Reset() + SetKeepIdleConnections(keep bool) CloseIdleConnections() Close() error } @@ -96,6 +99,7 @@ func NewOutbound(ctx context.Context, router adapter.Router, logger log.ContextL dialer: outboundDialer, client: client, serverAddr: serverAddr, + reuse: options.Reuse, } return outbound, nil } @@ -147,6 +151,14 @@ func (h *Outbound) InterfaceUpdated(ctx context.Context) { h.client.Reset() } +func (h *Outbound) MultiplexEnabled() bool { + return h.reuse +} + +func (h *Outbound) SetKeepIdleConnections(keep bool) { + h.client.SetKeepIdleConnections(keep) +} + func (h *Outbound) CloseIdleConnections() { h.client.CloseIdleConnections() } diff --git a/protocol/ssh/outbound.go b/protocol/ssh/outbound.go index 0033e92b..7d82a3e1 100644 --- a/protocol/ssh/outbound.go +++ b/protocol/ssh/outbound.go @@ -34,7 +34,8 @@ func RegisterOutbound(registry *outbound.Registry) { var ( _ adapter.InterfaceUpdateListener = (*Outbound)(nil) - _ adapter.IdleConnectionCloser = (*Outbound)(nil) + _ adapter.IdleConnectionKeeper = (*Outbound)(nil) + _ adapter.OutboundWithMultiplex = (*Outbound)(nil) ) type Outbound struct { @@ -54,7 +55,7 @@ type Outbound struct { clientConn net.Conn client *ssh.Client streams int - draining bool + closeIdle bool } func NewOutbound(ctx context.Context, router adapter.Router, logger log.ContextLogger, tag string, options option.SSHOutboundOptions) (adapter.Outbound, error) { @@ -137,7 +138,6 @@ func (s *Outbound) connect(ctx context.Context) (client *ssh.Client, err error) if s.client != nil { s.streams++ - s.draining = false return s.client, nil } @@ -197,7 +197,6 @@ func (s *Outbound) connect(ctx context.Context) (client *ssh.Client, err error) s.clientConn = conn s.client = client s.streams = 1 - s.draining = false go func() { client.Wait() @@ -217,14 +216,22 @@ func (s *Outbound) InterfaceUpdated(ctx context.Context) { common.Close(s.clientConn) } +func (s *Outbound) MultiplexEnabled() bool { + return true +} + +func (s *Outbound) SetKeepIdleConnections(keep bool) { + s.clientAccess.Lock() + s.closeIdle = !keep + s.clientAccess.Unlock() + if !keep { + s.CloseIdleConnections() + } +} + func (s *Outbound) CloseIdleConnections() { s.clientAccess.Lock() - if s.client == nil { - s.clientAccess.Unlock() - return - } - s.draining = true - if s.streams > 0 { + if s.client == nil || s.streams > 0 { s.clientAccess.Unlock() return } @@ -235,14 +242,14 @@ func (s *Outbound) CloseIdleConnections() { common.Close(clientConn) } -func (s *Outbound) releaseStream(client *ssh.Client) { +func (s *Outbound) releaseStream(client *ssh.Client, keepSession bool) { s.clientAccess.Lock() if s.client != client { s.clientAccess.Unlock() return } s.streams-- - if !s.draining || s.streams > 0 { + if !s.closeIdle || keepSession || s.streams > 0 { s.clientAccess.Unlock() return } @@ -264,10 +271,11 @@ func (s *Outbound) DialContext(ctx context.Context, network string, destination } conn, err := client.Dial(network, destination.String()) if err != nil { - s.releaseStream(client) + s.releaseStream(client, false) return nil, err } - return &chanConnWrapper{Conn: conn, onClose: func() { s.releaseStream(client) }}, nil + keepSession := adapter.KeepSessionFromContext(ctx) + return &chanConnWrapper{Conn: conn, onClose: func() { s.releaseStream(client, keepSession) }}, nil } func (s *Outbound) ListenPacket(ctx context.Context, destination M.Socksaddr) (net.PacketConn, error) { diff --git a/protocol/trojan/outbound.go b/protocol/trojan/outbound.go index 591d44a9..2273c4b5 100644 --- a/protocol/trojan/outbound.go +++ b/protocol/trojan/outbound.go @@ -29,7 +29,7 @@ func RegisterOutbound(registry *outbound.Registry) { var ( _ adapter.OutboundWithMultiplex = (*Outbound)(nil) _ adapter.InterfaceUpdateListener = (*Outbound)(nil) - _ adapter.IdleConnectionCloser = (*Outbound)(nil) + _ adapter.IdleConnectionKeeper = (*Outbound)(nil) ) type Outbound struct { @@ -114,7 +114,11 @@ func (h *Outbound) ListenPacket(ctx context.Context, destination M.Socksaddr) (n } func (h *Outbound) MultiplexEnabled() bool { - return h.multiplexDialer != nil + if h.multiplexDialer != nil { + return true + } + multiplexTransport, isMultiplexTransport := h.transport.(adapter.V2RayMultiplexClientTransport) + return isMultiplexTransport && multiplexTransport.MultiplexEnabled() } func (h *Outbound) InterfaceUpdated(ctx context.Context) { @@ -126,10 +130,20 @@ func (h *Outbound) InterfaceUpdated(ctx context.Context) { } } +func (h *Outbound) SetKeepIdleConnections(keep bool) { + transportKeeper, isTransportKeeper := h.transport.(adapter.IdleConnectionKeeper) + if isTransportKeeper { + transportKeeper.SetKeepIdleConnections(keep) + } + if h.multiplexDialer != nil { + h.multiplexDialer.SetKeepIdleConnections(keep) + } +} + func (h *Outbound) CloseIdleConnections() { - transportCloser, isTransportCloser := h.transport.(adapter.IdleConnectionCloser) - if isTransportCloser { - transportCloser.CloseIdleConnections() + transportKeeper, isTransportKeeper := h.transport.(adapter.IdleConnectionKeeper) + if isTransportKeeper { + transportKeeper.CloseIdleConnections() } if h.multiplexDialer != nil { h.multiplexDialer.CloseIdleConnections() diff --git a/protocol/tuic/outbound.go b/protocol/tuic/outbound.go index 615d41fe..db7a5658 100644 --- a/protocol/tuic/outbound.go +++ b/protocol/tuic/outbound.go @@ -32,7 +32,8 @@ func RegisterOutbound(registry *outbound.Registry) { var ( _ adapter.InterfaceUpdateListener = (*Outbound)(nil) - _ adapter.IdleConnectionCloser = (*Outbound)(nil) + _ adapter.IdleConnectionKeeper = (*Outbound)(nil) + _ adapter.OutboundWithMultiplex = (*Outbound)(nil) ) type Outbound struct { @@ -148,6 +149,14 @@ func (h *Outbound) InterfaceUpdated(ctx context.Context) { _ = h.client.CloseWithError(E.New("network changed")) } +func (h *Outbound) MultiplexEnabled() bool { + return true +} + +func (h *Outbound) SetKeepIdleConnections(keep bool) { + h.client.SetKeepIdleConnections(keep) +} + func (h *Outbound) CloseIdleConnections() { h.client.CloseIdleConnections() } diff --git a/protocol/vless/outbound.go b/protocol/vless/outbound.go index 8be11db2..f837ac04 100644 --- a/protocol/vless/outbound.go +++ b/protocol/vless/outbound.go @@ -30,7 +30,7 @@ func RegisterOutbound(registry *outbound.Registry) { var ( _ adapter.OutboundWithMultiplex = (*Outbound)(nil) _ adapter.InterfaceUpdateListener = (*Outbound)(nil) - _ adapter.IdleConnectionCloser = (*Outbound)(nil) + _ adapter.IdleConnectionKeeper = (*Outbound)(nil) ) type Outbound struct { @@ -134,7 +134,11 @@ func (h *Outbound) ListenPacket(ctx context.Context, destination M.Socksaddr) (n } func (h *Outbound) MultiplexEnabled() bool { - return h.multiplexDialer != nil + if h.multiplexDialer != nil { + return true + } + multiplexTransport, isMultiplexTransport := h.transport.(adapter.V2RayMultiplexClientTransport) + return isMultiplexTransport && multiplexTransport.MultiplexEnabled() } func (h *Outbound) InterfaceUpdated(ctx context.Context) { @@ -146,10 +150,20 @@ func (h *Outbound) InterfaceUpdated(ctx context.Context) { } } +func (h *Outbound) SetKeepIdleConnections(keep bool) { + transportKeeper, isTransportKeeper := h.transport.(adapter.IdleConnectionKeeper) + if isTransportKeeper { + transportKeeper.SetKeepIdleConnections(keep) + } + if h.multiplexDialer != nil { + h.multiplexDialer.SetKeepIdleConnections(keep) + } +} + func (h *Outbound) CloseIdleConnections() { - transportCloser, isTransportCloser := h.transport.(adapter.IdleConnectionCloser) - if isTransportCloser { - transportCloser.CloseIdleConnections() + transportKeeper, isTransportKeeper := h.transport.(adapter.IdleConnectionKeeper) + if isTransportKeeper { + transportKeeper.CloseIdleConnections() } if h.multiplexDialer != nil { h.multiplexDialer.CloseIdleConnections() diff --git a/protocol/vmess/outbound.go b/protocol/vmess/outbound.go index 4ac274b5..14597c17 100644 --- a/protocol/vmess/outbound.go +++ b/protocol/vmess/outbound.go @@ -30,7 +30,7 @@ func RegisterOutbound(registry *outbound.Registry) { var ( _ adapter.OutboundWithMultiplex = (*Outbound)(nil) _ adapter.InterfaceUpdateListener = (*Outbound)(nil) - _ adapter.IdleConnectionCloser = (*Outbound)(nil) + _ adapter.IdleConnectionKeeper = (*Outbound)(nil) ) type Outbound struct { @@ -112,7 +112,11 @@ func NewOutbound(ctx context.Context, router adapter.Router, logger log.ContextL } func (h *Outbound) MultiplexEnabled() bool { - return h.multiplexDialer != nil + if h.multiplexDialer != nil { + return true + } + multiplexTransport, isMultiplexTransport := h.transport.(adapter.V2RayMultiplexClientTransport) + return isMultiplexTransport && multiplexTransport.MultiplexEnabled() } func (h *Outbound) InterfaceUpdated(ctx context.Context) { @@ -124,10 +128,20 @@ func (h *Outbound) InterfaceUpdated(ctx context.Context) { } } +func (h *Outbound) SetKeepIdleConnections(keep bool) { + transportKeeper, isTransportKeeper := h.transport.(adapter.IdleConnectionKeeper) + if isTransportKeeper { + transportKeeper.SetKeepIdleConnections(keep) + } + if h.multiplexDialer != nil { + h.multiplexDialer.SetKeepIdleConnections(keep) + } +} + func (h *Outbound) CloseIdleConnections() { - transportCloser, isTransportCloser := h.transport.(adapter.IdleConnectionCloser) - if isTransportCloser { - transportCloser.CloseIdleConnections() + transportKeeper, isTransportKeeper := h.transport.(adapter.IdleConnectionKeeper) + if isTransportKeeper { + transportKeeper.CloseIdleConnections() } if h.multiplexDialer != nil { h.multiplexDialer.CloseIdleConnections() diff --git a/route/reference.go b/route/reference.go index e2f5ba41..708efe5b 100644 --- a/route/reference.go +++ b/route/reference.go @@ -23,7 +23,7 @@ type ReferenceManager struct { staticOutbounds []string staticTransports []string subscriber *observable.Subscriber[struct{}] - unreferencedOutbounds map[string]bool + keepIdle map[any]bool unreferencedTransports map[string]bool } @@ -201,28 +201,40 @@ func (m *ReferenceManager) update() { } } - unreferencedOutbounds := make(map[string]bool) + keepIdle := make(map[any]bool) for _, outbound := range outboundManager.Outbounds() { - tag := outbound.Tag() - if referencedOutbounds[tag] { + keeper, isKeeper := outbound.(adapter.IdleConnectionKeeper) + if !isKeeper { continue } - unreferencedOutbounds[tag] = true - closer, isCloser := outbound.(adapter.IdleConnectionCloser) - if !isCloser { - continue - } - if !m.unreferencedOutbounds[tag] { - m.logger.Debug("outbound/", outbound.Type(), "[", tag, "] is unreferenced, closing idle connections") - } - closer.CloseIdleConnections() + m.applyKeepIdle(keepIdle, idleTarget{ + value: outbound, + keeper: keeper, + kind: "outbound/", + action: "closing idle connections", + typeName: outbound.Type(), + tag: outbound.Tag(), + keep: referencedOutbounds[outbound.Tag()], + }) } - m.unreferencedOutbounds = unreferencedOutbounds - unreferencedTransports := make(map[string]bool) for _, transport := range transportManager.Transports() { tag := transport.Tag() - if referencedTransports[tag] { + referenced := referencedTransports[tag] + keeper, isKeeper := transport.(adapter.IdleConnectionKeeper) + if isKeeper { + m.applyKeepIdle(keepIdle, idleTarget{ + value: transport, + keeper: keeper, + kind: "dns/", + action: "closing idle connections", + typeName: transport.Type(), + tag: tag, + keep: referenced, + }) + continue + } + if referenced { continue } unreferencedTransports[tag] = true @@ -231,5 +243,49 @@ func (m *ReferenceManager) update() { transport.Reset() } } + m.keepIdle = keepIdle m.unreferencedTransports = unreferencedTransports } + +type idleKeeper interface { + SetKeepIdleConnections(keep bool) +} + +type idleTarget struct { + value any + keeper idleKeeper + kind string + action string + typeName string + tag string + keep bool +} + +func (m *ReferenceManager) applyKeepIdle(keepIdle map[any]bool, target idleTarget) { + keepIdle[target.value] = target.keep + previous, tracked := m.keepIdle[target.value] + if tracked && previous == target.keep { + return + } + if !target.keep { + m.logger.Debug(target.kind, target.typeName, "[", target.tag, "] is unreferenced, ", target.action) + } + target.keeper.SetKeepIdleConnections(target.keep) +} + +func (m *ReferenceManager) CloseIdleConnections() { + outboundManager := service.FromContext[adapter.OutboundManager](m.ctx) + transportManager := service.FromContext[adapter.DNSTransportManager](m.ctx) + for _, outbound := range outboundManager.Outbounds() { + keeper, isKeeper := outbound.(adapter.IdleConnectionKeeper) + if isKeeper { + keeper.CloseIdleConnections() + } + } + for _, transport := range transportManager.Transports() { + keeper, isKeeper := transport.(adapter.IdleConnectionKeeper) + if isKeeper { + keeper.CloseIdleConnections() + } + } +} diff --git a/service/powerreport/record.go b/service/powerreport/record.go index 94e37aa0..d8e555ef 100644 --- a/service/powerreport/record.go +++ b/service/powerreport/record.go @@ -45,6 +45,8 @@ type ProcessAttribution struct { type timelineRow struct { From string `json:"from"` To string `json:"to"` + MonoMS int64 `json:"monoMS"` + AwakeMS int64 `json:"awakeMS,omitempty"` CPUUserMS int64 `json:"cpuUserMS,omitempty"` CPUSystemMS int64 `json:"cpuSystemMS,omitempty"` CPUPerformanceMS int64 `json:"cpuPerformanceMS,omitempty"` @@ -74,8 +76,9 @@ type timelineRow struct { type trafficBreakdown map[TrafficKind]map[string]*trafficDelta type trafficDelta struct { - In uint64 `json:"in,omitempty"` - Out uint64 `json:"out,omitempty"` + In uint64 `json:"in,omitempty"` + Out uint64 `json:"out,omitempty"` + Dials uint64 `json:"dials,omitempty"` } type trafficKey struct { @@ -86,6 +89,7 @@ type trafficKey struct { type trafficBytes struct { inBytes uint64 outBytes uint64 + dials uint64 } type qosBreakdown struct { @@ -107,18 +111,38 @@ const ( eventTypeDeviceWake = "device-wake" eventTypeNetworkPause = "network-pause" eventTypeNetworkWake = "network-wake" + eventTypeSleep = "ne-sleep" + eventTypeWake = "ne-wake" + eventTypeService = "service" ) type eventRecord struct { - Type string `json:"t"` - At string `json:"at"` - IdleMS int64 `json:"idleMS,omitempty"` - Direction string `json:"direction,omitempty"` - Size int `json:"size,omitempty"` - NetworkType string `json:"network,omitempty"` - NetworkPath string `json:"path,omitempty"` - By *Attribution `json:"by,omitempty"` - Device *deviceState `json:"device,omitempty"` + Type string `json:"t"` + At string `json:"at"` + MonoMS int64 `json:"monoMS"` + IdleMS int64 `json:"idleMS,omitempty"` + WallIdleMS int64 `json:"wallIdleMS,omitempty"` + AfterWake bool `json:"afterWake,omitempty"` + SinceWakeMS int64 `json:"sinceWakeMS,omitempty"` + Direction string `json:"direction,omitempty"` + Size int `json:"size,omitempty"` + NetworkType string `json:"network,omitempty"` + NetworkPath string `json:"path,omitempty"` + By *Attribution `json:"by,omitempty"` + Device *deviceState `json:"device,omitempty"` + Window *wakeWindow `json:"window,omitempty"` + LogBaseMonoMS int64 `json:"logBaseMonoMS,omitempty"` +} + +type wakeWindow struct { + AwakeMS int64 `json:"awakeMS"` + CPUUserMS int64 `json:"cpuUserMS,omitempty"` + CPUSystemMS int64 `json:"cpuSystemMS,omitempty"` + EnergyNanojoules uint64 `json:"energyNJ,omitempty"` + PackageIdleWakeups uint64 `json:"packageIdleWakeups,omitempty"` + InterruptWakeups uint64 `json:"interruptWakeups,omitempty"` + DNSQueries uint64 `json:"dnsQueries,omitempty"` + ConnectionsOpened uint64 `json:"connectionsOpened,omitempty"` } type deviceState struct { diff --git a/service/powerreport/recorder.go b/service/powerreport/recorder.go index 920464aa..782fbcf1 100644 --- a/service/powerreport/recorder.go +++ b/service/powerreport/recorder.go @@ -42,6 +42,7 @@ const ( maxRowDNSDomains = 8 maxProfiles = 3 profileMinInterval = 2 * time.Minute + profileMaxDuration = 10 * time.Second ) type Options struct { @@ -70,10 +71,13 @@ type Recorder struct { fallbackInterval time.Duration baseTime time.Time - _ [64]byte - lastActivity atomic.Int64 - _ [64]byte + _ [64]byte + lastActivity atomic.Int64 + lastActivityWall atomic.Int64 + _ [64]byte + wakePending atomic.Bool + lastWakeNano atomic.Int64 lastSampleAt atomic.Int64 pendingBreak atomic.Pointer[breakRecord] notify chan struct{} @@ -102,17 +106,28 @@ type Recorder struct { deviceStateValid bool profiles []profileEntry lastProfileAt time.Time + profileFile *os.File + profileTimer *time.Timer + awake bool + wakeAt time.Time + wakeUsage systemUsage + wakeDNSQueries uint64 + wakeConnections uint64 + windowRate float64 done chan struct{} workerDone chan struct{} } type breakRecord struct { - at time.Time - idleMS int64 - direction Direction - size int - by *Attribution + at time.Time + idleMS int64 + wallIdleMS int64 + afterWake bool + sinceWakeMS int64 + direction Direction + size int + by *Attribution } type previousSample struct { @@ -212,6 +227,7 @@ func (r *Recorder) Start() error { r.lastSampleAt.Store(int64(now.Sub(r.baseTime))) r.lastFlushAt = now r.started = true + r.beginWakeWindowLocked(now) r.updateDeviceStateLocked(now) go r.worker() return nil @@ -229,6 +245,7 @@ func (r *Recorder) Close() error { <-r.workerDone now := time.Now() r.access.Lock() + r.stopProfileLocked() r.consumeBreakLocked() r.sampleLocked(now) r.flushLocked(now) @@ -260,26 +277,45 @@ func (r *Recorder) writeLog() { // and characterizes what ended an idle period; it is not accumulated. Volume totals come from // the sampled interface counters instead. func (r *Recorder) Touch(direction Direction, size int, by *Attribution) { - nowNano := int64(time.Since(r.baseTime)) + now := time.Now() + nowNano := int64(now.Sub(r.baseTime)) + if r.wakePending.Load() && r.wakePending.CompareAndSwap(true, false) { + record := r.newBreakRecord(now, nowNano, direction, size, by) + record.afterWake = true + record.sinceWakeMS = (nowNano - r.lastWakeNano.Load()) / int64(time.Millisecond) + r.pendingBreak.Store(record) + r.notifyWorker() + return + } lastNano := r.lastActivity.Load() if nowNano-lastNano < activityRefreshNano { return } - previousNano := r.lastActivity.Swap(nowNano) - if previousNano != 0 && nowNano-previousNano >= r.gateNano { - r.pendingBreak.Store(&breakRecord{ - at: time.Now(), - idleMS: (nowNano - previousNano) / int64(time.Millisecond), - direction: direction, - size: size, - by: by, - }) + record := r.newBreakRecord(now, nowNano, direction, size, by) + if record.idleMS >= r.gateNano/int64(time.Millisecond) { + r.pendingBreak.Store(record) r.notifyWorker() } else if nowNano-r.lastSampleAt.Load() >= r.sampleNano { r.notifyWorker() } } +func (r *Recorder) newBreakRecord(now time.Time, nowNano int64, direction Direction, size int, by *Attribution) *breakRecord { + previousNano := r.lastActivity.Swap(nowNano) + previousWall := r.lastActivityWall.Swap(now.UnixNano()) + record := &breakRecord{ + at: now, + direction: direction, + size: size, + by: by, + } + if previousNano != 0 { + record.idleMS = (nowNano - previousNano) / int64(time.Millisecond) + record.wallIdleMS = (now.UnixNano() - previousWall) / int64(time.Millisecond) + } + return record +} + func (r *Recorder) CountDNSQuery(domain string) { r.dnsQueries.Add(1) if domain == "" { @@ -296,13 +332,13 @@ func (r *Recorder) CountDNSQuery(domain string) { func (r *Recorder) RecordPauseEvent(event int) { switch event { case pause.EventDevicePaused: - r.RecordPlatformEvent(eventTypeDevicePause) + r.recordPlatformEvent(eventTypeDevicePause) case pause.EventDeviceWake: - r.RecordPlatformEvent(eventTypeDeviceWake) + r.recordPlatformEvent(eventTypeDeviceWake) case pause.EventNetworkPause: - r.RecordPlatformEvent(eventTypeNetworkPause) + r.recordPlatformEvent(eventTypeNetworkPause) case pause.EventNetworkWake: - r.RecordPlatformEvent(eventTypeNetworkWake) + r.recordPlatformEvent(eventTypeNetworkWake) } } @@ -310,17 +346,98 @@ func (r *Recorder) CountConnectionOpened() { r.connectionsOpened.Add(1) } -func (r *Recorder) RecordPlatformEvent(eventType string) { +// The service log prefixes each line with whole monotonic seconds since createdAt, so a +// line at [N] sits at logBaseMonoMS + N*1000 on the monoMS axis of events and timeline rows. +func (r *Recorder) RecordServiceStart(createdAt time.Time) { now := time.Now() r.access.Lock() if !r.started || r.closed { r.access.Unlock() return } - r.events = append(r.events, eventRecord{ - Type: eventType, - At: now.UTC().Format(time.RFC3339), - }) + event := r.newEvent(eventTypeService, now) + event.LogBaseMonoMS = int64(createdAt.Sub(r.baseTime) / time.Millisecond) + r.events = append(r.events, event) + r.access.Unlock() + r.notifyWorker() +} + +func (r *Recorder) RecordDeviceSleep() { + now := time.Now() + r.access.Lock() + if !r.started || r.closed { + r.access.Unlock() + return + } + r.wakePending.Store(false) + r.stopProfileLocked() + event := r.newEvent(eventTypeSleep, now) + if r.awake { + event.Window = r.endWakeWindowLocked(now) + } + r.events = append(r.events, event) + r.access.Unlock() + r.notifyWorker() +} + +func (r *Recorder) RecordDeviceWake() { + now := time.Now() + r.access.Lock() + if !r.started || r.closed { + r.access.Unlock() + return + } + r.events = append(r.events, r.newEvent(eventTypeWake, now)) + var request *profileRequest + if !r.awake { + r.beginWakeWindowLocked(now) + r.lastWakeNano.Store(int64(now.Sub(r.baseTime))) + r.wakePending.Store(true) + request = r.requestProfileLocked(now, r.windowRate) + r.windowRate = 0 + } + r.access.Unlock() + if request != nil { + r.startProfile(request) + } + r.notifyWorker() +} + +func (r *Recorder) beginWakeWindowLocked(now time.Time) { + r.awake = true + r.wakeAt = now + r.wakeUsage = readSystemUsage() + r.wakeDNSQueries = r.dnsQueries.Load() + r.wakeConnections = r.connectionsOpened.Load() +} + +func (r *Recorder) endWakeWindowLocked(now time.Time) *wakeWindow { + r.awake = false + usage := readSystemUsage() + window := &wakeWindow{ + AwakeMS: int64(now.Sub(r.wakeAt) / time.Millisecond), + DNSQueries: r.dnsQueries.Load() - r.wakeDNSQueries, + ConnectionsOpened: r.connectionsOpened.Load() - r.wakeConnections, + } + if usage.valid && r.wakeUsage.valid { + window.CPUUserMS = (usage.userTime - r.wakeUsage.userTime) / int64(time.Millisecond) + window.CPUSystemMS = (usage.systemTime - r.wakeUsage.systemTime) / int64(time.Millisecond) + window.EnergyNanojoules = usage.energyNanojoules - r.wakeUsage.energyNanojoules + window.PackageIdleWakeups = usage.packageIdleWakeups - r.wakeUsage.packageIdleWakeups + window.InterruptWakeups = usage.interruptWakeups - r.wakeUsage.interruptWakeups + } + r.windowRate = profileRate(window.EnergyNanojoules, window.CPUUserMS+window.CPUSystemMS, now.Sub(r.wakeAt)) + return window +} + +func (r *Recorder) recordPlatformEvent(eventType string) { + now := time.Now() + r.access.Lock() + if !r.started || r.closed { + r.access.Unlock() + return + } + r.events = append(r.events, r.newEvent(eventType, now)) r.access.Unlock() r.notifyWorker() } @@ -333,11 +450,9 @@ func (r *Recorder) UpdateNetworkType(networkType string) { return } r.networkType = networkType - r.events = append(r.events, eventRecord{ - Type: eventTypeNetwork, - At: now.UTC().Format(time.RFC3339), - NetworkType: networkType, - }) + event := r.newEvent(eventTypeNetwork, now) + event.NetworkType = networkType + r.events = append(r.events, event) r.access.Unlock() r.notifyWorker() } @@ -351,15 +466,21 @@ func (r *Recorder) UpdateNetworkPath(description string) { return } r.networkPath = description - r.events = append(r.events, eventRecord{ - Type: eventTypePath, - At: now.UTC().Format(time.RFC3339), - NetworkPath: description, - }) + event := r.newEvent(eventTypePath, now) + event.NetworkPath = description + r.events = append(r.events, event) r.access.Unlock() r.notifyWorker() } +func (r *Recorder) newEvent(eventType string, now time.Time) eventRecord { + return eventRecord{ + Type: eventType, + At: now.UTC().Format(time.RFC3339), + MonoMS: int64(now.Sub(r.baseTime) / time.Millisecond), + } +} + func (r *Recorder) notifyWorker() { select { case r.notify <- struct{}{}: @@ -402,7 +523,7 @@ func (r *Recorder) process() { } r.access.Unlock() if request != nil { - r.writeSampleProfile(request) + r.startProfile(request) } } @@ -411,15 +532,16 @@ func (r *Recorder) consumeBreakLocked() { if record == nil { return } - r.events = append(r.events, eventRecord{ - Type: eventTypeBreak, - At: record.at.UTC().Format(time.RFC3339), - IdleMS: record.idleMS, - Direction: record.direction.String(), - Size: record.size, - NetworkType: r.networkType, - By: record.by, - }) + event := r.newEvent(eventTypeBreak, record.at) + event.IdleMS = record.idleMS + event.WallIdleMS = record.wallIdleMS + event.AfterWake = record.afterWake + event.SinceWakeMS = record.sinceWakeMS + event.Direction = record.direction.String() + event.Size = record.size + event.NetworkType = r.networkType + event.By = record.by + r.events = append(r.events, event) } func (r *Recorder) resetPreviousLocked(now time.Time) { @@ -445,6 +567,8 @@ func (r *Recorder) sampleLocked(now time.Time) *profileRequest { row := timelineRow{ From: previous.at.UTC().Format(time.RFC3339), To: now.UTC().Format(time.RFC3339), + MonoMS: int64(now.Sub(r.baseTime) / time.Millisecond), + AwakeMS: int64(now.Sub(previous.at) / time.Millisecond), CPUGCMS: int64((current.gcSeconds - previous.gcSeconds) * 1000), Goroutines: r.metricsSamples[1].Value.Uint64(), GCCycles: current.gcCycles - previous.gcCycles, @@ -485,7 +609,7 @@ func (r *Recorder) sampleLocked(now time.Time) *profileRequest { } if current.absoluteTime != 0 && previous.absoluteTime != 0 && current.absoluteTime >= previous.absoluteTime { sleptNano := (current.continuousTime - previous.continuousTime) - (current.absoluteTime - previous.absoluteTime) - wallNano := now.Sub(previous.at).Nanoseconds() + wallNano := now.UnixNano() - previous.at.UnixNano() if sleptNano > wallNano { sleptNano = wallNano } @@ -519,7 +643,7 @@ func (r *Recorder) sampleLocked(now time.Time) *profileRequest { } r.updateDeviceStateLocked(now) r.rows = append(r.rows, row) - return r.requestProfileLocked(now, &row, now.Sub(previous.at)) + return r.requestProfileLocked(now, profileRate(row.EnergyNanojoules, row.CPUUserMS+row.CPUSystemMS, now.Sub(previous.at))) } func (r *Recorder) collectDNSDomains() map[string]uint64 { @@ -555,8 +679,9 @@ func trafficDeltas(current map[trafficKey]trafficBytes, previous map[trafficKey] for key, currentBytes := range current { previousBytes := previous[key] delta := trafficDelta{ - In: currentBytes.inBytes - previousBytes.inBytes, - Out: currentBytes.outBytes - previousBytes.outBytes, + In: currentBytes.inBytes - previousBytes.inBytes, + Out: currentBytes.outBytes - previousBytes.outBytes, + Dials: currentBytes.dials - previousBytes.dials, } if delta == (trafficDelta{}) { continue @@ -584,25 +709,26 @@ func (r *Recorder) updateDeviceStateLocked(now time.Time) { } r.lastDeviceState = state r.deviceStateValid = true - r.events = append(r.events, eventRecord{ - Type: eventTypeDevice, - At: now.UTC().Format(time.RFC3339), - Device: &state, - }) + event := r.newEvent(eventTypeDevice, now) + event.Device = &state + r.events = append(r.events, event) } -func (r *Recorder) requestProfileLocked(now time.Time, row *timelineRow, duration time.Duration) *profileRequest { - if r.closed || duration <= 0 { +func profileRate(energyNanojoules uint64, cpuMS int64, duration time.Duration) float64 { + if duration <= 0 { + return 0 + } + work := float64(energyNanojoules) + if work == 0 { + work = float64(cpuMS) + } + return work / duration.Seconds() +} + +func (r *Recorder) requestProfileLocked(now time.Time, rate float64) *profileRequest { + if r.closed || rate == 0 || r.profileFile != nil { return nil } - work := float64(row.EnergyNanojoules) - if work == 0 { - work = float64(row.CPUUserMS + row.CPUSystemMS) - } - if work == 0 { - return nil - } - rate := work / duration.Seconds() lowestIndex := -1 if len(r.profiles) >= maxProfiles { lowestIndex = 0 @@ -619,7 +745,7 @@ func (r *Recorder) requestProfileLocked(now time.Time, row *timelineRow, duratio return nil } request := &profileRequest{ - writePath: filepath.Join(r.draftPath, "goroutine-"+now.UTC().Format("150405")+".pb"), + writePath: filepath.Join(r.draftPath, "cpu-"+now.UTC().Format("150405")+".pb"), } if lowestIndex >= 0 { request.removePath = r.profiles[lowestIndex].path @@ -630,7 +756,7 @@ func (r *Recorder) requestProfileLocked(now time.Time, row *timelineRow, duratio return request } -func (r *Recorder) writeSampleProfile(request *profileRequest) { +func (r *Recorder) startProfile(request *profileRequest) { if request.removePath != "" { os.Remove(request.removePath) } @@ -638,9 +764,38 @@ func (r *Recorder) writeSampleProfile(request *profileRequest) { if err != nil { return } - defer file.Close() r.chown(request.writePath) - pprof.Lookup("goroutine").WriteTo(file, 0) + r.access.Lock() + defer r.access.Unlock() + if r.closed || r.profileFile != nil { + file.Close() + os.Remove(request.writePath) + return + } + err = pprof.StartCPUProfile(file) + if err != nil { + file.Close() + os.Remove(request.writePath) + return + } + r.profileFile = file + r.profileTimer = time.AfterFunc(profileMaxDuration, r.stopProfile) +} + +func (r *Recorder) stopProfile() { + r.access.Lock() + defer r.access.Unlock() + r.stopProfileLocked() +} + +func (r *Recorder) stopProfileLocked() { + if r.profileFile == nil { + return + } + r.profileTimer.Stop() + pprof.StopCPUProfile() + r.profileFile.Close() + r.profileFile = nil } func (r *Recorder) chown(path string) { diff --git a/service/powerreport/traffic.go b/service/powerreport/traffic.go index 41c0f3ea..5b530403 100644 --- a/service/powerreport/traffic.go +++ b/service/powerreport/traffic.go @@ -15,6 +15,14 @@ const maxTrafficCounters = 512 type TrafficCounter struct { inBytes atomic.Uint64 outBytes atomic.Uint64 + dials atomic.Uint64 +} + +func (c *TrafficCounter) CountDial() { + if c == nil { + return + } + c.dials.Add(1) } func (c *TrafficCounter) CountIn(n int64) { @@ -67,6 +75,7 @@ func (r *Recorder) snapshotTrafficLocked() map[trafficKey]trafficBytes { snapshot[key] = trafficBytes{ inBytes: counter.inBytes.Load(), outBytes: counter.outBytes.Load(), + dials: counter.dials.Load(), } } return snapshot diff --git a/transport/v2raygrpclite/client.go b/transport/v2raygrpclite/client.go index d8b3400e..62163e44 100644 --- a/transport/v2raygrpclite/client.go +++ b/transport/v2raygrpclite/client.go @@ -23,7 +23,7 @@ import ( "golang.org/x/net/http2" ) -var _ adapter.V2RayClientTransport = (*Client)(nil) +var _ adapter.V2RayMultiplexClientTransport = (*Client)(nil) var defaultClientHeader = http.Header{ "Content-Type": []string{"application/grpc"}, @@ -38,6 +38,7 @@ type Client struct { options option.V2RayGRPCOptions url *url.URL host string + closeIdle atomic.Bool } func NewClient(ctx context.Context, dialer N.Dialer, serverAddr M.Socksaddr, options option.V2RayGRPCOptions, tlsConfig tls.Config) adapter.V2RayClientTransport { @@ -92,6 +93,12 @@ func (c *Client) DialContext(ctx context.Context) (net.Conn, error) { Host: c.host, } conn := newLateGunConn(pipeInWriter, cancel) + keepSession := adapter.KeepSessionFromContext(ctx) + conn.onClose = func() { + if c.closeIdle.Load() && !keepSession { + c.transport.CloseIdleConnections() + } + } handshakeTimeout := C.TCPTimeout if deadline, hasDeadline := ctx.Deadline(); hasDeadline { handshakeTimeout = time.Until(deadline) @@ -126,6 +133,17 @@ func (c *Client) DialContext(ctx context.Context) (net.Conn, error) { return conn, nil } +func (c *Client) MultiplexEnabled() bool { + return true +} + +func (c *Client) SetKeepIdleConnections(keep bool) { + c.closeIdle.Store(!keep) + if !keep { + c.CloseIdleConnections() + } +} + func (c *Client) CloseIdleConnections() { c.transport.CloseIdleConnections() } diff --git a/transport/v2raygrpclite/conn.go b/transport/v2raygrpclite/conn.go index 896183b9..52911e62 100644 --- a/transport/v2raygrpclite/conn.go +++ b/transport/v2raygrpclite/conn.go @@ -30,6 +30,7 @@ type GunConn struct { err error cancel context.CancelFunc readRemaining int + onClose func() } func newGunConn(reader io.Reader, writer io.Writer, flusher http.Flusher) *GunConn { @@ -158,6 +159,9 @@ func (c *GunConn) Close() error { if c.cancel != nil { c.cancel() } + if c.onClose != nil { + c.onClose() + } return err } diff --git a/transport/v2rayhttp/client.go b/transport/v2rayhttp/client.go index 0b64c09f..9d21d91f 100644 --- a/transport/v2rayhttp/client.go +++ b/transport/v2rayhttp/client.go @@ -25,7 +25,7 @@ import ( "golang.org/x/net/http2" ) -var _ adapter.V2RayClientTransport = (*Client)(nil) +var _ adapter.V2RayMultiplexClientTransport = (*Client)(nil) type Client struct { ctx context.Context @@ -37,6 +37,7 @@ type Client struct { host []string method string headers http.Header + closeIdle atomic.Bool } func NewClient(ctx context.Context, dialer N.Dialer, serverAddr M.Socksaddr, options option.V2RayHTTPOptions, tlsConfig tls.Config) (adapter.V2RayClientTransport, error) { @@ -143,6 +144,12 @@ func (c *Client) dialHTTP2(ctx context.Context) (net.Conn, error) { request.Host = c.host[rand.Intn(hostLen)] } conn := NewLateHTTPConn(pipeInWriter, cancel) + keepSession := adapter.KeepSessionFromContext(ctx) + conn.onClose = func() { + if c.closeIdle.Load() && !keepSession { + CloseIdleConnections(c.transport.Load()) + } + } handshakeTimeout := C.TCPTimeout if deadline, hasDeadline := ctx.Deadline(); hasDeadline { handshakeTimeout = time.Until(deadline) @@ -170,8 +177,19 @@ func (c *Client) dialHTTP2(ctx context.Context) (net.Conn, error) { return conn, nil } +func (c *Client) MultiplexEnabled() bool { + return c.http2 +} + +func (c *Client) SetKeepIdleConnections(keep bool) { + c.closeIdle.Store(!keep) + if !keep { + c.CloseIdleConnections() + } +} + func (c *Client) CloseIdleConnections() { - CloseIdleConnections(c.transport) + CloseIdleConnections(c.transport.Load()) } func (c *Client) Close() error { diff --git a/transport/v2rayhttp/conn.go b/transport/v2rayhttp/conn.go index 758eda7f..80689aea 100644 --- a/transport/v2rayhttp/conn.go +++ b/transport/v2rayhttp/conn.go @@ -130,11 +130,12 @@ func (c *HTTPConn) Upstream() any { } type HTTP2Conn struct { - reader io.Reader - writer io.Writer - create chan struct{} - err error - cancel context.CancelFunc + reader io.Reader + writer io.Writer + create chan struct{} + err error + cancel context.CancelFunc + onClose func() } func NewHTTPConn(reader io.Reader, writer io.Writer) HTTP2Conn { @@ -189,6 +190,9 @@ func (c *HTTP2Conn) Close() error { if c.cancel != nil { c.cancel() } + if c.onClose != nil { + c.onClose() + } return err } diff --git a/transport/v2rayquic/client.go b/transport/v2rayquic/client.go index 105785d5..f780fb5e 100644 --- a/transport/v2rayquic/client.go +++ b/transport/v2rayquic/client.go @@ -6,6 +6,7 @@ import ( "context" "net" "sync" + "sync/atomic" "github.com/sagernet/quic-go" "github.com/sagernet/quic-go/http3" @@ -19,7 +20,7 @@ import ( N "github.com/sagernet/sing/common/network" ) -var _ adapter.V2RayClientTransport = (*Client)(nil) +var _ adapter.V2RayMultiplexClientTransport = (*Client)(nil) type Client struct { ctx context.Context @@ -30,14 +31,15 @@ type Client struct { connAccess sync.Mutex conn common.TypedValue[*clientConnection] rawConn net.Conn + closeIdle atomic.Bool } type clientConnection struct { *quic.Conn - access sync.Mutex - streams int - draining bool - closed bool + access sync.Mutex + streams int + closed bool + closeIdle *atomic.Bool } func (c *clientConnection) active() bool { @@ -54,14 +56,13 @@ func (c *clientConnection) acquireStream() bool { return false } c.streams++ - c.draining = false return true } -func (c *clientConnection) releaseStream() { +func (c *clientConnection) releaseStream(keepSession bool) { c.access.Lock() c.streams-- - closed := c.draining && c.streams == 0 && c.markClosedLocked() + closed := c.closeIdle.Load() && !keepSession && c.streams == 0 && c.markClosedLocked() c.access.Unlock() if closed { c.CloseWithError(0, "") @@ -126,7 +127,7 @@ func (c *Client) offerNew() (*clientConnection, error) { <-quicConn.Context().Done() udpConn.Close() }() - conn := &clientConnection{Conn: quicConn} + conn := &clientConnection{Conn: quicConn, closeIdle: &c.closeIdle} c.conn.Store(conn) c.rawConn = udpConn return conn, nil @@ -143,10 +144,22 @@ func (c *Client) DialContext(ctx context.Context) (net.Conn, error) { } stream, err := conn.OpenStream() if err != nil { - conn.releaseStream() + conn.releaseStream(false) return nil, err } - return &StreamWrapper{Conn: conn.Conn, Stream: stream, onClose: conn.releaseStream}, nil + keepSession := adapter.KeepSessionFromContext(ctx) + return &StreamWrapper{Conn: conn.Conn, Stream: stream, onClose: func() { conn.releaseStream(keepSession) }}, nil + } +} + +func (c *Client) MultiplexEnabled() bool { + return true +} + +func (c *Client) SetKeepIdleConnections(keep bool) { + c.closeIdle.Store(!keep) + if !keep { + c.CloseIdleConnections() } } @@ -156,7 +169,6 @@ func (c *Client) CloseIdleConnections() { return } conn.access.Lock() - conn.draining = true closed := conn.streams == 0 && conn.markClosedLocked() conn.access.Unlock() if closed {