Skip to content

Commit a93170b

Browse files
marten-seemannmarten-seemann
authored
Merge pull request #278 from libp2p/remove-goprocess
stop using goprocess to control teardown
2 parents 014b22f + 0537306 commit a93170b

12 files changed

+102
-202
lines changed

‎p2p/net/swarm/dial_test.go

+24-39
Original file line numberDiff line numberDiff line change
@@ -7,21 +7,21 @@ import (
77
"testing"
88
"time"
99

10-
addrutil "github.com/libp2p/go-addr-util"
10+
. "github.com/libp2p/go-libp2p-swarm"
1111

12+
addrutil "github.com/libp2p/go-addr-util"
1213
"github.com/libp2p/go-libp2p-core/network"
1314
"github.com/libp2p/go-libp2p-core/peer"
1415
"github.com/libp2p/go-libp2p-core/peerstore"
15-
"github.com/libp2p/go-libp2p-core/transport"
16-
1716
testutil "github.com/libp2p/go-libp2p-core/test"
17+
"github.com/libp2p/go-libp2p-core/transport"
1818
swarmt "github.com/libp2p/go-libp2p-swarm/testing"
1919
"github.com/libp2p/go-libp2p-testing/ci"
2020

2121
ma "github.com/multiformats/go-multiaddr"
2222
manet "github.com/multiformats/go-multiaddr/net"
2323

24-
. "github.com/libp2p/go-libp2p-swarm"
24+
"github.com/stretchr/testify/require"
2525
)
2626

