fix(routing/dht) _always_ close chan on exit of FindProvidersAsync
the important change here is that within FindProvidersAsync, the channel is closed using a `defer`. This ensures the channel is always closed, regardless of the path taken to exit. + misc cleanup cc @whyrusleeping @jbenet License: MIT Signed-off-by: Brian Tiger Chow <brian@perfmode.com>
Brian Tiger Chow committed
Dec 2, 2014 at 00:19 UTC
f756088d26525bd2b9242ac06235137890f334f2
2 files changed
+22
-17
routing/dht/dht.go
+3
-6
@@ -194,24 +194,21 @@ func (dht *IpfsDHT) sendRequest(ctx context.Context, p peer.Peer, pmes *pb.Messa
194
195
start := time.Now()
196
197
- log.Event(ctx, "sentMessage", dht.self, p, pmes)
198
-
199
- rmes, err := dht.sender.SendRequest(ctx, mes)
197
+ rmes, err := dht.sender.SendRequest(ctx, mes) // respect?
198
if err != nil {
199
return nil, err
200
}
201
if rmes == nil {
202
return nil, errors.New("no response to request")
203
}
204
+ log.Event(ctx, "sentMessage", dht.self, p, pmes)
205
207
- rtt := time.Since(start)
208
- rmes.Peer().SetLatency(rtt)
206
+ rmes.Peer().SetLatency(time.Since(start))
207
208
rpmes := new(pb.Message)
209
if err := proto.Unmarshal(rmes.Data(), rpmes); err != nil {
210
return nil, err
211
}
214
-
212
return rpmes, nil
213
}
214
routing/dht/routing.go
+19
-11
@@ -129,21 +129,27 @@ func (dht *IpfsDHT) FindProvidersAsync(ctx context.Context, key u.Key, count int
129
log.Event(ctx, "findProviders", &key)
130
peerOut := make(chan peer.Peer, count)
131
go func() {
132
+ defer close(peerOut)
133
+
134
ps := newPeerSet()
135
+ // TODO may want to make this function async to hide latency
136
provs := dht.providers.GetProviders(key)
137
for _, p := range provs {
138
count--
139
// NOTE: assuming that this list of peers is unique
140
ps.Add(p)
138
- peerOut <- p
141
+ select {
142
+ case peerOut <- p:
143
+ case <-ctx.Done():
144
+ return
145
+ }
146
if count <= 0 {
147
return
148
}
149
}
150
144
- wg := new(sync.WaitGroup)
145
- peers := dht.routingTables[0].NearestPeers(kb.ConvertKey(key), AlphaValue)
146
- for _, pp := range peers {
151
+ var wg sync.WaitGroup
152
+ for _, pp := range dht.routingTables[0].NearestPeers(kb.ConvertKey(key), AlphaValue) {
153
wg.Add(1)
154
go func(p peer.Peer) {
155
defer wg.Done()
@@ -156,16 +162,16 @@ func (dht *IpfsDHT) FindProvidersAsync(ctx context.Context, key u.Key, count int
162
}(pp)
163
}
164
wg.Wait()
159
- close(peerOut)
165
}()
166
return peerOut
167
}
168
169
func (dht *IpfsDHT) addPeerListAsync(ctx context.Context, k u.Key, peers []*pb.Message_Peer, ps *peerSet, count int, out chan peer.Peer) {
165
- done := make(chan struct{})
170
+ var wg sync.WaitGroup
171
for _, pbp := range peers {
172
+ wg.Add(1)
173
go func(mp *pb.Message_Peer) {
168
- defer func() { done <- struct{}{} }()
174
+ defer wg.Done()
175
// construct new peer
176
p, err := dht.ensureConnectedToPeer(ctx, mp)
177
if err != nil {
@@ -179,15 +185,17 @@ func (dht *IpfsDHT) addPeerListAsync(ctx context.Context, k u.Key, peers []*pb.M
185
186
dht.providers.AddProvider(k, p)
187
if ps.AddIfSmallerThan(p, count) {
182
- out <- p
188
+ select {
189
+ case out <- p:
190
+ case <-ctx.Done():
191
+ return
192
+ }
193
} else if ps.Size() >= count {
194
return
195
}
196
}(pbp)
197
}
188
- for _ = range peers {
189
- <-done
190
- }
198
+ wg.Wait()
199
}
200
201
// Find specific Peer