Skip to content

Commit c84d634

Browse files
nekohasekainekohasekai
committed
Add logger parameter to stream and packet reactors
1 parent 99b7f22 commit c84d634

4 files changed

Lines changed: 69 additions & 37 deletions

File tree

‎common/bufio/packet_reactor.go‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import (
99
"github.com/sagernet/sing/common"
1010
"github.com/sagernet/sing/common/buf"
1111
E "github.com/sagernet/sing/common/exceptions"
12+
"github.com/sagernet/sing/common/logger"
1213
M "github.com/sagernet/sing/common/metadata"
1314
N "github.com/sagernet/sing/common/network"
1415
)
@@ -69,16 +70,21 @@ func CreatePacketPollable(reader N.PacketReader) (N.PacketPollable, bool) {
6970
type PacketReactor struct {
7071
ctx context.Context
7172
cancel context.CancelFunc
73+
logger logger.Logger
7274
fdPoller *FDPoller
7375
fdPollerOnce sync.Once
7476
fdPollerErr error
7577
}
7678

77-
func NewPacketReactor(ctx context.Context) *PacketReactor {
79+
func NewPacketReactor(ctx context.Context, l logger.Logger) *PacketReactor {
7880
ctx, cancel := context.WithCancel(ctx)
81+
if l == nil {
82+
l = logger.NOP()
83+
}
7984
return &PacketReactor{
8085
ctx: ctx,
8186
cancel: cancel,
87+
logger: l,
8288
}
8389
}
8490

@@ -129,6 +135,7 @@ type reactorStream struct {
129135
}
130136

131137
func (r *PacketReactor) Copy(ctx context.Context, source N.PacketConn, destination N.PacketConn, onClose N.CloseHandlerFunc) {
138+
r.logger.Trace("packet copy: starting")
132139
ctx, cancel := context.WithCancel(ctx)
133140
conn := &reactorConnection{
134141
ctx: ctx,
@@ -216,6 +223,7 @@ func (r *PacketReactor) prepareStream(conn *reactorConnection, source N.PacketRe
216223

217224
func (r *PacketReactor) registerStream(stream *reactorStream) {
218225
if stream.pushable != nil {
226+
r.logger.Trace("packet stream: using pushable mode")
219227
stream.pushable.SetOnDataReady(func() {
220228
go stream.runActiveLoop(nil)
221229
})
@@ -226,18 +234,23 @@ func (r *PacketReactor) registerStream(stream *reactorStream) {
226234
}
227235

228236
if stream.pollable == nil {
237+
r.logger.Trace("packet stream: using legacy copy")
229238
go stream.runLegacyCopy()
230239
return
231240
}
232241

233242
fdPoller, err := r.getFDPoller()
234243
if err != nil {
244+
r.logger.Trace("packet stream: FD poller unavailable, using legacy copy")
235245
go stream.runLegacyCopy()
236246
return
237247
}
238248
err = fdPoller.Add(stream, stream.pollable.FD())
239249
if err != nil {
250+
r.logger.Trace("packet stream: failed to add to FD poller, using legacy copy")
240251
go stream.runLegacyCopy()
252+
} else {
253+
r.logger.Trace("packet stream: registered with FD poller")
241254
}
242255
}
243256

@@ -311,6 +324,7 @@ func (s *reactorStream) runActiveLoop(firstPacket *N.PacketBuffer) {
311324
if !s.state.CompareAndSwap(stateActive, stateIdle) {
312325
return
313326
}
327+
s.connection.reactor.logger.Trace("packet stream: timeout, returning to idle pool")
314328
if s.pushable != nil {
315329
if s.pushable.HasPendingData() {
316330
if s.state.CompareAndSwap(stateIdle, stateActive) {
@@ -325,6 +339,7 @@ func (s *reactorStream) runActiveLoop(firstPacket *N.PacketBuffer) {
325339
if !notFirstTime {
326340
err = N.ReportHandshakeFailure(s.originSource, err)
327341
}
342+
s.connection.reactor.logger.Trace("packet stream: error occurred: ", err)
328343
s.closeWithError(err)
329344
return
330345
}
@@ -399,6 +414,7 @@ func (s *reactorStream) closeWithError(err error) {
399414
func (c *reactorConnection) closeWithError(err error) {
400415
c.closeOnce.Do(func() {
401416
defer close(c.done)
417+
c.reactor.logger.Trace("packet connection: closing with error: ", err)
402418

403419
if c.stopReactorWatch != nil {
404420
c.stopReactorWatch()

‎common/bufio/packet_reactor_test.go‎

Lines changed: 23 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -274,7 +274,7 @@ func TestBatchCopy_Pipe_DataIntegrity(t *testing.T) {
274274
defer pipeA.Close()
275275
defer pipeB.Close()
276276

277-
copier := NewPacketReactor(context.Background())
277+
copier := NewPacketReactor(context.Background(), nil)
278278
defer copier.Close()
279279

280280
go func() {
@@ -335,7 +335,7 @@ func TestBatchCopy_Pipe_Bidirectional(t *testing.T) {
335335
defer pipeA.Close()
336336
defer pipeB.Close()
337337

338-
copier := NewPacketReactor(context.Background())
338+
copier := NewPacketReactor(context.Background(), nil)
339339
defer copier.Close()
340340

341341
go func() {
@@ -447,7 +447,7 @@ func TestBatchCopy_FDPoller_DataIntegrity(t *testing.T) {
447447
proxyA := newFDPacketConn(t, proxyAConn, serverAddr)
448448
proxyB := newFDPacketConn(t, proxyBConn, clientAddr)
449449

450-
copier := NewPacketReactor(context.Background())
450+
copier := NewPacketReactor(context.Background(), nil)
451451
defer copier.Close()
452452

453453
go func() {
@@ -568,7 +568,7 @@ func TestBatchCopy_LegacyChannel_DataIntegrity(t *testing.T) {
568568
proxyA := newChannelPacketConn(proxyAConn, serverAddr)
569569
proxyB := newChannelPacketConn(proxyBConn, clientAddr)
570570

571-
copier := NewPacketReactor(context.Background())
571+
copier := NewPacketReactor(context.Background(), nil)
572572
defer copier.Close()
573573

574574
go func() {
@@ -689,7 +689,7 @@ func TestBatchCopy_MixedMode_DataIntegrity(t *testing.T) {
689689
proxyA := newFDPacketConn(t, proxyAConn, serverAddr)
690690
proxyB := newChannelPacketConn(proxyBConn, clientAddr)
691691

692-
copier := NewPacketReactor(context.Background())
692+
copier := NewPacketReactor(context.Background(), nil)
693693
defer copier.Close()
694694

695695
go func() {
@@ -790,7 +790,7 @@ func TestBatchCopy_MultipleConnections_DataIntegrity(t *testing.T) {
790790

791791
const numConnections = 5
792792

793-
copier := NewPacketReactor(context.Background())
793+
copier := NewPacketReactor(context.Background(), nil)
794794
defer copier.Close()
795795

796796
var wg sync.WaitGroup
@@ -881,7 +881,7 @@ func TestBatchCopy_TimeoutAndResume_DataIntegrity(t *testing.T) {
881881
defer pipeA.Close()
882882
defer pipeB.Close()
883883

884-
copier := NewPacketReactor(context.Background())
884+
copier := NewPacketReactor(context.Background(), nil)
885885
defer copier.Close()
886886

887887
go func() {
@@ -945,7 +945,7 @@ func TestBatchCopy_CloseWhileTransferring(t *testing.T) {
945945
pipeA := newTestPacketPipe(addr1)
946946
pipeB := newTestPacketPipe(addr2)
947947

948-
copier := NewPacketReactor(context.Background())
948+
copier := NewPacketReactor(context.Background(), nil)
949949

950950
copyDone := make(chan struct{})
951951
go func() {
@@ -995,7 +995,7 @@ func TestBatchCopy_HighThroughput(t *testing.T) {
995995
defer pipeA.Close()
996996
defer pipeB.Close()
997997

998-
copier := NewPacketReactor(context.Background())
998+
copier := NewPacketReactor(context.Background(), nil)
999999
defer copier.Close()
10001000

10011001
go func() {
@@ -1080,7 +1080,7 @@ func TestBatchCopy_LegacyFallback_DataIntegrity(t *testing.T) {
10801080
proxyA := &legacyPacketConn{NetPacketConn: NewPacketConn(proxyAConn), targetAddr: serverAddr}
10811081
proxyB := &legacyPacketConn{NetPacketConn: NewPacketConn(proxyBConn), targetAddr: clientAddr}
10821082

1083-
copier := NewPacketReactor(context.Background())
1083+
copier := NewPacketReactor(context.Background(), nil)
10841084
defer copier.Close()
10851085

10861086
go func() {
@@ -1195,7 +1195,7 @@ func TestBatchCopy_ReactorClose(t *testing.T) {
11951195
defer pipeA.Close()
11961196
defer pipeB.Close()
11971197

1198-
copier := NewPacketReactor(context.Background())
1198+
copier := NewPacketReactor(context.Background(), nil)
11991199

12001200
copyDone := make(chan struct{})
12011201
go func() {
@@ -1243,7 +1243,7 @@ func TestBatchCopy_SmallPackets(t *testing.T) {
12431243
defer pipeA.Close()
12441244
defer pipeB.Close()
12451245

1246-
copier := NewPacketReactor(context.Background())
1246+
copier := NewPacketReactor(context.Background(), nil)
12471247
defer copier.Close()
12481248

12491249
go func() {
@@ -1297,7 +1297,7 @@ func TestBatchCopy_VaryingPacketSizes(t *testing.T) {
12971297
defer pipeA.Close()
12981298
defer pipeB.Close()
12991299

1300-
copier := NewPacketReactor(context.Background())
1300+
copier := NewPacketReactor(context.Background(), nil)
13011301
defer copier.Close()
13021302

13031303
go func() {
@@ -1365,7 +1365,7 @@ func TestBatchCopy_OnCloseCallback(t *testing.T) {
13651365
pipeA := newTestPacketPipe(addr1)
13661366
pipeB := newTestPacketPipe(addr2)
13671367

1368-
copier := NewPacketReactor(context.Background())
1368+
copier := NewPacketReactor(context.Background(), nil)
13691369
defer copier.Close()
13701370

13711371
callbackCalled := make(chan error, 1)
@@ -1409,7 +1409,7 @@ func TestBatchCopy_SourceClose(t *testing.T) {
14091409
pipeA := newTestPacketPipe(addr1)
14101410
pipeB := newTestPacketPipe(addr2)
14111411

1412-
copier := NewPacketReactor(context.Background())
1412+
copier := NewPacketReactor(context.Background(), nil)
14131413
defer copier.Close()
14141414

14151415
var capturedErr error
@@ -1667,7 +1667,7 @@ func TestPacketReactor_Pushable_Basic(t *testing.T) {
16671667
defer pipeA.Close()
16681668
defer pipeB.Close()
16691669

1670-
copier := NewPacketReactor(context.Background())
1670+
copier := NewPacketReactor(context.Background(), nil)
16711671
defer copier.Close()
16721672

16731673
copyDone := make(chan struct{})
@@ -1750,7 +1750,7 @@ func TestPacketReactor_Pushable_HasPendingData(t *testing.T) {
17501750
pipeA.send(data, addr2)
17511751
}
17521752

1753-
copier := NewPacketReactor(context.Background())
1753+
copier := NewPacketReactor(context.Background(), nil)
17541754
defer copier.Close()
17551755

17561756
copyDone := make(chan struct{})
@@ -1799,7 +1799,7 @@ func TestPacketReactor_Pushable_TimeoutResume(t *testing.T) {
17991799
defer pipeA.Close()
18001800
defer pipeB.Close()
18011801

1802-
copier := NewPacketReactor(context.Background())
1802+
copier := NewPacketReactor(context.Background(), nil)
18031803
defer copier.Close()
18041804

18051805
copyDone := make(chan struct{})
@@ -1874,7 +1874,7 @@ func TestPacketReactor_WriteError(t *testing.T) {
18741874
// Wrap destination with failing writer that fails after 3 packets
18751875
failingDest := &failingPacketWriter{PacketConn: pipeB, failAfter: 3}
18761876

1877-
copier := NewPacketReactor(context.Background())
1877+
copier := NewPacketReactor(context.Background(), nil)
18781878
defer copier.Close()
18791879

18801880
var capturedErr error
@@ -1929,7 +1929,7 @@ func TestPacketReactor_ReadError(t *testing.T) {
19291929
readErr := errors.New("simulated packet read error")
19301930
errorSrc := &errorPacketReader{PacketConn: pipeA, readError: readErr}
19311931

1932-
copier := NewPacketReactor(context.Background())
1932+
copier := NewPacketReactor(context.Background(), nil)
19331933
defer copier.Close()
19341934

19351935
var capturedErr error
@@ -1981,7 +1981,7 @@ func TestPacketReactor_StateMachine_ConcurrentWakeup(t *testing.T) {
19811981
proxyA := newFDPacketConn(t, proxyAConn, serverAddr)
19821982
proxyB := newFDPacketConn(t, proxyBConn, clientAddr)
19831983

1984-
copier := NewPacketReactor(context.Background())
1984+
copier := NewPacketReactor(context.Background(), nil)
19851985
defer copier.Close()
19861986

19871987
copyDone := make(chan struct{})
@@ -2034,7 +2034,7 @@ func TestPacketReactor_StateMachine_CloseWhileActive(t *testing.T) {
20342034
pipeA := newTestPacketPipe(addr1)
20352035
pipeB := newTestPacketPipe(addr2)
20362036

2037-
copier := NewPacketReactor(context.Background())
2037+
copier := NewPacketReactor(context.Background(), nil)
20382038

20392039
copyDone := make(chan struct{})
20402040
go func() {
@@ -2088,7 +2088,7 @@ func TestPacketReactor_Counters(t *testing.T) {
20882088
defer pipeA.Close()
20892089
defer pipeB.Close()
20902090

2091-
copier := NewPacketReactor(context.Background())
2091+
copier := NewPacketReactor(context.Background(), nil)
20922092
defer copier.Close()
20932093

20942094
copyDone := make(chan struct{})

0 commit comments

Comments
 (0)