dht ctxcloserify
Juan Batiz-Benet committed
Oct 25, 2014 at 07:12 UTC
d79ebe68d24611cfe76c7ae15e60a520b0529a0f
6 files changed
+42
-31
core/core.go
+1
@@ -138,6 +138,7 @@ func NewIpfsNode(cfg *config.Config, online bool) (n *IpfsNode, err error) {
138
// TODO(brian): perform this inside NewDHT factory method
139
dhtService.SetHandler(dhtRouting) // wire the handler to the service.
140
n.Routing = dhtRouting
141
+ n.AddCloserChild(dhtRouting)
142
143
// setup exchange service
144
const alwaysSendToPeer = true // use YesManStrategy
routing/dht/dht.go
+10
-5
@@ -14,6 +14,7 @@ import (
14
pb "github.com/jbenet/go-ipfs/routing/dht/pb"
15
kb "github.com/jbenet/go-ipfs/routing/kbucket"
16
u "github.com/jbenet/go-ipfs/util"
17
+ ctxc "github.com/jbenet/go-ipfs/util/ctxcloser"
18
19
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
20
ds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
@@ -56,7 +57,7 @@ type IpfsDHT struct {
57
//lock to make diagnostics work better
58
diaglock sync.Mutex
59
59
- ctx context.Context
60
+ ctxc.ContextCloser
61
}
62
63
// NewDHT creates a new DHT object with the given peer as the 'local' host
@@ -67,9 +68,10 @@ func NewDHT(ctx context.Context, p peer.Peer, ps peer.Peerstore, dialer inet.Dia
68
dht.datastore = dstore
69
dht.self = p
70
dht.peerstore = ps
70
- dht.ctx = ctx
71
+ dht.ContextCloser = ctxc.NewContextCloser(ctx, nil)
72
72
- dht.providers = NewProviderManager(p.ID())
73
+ dht.providers = NewProviderManager(dht.Context(), p.ID())
74
+ dht.AddCloserChild(dht.providers)
75
76
dht.routingTables = make([]*kb.RoutingTable, 3)
77
dht.routingTables[0] = kb.NewRoutingTable(20, kb.ConvertPeerID(p.ID()), time.Millisecond*1000)
@@ -78,6 +80,7 @@ func NewDHT(ctx context.Context, p peer.Peer, ps peer.Peerstore, dialer inet.Dia
80
dht.birth = time.Now()
81
82
if doPinging {
83
+ dht.Children().Add(1)
84
go dht.PingRoutine(time.Second * 10)
85
}
86
return dht
@@ -516,6 +519,8 @@ func (dht *IpfsDHT) loadProvidableKeys() error {
519
520
// PingRoutine periodically pings nearest neighbors.
521
func (dht *IpfsDHT) PingRoutine(t time.Duration) {
522
+ defer dht.Children().Done()
523
+
524
tick := time.Tick(t)
525
for {
526
select {
@@ -524,13 +529,13 @@ func (dht *IpfsDHT) PingRoutine(t time.Duration) {
529
rand.Read(id)
530
peers := dht.routingTables[0].NearestPeers(kb.ConvertKey(u.Key(id)), 5)
531
for _, p := range peers {
527
- ctx, _ := context.WithTimeout(dht.ctx, time.Second*5)
532
+ ctx, _ := context.WithTimeout(dht.Context(), time.Second*5)
533
err := dht.Ping(ctx, p)
534
if err != nil {
535
log.Errorf("Ping error: %s", err)
536
}
537
}
533
- case <-dht.ctx.Done():
538
+ case <-dht.Closing():
539
return
540
}
541
}
routing/dht/dht_test.go
+10
-10
@@ -92,8 +92,8 @@ func TestPing(t *testing.T) {
92
dhtA := setupDHT(ctx, t, peerA)
93
dhtB := setupDHT(ctx, t, peerB)
94
95
- defer dhtA.Halt()
96
- defer dhtB.Halt()
95
+ defer dhtA.Close()
96
+ defer dhtB.Close()
97
defer dhtA.dialer.(inet.Network).Close()
98
defer dhtB.dialer.(inet.Network).Close()
99
@@ -136,8 +136,8 @@ func TestValueGetSet(t *testing.T) {
136
dhtA := setupDHT(ctx, t, peerA)
137
dhtB := setupDHT(ctx, t, peerB)
138
139
- defer dhtA.Halt()
140
- defer dhtB.Halt()
139
+ defer dhtA.Close()
140
+ defer dhtB.Close()
141
defer dhtA.dialer.(inet.Network).Close()
142
defer dhtB.dialer.(inet.Network).Close()
143
@@ -179,7 +179,7 @@ func TestProvides(t *testing.T) {
179
_, peers, dhts := setupDHTS(ctx, 4, t)
180
defer func() {
181
for i := 0; i < 4; i++ {
182
- dhts[i].Halt()
182
+ dhts[i].Close()
183
defer dhts[i].dialer.(inet.Network).Close()
184
}
185
}()
@@ -239,7 +239,7 @@ func TestProvidesAsync(t *testing.T) {
239
_, peers, dhts := setupDHTS(ctx, 4, t)
240
defer func() {
241
for i := 0; i < 4; i++ {
242
- dhts[i].Halt()
242
+ dhts[i].Close()
243
defer dhts[i].dialer.(inet.Network).Close()
244
}
245
}()
@@ -302,7 +302,7 @@ func TestLayeredGet(t *testing.T) {
302
_, peers, dhts := setupDHTS(ctx, 4, t)
303
defer func() {
304
for i := 0; i < 4; i++ {
305
- dhts[i].Halt()
305
+ dhts[i].Close()
306
defer dhts[i].dialer.(inet.Network).Close()
307
}
308
}()
@@ -355,7 +355,7 @@ func TestFindPeer(t *testing.T) {
355
_, peers, dhts := setupDHTS(ctx, 4, t)
356
defer func() {
357
for i := 0; i < 4; i++ {
358
- dhts[i].Halt()
358
+ dhts[i].Close()
359
dhts[i].dialer.(inet.Network).Close()
360
}
361
}()
@@ -443,8 +443,8 @@ func TestConnectCollision(t *testing.T) {
443
t.Fatal("Timeout received!")
444
}
445
446
- dhtA.Halt()
447
- dhtB.Halt()
446
+ dhtA.Close()
447
+ dhtB.Close()
448
dhtA.dialer.(inet.Network).Close()
449
dhtB.dialer.(inet.Network).Close()
450
routing/dht/handlers.go
-6
@@ -205,9 +205,3 @@ func (dht *IpfsDHT) handleAddProvider(p peer.Peer, pmes *pb.Message) (*pb.Messag
205
206
return pmes, nil // send back same msg as confirmation.
207
}
208
-
209
-// Halt stops all communications from this peer and shut down
210
-// TODO -- remove this in favor of context
211
-func (dht *IpfsDHT) Halt() {
212
- dht.providers.Halt()
213
-}
routing/dht/providers.go
+16
-8
@@ -5,6 +5,9 @@ import (
5
6
peer "github.com/jbenet/go-ipfs/peer"
7
u "github.com/jbenet/go-ipfs/util"
8
+ ctxc "github.com/jbenet/go-ipfs/util/ctxcloser"
9
+
10
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
11
)
12
13
type ProviderManager struct {
@@ -14,8 +17,8 @@ type ProviderManager struct {
17
getlocal chan chan []u.Key
18
newprovs chan *addProv
19
getprovs chan *getProv
17
- halt chan struct{}
20
period time.Duration
21
+ ctxc.ContextCloser
22
}
23
24
type addProv struct {
@@ -28,19 +31,24 @@ type getProv struct {
31
resp chan []peer.Peer
32
}
33
31
-func NewProviderManager(local peer.ID) *ProviderManager {
34
+func NewProviderManager(ctx context.Context, local peer.ID) *ProviderManager {
35
pm := new(ProviderManager)
36
pm.getprovs = make(chan *getProv)
37
pm.newprovs = make(chan *addProv)
38
pm.providers = make(map[u.Key][]*providerInfo)
39
pm.getlocal = make(chan chan []u.Key)
40
pm.local = make(map[u.Key]struct{})
38
- pm.halt = make(chan struct{})
41
+ pm.ContextCloser = ctxc.NewContextCloser(ctx, nil)
42
+
43
+ pm.Children().Add(1)
44
go pm.run()
45
+
46
return pm
47
}
48
49
func (pm *ProviderManager) run() {
50
+ defer pm.Children().Done()
51
+
52
tick := time.NewTicker(time.Hour)
53
for {
54
select {
@@ -53,6 +61,7 @@ func (pm *ProviderManager) run() {
61
pi.Value = np.val
62
arr := pm.providers[np.k]
63
pm.providers[np.k] = append(arr, pi)
64
+
65
case gp := <-pm.getprovs:
66
var parr []peer.Peer
67
provs := pm.providers[gp.k]
@@ -60,12 +69,14 @@ func (pm *ProviderManager) run() {
69
parr = append(parr, p.Value)
70
}
71
gp.resp <- parr
72
+
73
case lc := <-pm.getlocal:
74
var keys []u.Key
75
for k, _ := range pm.local {
76
keys = append(keys, k)
77
}
78
lc <- keys
79
+
80
case <-tick.C:
81
for k, provs := range pm.providers {
82
var filtered []*providerInfo
@@ -76,7 +87,8 @@ func (pm *ProviderManager) run() {
87
}
88
pm.providers[k] = filtered
89
}
79
- case <-pm.halt:
90
+
91
+ case <-pm.Closing():
92
return
93
}
94
}
@@ -102,7 +114,3 @@ func (pm *ProviderManager) GetLocal() []u.Key {
114
pm.getlocal <- resp
115
return <-resp
116
}
105
-
106
-func (pm *ProviderManager) Halt() {
107
- pm.halt <- struct{}{}
108
-}
routing/dht/providers_test.go
+5
-2
@@ -5,16 +5,19 @@ import (
5
6
"github.com/jbenet/go-ipfs/peer"
7
u "github.com/jbenet/go-ipfs/util"
8
+
9
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
10
)
11
12
func TestProviderManager(t *testing.T) {
13
+ ctx := context.Background()
14
mid := peer.ID("testing")
12
- p := NewProviderManager(mid)
15
+ p := NewProviderManager(ctx, mid)
16
a := u.Key("test")
17
p.AddProvider(a, peer.WithIDString("testingprovider"))
18
resp := p.GetProviders(a)
19
if len(resp) != 1 {
20
t.Fatal("Could not retrieve provider.")
21
}
19
- p.Halt()
22
+ p.Close()
23
}