2727
func init() {
@@ -36,50 +36,37 @@ func closeSwarms(swarms []*Swarm) {
3636

3737
func TestBasicDialPeer(t *testing.T) {
3838
t.Parallel()
39-
ctx := context.Background()
4039

41-
swarms := makeSwarms(ctx, t, 2)
40+
swarms := makeSwarms(t, 2)
4241
defer closeSwarms(swarms)
4342
s1 := swarms[0]
4443
s2 := swarms[1]
4544

4645
s1.Peerstore().AddAddrs(s2.LocalPeer(), s2.ListenAddresses(), peerstore.PermanentAddrTTL)
4746

48-
c, err := s1.DialPeer(ctx, s2.LocalPeer())
49-
if err != nil {
50-
t.Fatal(err)
51-
}
52-
53-
s, err := c.NewStream(ctx)
54-
if err != nil {
55-
t.Fatal(err)
56-
}
47+
c, err := s1.DialPeer(context.Background(), s2.LocalPeer())
48+
require.NoError(t, err)
5749

50+
s, err := c.NewStream(context.Background())
51+
require.NoError(t, err)
5852
s.Close()
5953
}
6054

6155
func TestDialWithNoListeners(t *testing.T) {
6256
t.Parallel()
63-
ctx := context.Background()
64-
65-
s1 := makeDialOnlySwarm(ctx, t)
6657

67-
swarms := makeSwarms(ctx, t, 1)
58+
s1 := makeDialOnlySwarm(t)
59+
swarms := makeSwarms(t, 1)
6860
defer closeSwarms(swarms)
6961
s2 := swarms[0]
7062

7163
s1.Peerstore().AddAddrs(s2.LocalPeer(), s2.ListenAddresses(), peerstore.PermanentAddrTTL)
7264

73-
c, err := s1.DialPeer(ctx, s2.LocalPeer())
74-
if err != nil {
75-
t.Fatal(err)
76-
}
77-
78-
s, err := c.NewStream(ctx)
79-
if err != nil {
80-
t.Fatal(err)
81-
}
65+
c, err := s1.DialPeer(context.Background(), s2.LocalPeer())
66+
require.NoError(t, err)
8267

68+
s, err := c.NewStream(context.Background())
69+
require.NoError(t, err)
8370
s.Close()
8471
}
8572

@@ -104,7 +91,7 @@ func TestSimultDials(t *testing.T) {
10491
t.Parallel()
10592

10693
ctx := context.Background()
107-
swarms := makeSwarms(ctx, t, 2, swarmt.OptDisableReuseport)
94+
swarms := makeSwarms(t, 2, swarmt.OptDisableReuseport)
10895

10996
// connect everyone
11097
{
@@ -175,7 +162,7 @@ func TestDialWait(t *testing.T) {
175162
t.Parallel()
176163

177164
ctx := context.Background()
178-
swarms := makeSwarms(ctx, t, 1)
165+
swarms := makeSwarms(t, 1)
179166
s1 := swarms[0]
180167
defer s1.Close()
181168

@@ -215,7 +202,7 @@ func TestDialBackoff(t *testing.T) {
215202
t.Parallel()
216203

217204
ctx := context.Background()
218-
swarms := makeSwarms(ctx, t, 2)
205+
swarms := makeSwarms(t, 2)
219206
s1 := swarms[0]
220207
s2 := swarms[1]
221208
defer s1.Close()
@@ -422,7 +409,7 @@ func TestDialBackoffClears(t *testing.T) {
422409
t.Parallel()
423410

424411
ctx := context.Background()
425-
swarms := makeSwarms(ctx, t, 2)
412+
swarms := makeSwarms(t, 2)
426413
s1 := swarms[0]
427414
s2 := swarms[1]
428415
defer s1.Close()
@@ -491,7 +478,7 @@ func TestDialPeerFailed(t *testing.T) {
491478
t.Parallel()
492479
ctx := context.Background()
493480

494-
swarms := makeSwarms(ctx, t, 2)
481+
swarms := makeSwarms(t, 2)
495482
defer closeSwarms(swarms)
496483
testedSwarm, targetSwarm := swarms[0], swarms[1]
497484

@@ -530,7 +517,7 @@ func TestDialPeerFailed(t *testing.T) {
530517
func TestDialExistingConnection(t *testing.T) {
531518
ctx := context.Background()
532519

533-
swarms := makeSwarms(ctx, t, 2)
520+
swarms := makeSwarms(t, 2)
534521
defer closeSwarms(swarms)
535522
s1 := swarms[0]
536523
s2 := swarms[1]
@@ -574,7 +561,7 @@ func TestDialSimultaneousJoin(t *testing.T) {
574561
ctx, cancel := context.WithCancel(context.Background())
575562
defer cancel()
576563

577-
swarms := makeSwarms(ctx, t, 2)
564+
swarms := makeSwarms(t, 2)
578565
s1 := swarms[0]
579566
s2 := swarms[1]
580567
defer s1.Close()
@@ -676,12 +663,10 @@ func TestDialSelf(t *testing.T) {
676663
ctx, cancel := context.WithCancel(context.Background())
677664
defer cancel()
678665

679-
swarms := makeSwarms(ctx, t, 2)
666+
swarms := makeSwarms(t, 2)
680667
s1 := swarms[0]
681668
defer s1.Close()
682669

683670
_, err := s1.DialPeer(ctx, s1.LocalPeer())
684-
if err != ErrDialToSelf {
685-
t.Fatal("expected error from self dial")
686-
}
671+
require.ErrorIs(t, err, ErrDialToSelf, "expected error from self dial")
687672
}

‎p2p/net/swarm/peers_test.go

+1-1
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ import (
1515

1616
func TestPeers(t *testing.T) {
1717
ctx := context.Background()
18-
swarms := makeSwarms(ctx, t, 2)
18+
swarms := makeSwarms(t, 2)
1919
s1 := swarms[0]
2020
s2 := swarms[1]
2121

‎p2p/net/swarm/simul_test.go

+2-5
Original file line numberDiff line numberDiff line change
@@ -18,11 +18,8 @@ import (
1818
)
1919

2020
func TestSimultOpen(t *testing.T) {
21-
2221
t.Parallel()
23-
24-
ctx := context.Background()
25-
swarms := makeSwarms(ctx, t, 2, swarmt.OptDisableReuseport)
22+
swarms := makeSwarms(t, 2, swarmt.OptDisableReuseport)
2623

2724
// connect everyone
2825
{
@@ -32,7 +29,7 @@ func TestSimultOpen(t *testing.T) {
3229
// copy for other peer
3330
log.Debugf("TestSimultOpen: connecting: %s --> %s (%s)", s.LocalPeer(), dst, addr)
3431
s.Peerstore().AddAddr(dst, addr, peerstore.PermanentAddrTTL)
35-
if _, err := s.DialPeer(ctx, dst); err != nil {
32+
if _, err := s.DialPeer(context.Background(), dst); err != nil {
3633
t.Error("error swarm dialing to peer", err)
3734
}
3835
}

‎p2p/net/swarm/swarm.go

+19-39
Original file line numberDiff line numberDiff line change
@@ -18,8 +18,6 @@ import (
1818
"github.com/libp2p/go-libp2p-core/transport"
1919

2020
logging "github.com/ipfs/go-log"
21-
"github.com/jbenet/goprocess"
22-
goprocessctx "github.com/jbenet/goprocess/context"
2321

2422
ma "github.com/multiformats/go-multiaddr"
2523
)
@@ -92,9 +90,11 @@ type Swarm struct {
9290
limiter *dialLimiter
9391
gater connmgr.ConnectionGater
9492

95-
proc goprocess.Process
96-
ctx context.Context
97-
bwc metrics.Reporter
93+
closeOnce sync.Once
94+
ctx context.Context // is canceled when Close is called
95+
ctxCancel context.CancelFunc
96+
97+
bwc metrics.Reporter
9898
}
9999

100100
// NewSwarm constructs a Swarm.
@@ -103,11 +103,14 @@ type Swarm struct {
103103
// `extra` interface{} parameter facilitates the future migration. Supported
104104
// elements are:
105105
// - connmgr.ConnectionGater
106-
func NewSwarm(ctx context.Context, local peer.ID, peers peerstore.Peerstore, bwc metrics.Reporter, extra ...interface{}) *Swarm {
106+
func NewSwarm(local peer.ID, peers peerstore.Peerstore, bwc metrics.Reporter, extra ...interface{}) *Swarm {
107+
ctx, cancel := context.WithCancel(context.Background())
107108
s := &Swarm{
108-
local: local,
109-
peers: peers,
110-
bwc: bwc,
109+
local: local,
110+
peers: peers,
111+
bwc: bwc,
112+
ctx: ctx,
113+
ctxCancel: cancel,
111114
}
112115

113116
s.conns.m = make(map[peer.ID][]*Conn)
@@ -124,25 +127,19 @@ func NewSwarm(ctx context.Context, local peer.ID, peers peerstore.Peerstore, bwc
124127

125128
s.dsync = newDialSync(s.dialWorkerLoop)
126129
s.limiter = newDialLimiter(s.dialAddr)
127-
s.proc = goprocessctx.WithContext(ctx)
128-
s.ctx = goprocessctx.OnClosingContext(s.proc)
129130
s.backf.init(s.ctx)
130-
131-
// Set teardown after setting the context/process so we don't start the
132-
// teardown process early.
133-
s.proc.SetTeardown(s.teardown)
134-
135131
return s
136132
}
137133

138-
func (s *Swarm) teardown() error {
139-
// Wait for the context to be canceled.
140-
// This allows other parts of the swarm to detect that we're shutting
141-
// down.
142-
<-s.ctx.Done()
134+
func (s *Swarm) Close() error {
135+
s.closeOnce.Do(s.close)
136+
return nil
137+
}
143138

144-
// Prevents new connections and/or listeners from being added to the swarm.
139+
func (s *Swarm) close() {
140+
s.ctxCancel()
145141

142+
// Prevents new connections and/or listeners from being added to the swarm.
146143
s.listeners.Lock()
147144
listeners := s.listeners.m
148145
s.listeners.m = nil
@@ -197,13 +194,6 @@ func (s *Swarm) teardown() error {
197194
}
198195
}
199196
wg.Wait()
200-
201-
return nil
202-
}
203-
204-
// Process returns the Process of the swarm
205-
func (s *Swarm) Process() goprocess.Process {
206-
return s.proc
207197
}
208198

209199
func (s *Swarm) addConn(tc transport.CapableConn, dir network.Direction) (*Conn, error) {
@@ -293,16 +283,6 @@ func (s *Swarm) Peerstore() peerstore.Peerstore {
293283
return s.peers
294284
}
295285

296-
// Context returns the context of the swarm
297-
func (s *Swarm) Context() context.Context {
298-
return s.ctx
299-
}
300-
301-
// Close stops the Swarm.
302-
func (s *Swarm) Close() error {
303-
return s.proc.Close()
304-
}
305-
306286
// TODO: We probably don't need the conn handlers.
307287

308288
// SetConnHandler assigns the handler for new connections.

‎p2p/net/swarm/swarm_addr_test.go

+9-23
Original file line numberDiff line numberDiff line change
@@ -6,14 +6,14 @@ import (
66

77
"github.com/libp2p/go-libp2p-core/peerstore"
88
"github.com/libp2p/go-libp2p-core/test"
9+
"github.com/stretchr/testify/require"
910

1011
ma "github.com/multiformats/go-multiaddr"
1112

1213
swarmt "github.com/libp2p/go-libp2p-swarm/testing"
1314
)
1415

1516
func TestDialBadAddrs(t *testing.T) {
16-
1717
m := func(s string) ma.Multiaddr {
1818
maddr, err := ma.NewMultiaddr(s)
1919
if err != nil {
@@ -22,13 +22,12 @@ func TestDialBadAddrs(t *testing.T) {
2222
return maddr
2323
}
2424

25-
ctx := context.Background()
26-
s := makeSwarms(ctx, t, 1)[0]
25+
s := makeSwarms(t, 1)[0]
2726

2827
test := func(a ma.Multiaddr) {
2928
p := test.RandPeerIDFatal(t)
3029
s.Peerstore().AddAddr(p, a, peerstore.PermanentAddrTTL)
31-
if _, err := s.DialPeer(ctx, p); err == nil {
30+
if _, err := s.DialPeer(context.Background(), p); err == nil {
3231
t.Errorf("swarm should not dial: %s", p)
3332
}
3433
}
@@ -39,35 +38,22 @@ func TestDialBadAddrs(t *testing.T) {
3938
}
4039

4140
func TestAddrRace(t *testing.T) {
42-
ctx, cancel := context.WithCancel(context.Background())
43-
defer cancel()
44-
s := makeSwarms(ctx, t, 1)[0]
41+
s := makeSwarms(t, 1)[0]
4542
defer s.Close()
4643

4744
a1, err := s.InterfaceListenAddresses()
48-
if err != nil {
49-
t.Fatal(err)
50-
}
45+
require.NoError(t, err)
5146
a2, err := s.InterfaceListenAddresses()
52-
if err != nil {
53-
t.Fatal(err)
54-
}
47+
require.NoError(t, err)
5548

5649
if len(a1) > 0 && len(a2) > 0 && &a1[0] == &a2[0] {
5750
t.Fatal("got the exact same address set twice; this could lead to data races")
5851
}
5952
}
6053

6154
func TestAddressesWithoutListening(t *testing.T) {
62-
ctx, cancel := context.WithCancel(context.Background())
63-
defer cancel()
64-
s := swarmt.GenSwarm(t, ctx, swarmt.OptDialOnly)
65-
55+
s := swarmt.GenSwarm(t, swarmt.OptDialOnly)
6656
a1, err := s.InterfaceListenAddresses()
67-
if err != nil {
68-
t.Fatal(err)
69-
}
70-
if len(a1) != 0 {
71-
t.Fatalf("expected to be listening on no addresses, was listening on %d", len(a1))
72-
}
57+
require.NoError(t, err)
58+
require.Empty(t, a1, "expected to be listening on no addresses")
7359
}

‎p2p/net/swarm/swarm_listen.go

+1-1
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ func (s *Swarm) AddListenAddr(a ma.Multiaddr) error {
4646
//
4747
// Distinguish between these two cases to avoid confusing users.
4848
select {
49-
case <-s.proc.Closing():
49+
case <-s.ctx.Done():
5050
return ErrSwarmClosed
5151
default:
5252
return ErrNoTransport

0 commit comments

Comments
 (0)