Improve idle connection management

This commit is contained in:
世界
2026-10-02 18:00:09 +08:00
parent eb7deeae8a
commit 07130365d1
37 changed files with 766 additions and 181 deletions
+2 -1
View File
@@ -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()
}
+14
View File
@@ -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
}
+5
View File
@@ -22,3 +22,8 @@ type V2RayClientTransport interface {
DialContext(ctx context.Context) (net.Conn, error)
Close() error
}
type V2RayMultiplexClientTransport interface {
V2RayClientTransport
MultiplexEnabled() bool
}
+12 -1
View File
@@ -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
}
+4
View File
@@ -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)
+13
View File
@@ -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)
}
+9
View File
@@ -0,0 +1,9 @@
//go:build !with_quic
package urltest
import "context"
func contextWithQUICKeepSession(ctx context.Context) context.Context {
return ctx
}
+7 -1
View File
@@ -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
}
+10 -3
View File
@@ -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) {
+31
View File
@@ -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
+24 -2
View File
@@ -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()
+4
View File
@@ -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()
}
+22 -1
View File
@@ -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) {
+12 -1
View File
@@ -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)
}
+12 -1
View File
@@ -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)
}
+12 -1
View File
@@ -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)
+1 -1
View File
@@ -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)
}
+8 -3
View File
@@ -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 {
+14 -8
View File
@@ -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())
+12 -4
View File
@@ -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()
}
+12 -4
View File
@@ -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()
}
+7 -1
View File
@@ -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()
+13 -1
View File
@@ -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()
}
+22 -14
View File
@@ -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) {
+19 -5
View File
@@ -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()
+10 -1
View File
@@ -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()
}
+19 -5
View File
@@ -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()
+19 -5
View File
@@ -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()
+72 -16
View File
@@ -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()
}
}
}
+35 -11
View File
@@ -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 {
+225 -70
View File
@@ -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) {
+9
View File
@@ -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
+19 -1
View File
@@ -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()
}
+4
View File
@@ -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
}
+20 -2
View File
@@ -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 {
+9 -5
View File
@@ -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
}
+24 -12
View File
@@ -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 {