dht: even more logging.
Juan Batiz-Benet committed
Jan 5, 2015 at 04:35 UTC
172801712e0eae07828797b3fd061dd5bbe11326
3 files changed
+72
-52
routing/dht/dht.go
+5
@@ -89,6 +89,11 @@ func (dht *IpfsDHT) LocalPeer() peer.ID {
89
return dht.self
90
}
91
92
+// log returns the dht's logger
93
+func (dht *IpfsDHT) log() eventlog.EventLogger {
94
+ return log.Prefix("dht(%s)", dht.self)
95
+}
96
+
97
// Connect to a new peer at the given address, ping and add to the routing table
98
func (dht *IpfsDHT) Connect(ctx context.Context, npeer peer.ID) error {
99
// TODO: change interface to accept a PeerInfo as well.
routing/dht/query.go
+39
-34
@@ -7,6 +7,7 @@ import (
7
queue "github.com/jbenet/go-ipfs/p2p/peer/queue"
8
"github.com/jbenet/go-ipfs/routing"
9
u "github.com/jbenet/go-ipfs/util"
10
+ eventlog "github.com/jbenet/go-ipfs/util/eventlog"
11
pset "github.com/jbenet/go-ipfs/util/peerset"
12
todoctr "github.com/jbenet/go-ipfs/util/todocounter"
13
@@ -55,32 +56,18 @@ func (q *dhtQuery) Run(ctx context.Context, peers []peer.ID) (*dhtQueryResult, e
56
}
57
58
type dhtQueryRunner struct {
59
+ query *dhtQuery // query to run
60
+ peersSeen *pset.PeerSet // all peers queried. prevent querying same peer 2x
61
+ peersToQuery *queue.ChanQueue // peers remaining to be queried
62
+ peersRemaining todoctr.Counter // peersToQuery + currently processing
63
59
- // the query to run
60
- query *dhtQuery
64
+ result *dhtQueryResult // query result
65
+ errs []error // result errors. maybe should be a map[peer.ID]error
66
62
- // peersToQuery is a list of peers remaining to query
63
- peersToQuery *queue.ChanQueue
67
+ rateLimit chan struct{} // processing semaphore
68
+ log eventlog.EventLogger
69
65
- // peersSeen are all the peers queried. used to prevent querying same peer 2x
66
- peersSeen *pset.PeerSet
67
-
68
- // rateLimit is a channel used to rate limit our processing (semaphore)
69
- rateLimit chan struct{}
70
-
71
- // peersRemaining is a counter of peers remaining (toQuery + processing)
72
- peersRemaining todoctr.Counter
73
-
74
- // context group
70
cg ctxgroup.ContextGroup
76
-
77
- // result
78
- result *dhtQueryResult
79
-
80
- // result errors
81
- errs []error
82
-
83
- // lock for concurrent access to fields
71
sync.RWMutex
72
}
73
@@ -96,6 +83,11 @@ func newQueryRunner(ctx context.Context, q *dhtQuery) *dhtQueryRunner {
83
}
84
85
func (r *dhtQueryRunner) Run(peers []peer.ID) (*dhtQueryResult, error) {
86
+ log := log.Prefix("dht(%s).Query(%s).Run(%d)", r.query.dht.self, r.query.key, len(peers))
87
+ r.log = log
88
+ log.Debug("enter")
89
+ defer log.Debug("end")
90
+
91
log.Debugf("Run query with %d peers.", len(peers))
92
if len(peers) == 0 {
93
log.Warning("Running query with no peers!")
@@ -115,6 +107,7 @@ func (r *dhtQueryRunner) Run(peers []peer.ID) (*dhtQueryResult, error) {
107
// go do this thing.
108
// do it as a child func to make sure Run exits
109
// ONLY AFTER spawn workers has exited.
110
+ log.Debugf("go spawn workers")
111
r.cg.AddChildFunc(r.spawnWorkers)
112
113
// so workers are working.
@@ -124,41 +117,45 @@ func (r *dhtQueryRunner) Run(peers []peer.ID) (*dhtQueryResult, error) {
117
118
select {
119
case <-r.peersRemaining.Done():
120
+ log.Debug("all peers ended")
121
r.cg.Close()
122
r.RLock()
123
defer r.RUnlock()
124
125
if len(r.errs) > 0 {
132
- err = r.errs[0]
126
+ err = r.errs[0] // take the first?
127
}
128
129
case <-r.cg.Closed():
130
+ log.Debug("r.cg.Closed()")
131
+
132
r.RLock()
133
defer r.RUnlock()
134
err = r.cg.Context().Err() // collect the error.
135
}
136
137
if r.result != nil && r.result.success {
138
+ log.Debug("success: %s", r.result)
139
return r.result, nil
140
}
141
142
+ log.Debug("failure: %s", err)
143
return nil, err
144
}
145
146
func (r *dhtQueryRunner) addPeerToQuery(ctx context.Context, next peer.ID) {
147
// if new peer is ourselves...
148
if next == r.query.dht.self {
149
+ r.log.Debug("addPeerToQuery skip self")
150
return
151
}
152
153
if !r.peersSeen.TryAdd(next) {
155
- log.Debug("query peer was already seen")
154
+ r.log.Debugf("addPeerToQuery skip seen %s", next)
155
return
156
}
157
159
- log.Debugf("adding peer to query: %v", next)
160
-
161
- // do this after unlocking to prevent possible deadlocks.
158
+ r.log.Debugf("addPeerToQuery adding %s", next)
159
r.peersRemaining.Increment(1)
160
select {
161
case r.peersToQuery.EnqChan <- next:
@@ -167,6 +164,10 @@ func (r *dhtQueryRunner) addPeerToQuery(ctx context.Context, next peer.ID) {
164
}
165
166
func (r *dhtQueryRunner) spawnWorkers(parent ctxgroup.ContextGroup) {
167
+ log := r.log.Prefix("spawnWorkers")
168
+ log.Debugf("begin")
169
+ defer log.Debugf("end")
170
+
171
for {
172
173
select {
@@ -192,7 +193,9 @@ func (r *dhtQueryRunner) spawnWorkers(parent ctxgroup.ContextGroup) {
193
}
194
195
func (r *dhtQueryRunner) queryPeer(cg ctxgroup.ContextGroup, p peer.ID) {
195
- log.Debugf("spawned worker for: %v", p)
196
+ log := r.log.Prefix("queryPeer(%s)", p)
197
+ log.Debugf("spawned")
198
+ defer log.Debugf("finished")
199
200
// make sure we rate limit concurrency.
201
select {
@@ -203,34 +206,36 @@ func (r *dhtQueryRunner) queryPeer(cg ctxgroup.ContextGroup, p peer.ID) {
206
}
207
208
// ok let's do this!
206
- log.Debugf("running worker for: %v", p)
209
+ log.Debugf("running")
210
211
// make sure we do this when we exit
212
defer func() {
213
// signal we're done proccessing peer p
211
- log.Debugf("completing worker for: %v", p)
214
+ log.Debugf("completed")
215
r.peersRemaining.Decrement(1)
216
r.rateLimit <- struct{}{}
217
}()
218
219
// make sure we're connected to the peer.
220
if conns := r.query.dht.host.Network().ConnsToPeer(p); len(conns) == 0 {
218
- log.Infof("worker for: %v -- not connected. dial start", p)
221
+ log.Infof("not connected. dialing.")
222
223
pi := peer.PeerInfo{ID: p}
224
if err := r.query.dht.host.Connect(cg.Context(), pi); err != nil {
222
- log.Debugf("ERROR worker for: %v -- err connecting: %v", p, err)
225
+ log.Debugf("Error connecting: %s", err)
226
r.Lock()
227
r.errs = append(r.errs, err)
228
r.Unlock()
229
return
230
}
231
229
- log.Infof("worker for: %v -- not connected. dial success!", p)
232
+ log.Debugf("connected. dial success.")
233
}
234
235
// finally, run the query against this peer
236
+ log.Debugf("query running")
237
res, err := r.query.qfunc(cg.Context(), p)
238
+ log.Debugf("query finished")
239
240
if err != nil {
241
log.Debugf("ERROR worker for: %v %v", p, err)
@@ -239,7 +244,7 @@ func (r *dhtQueryRunner) queryPeer(cg ctxgroup.ContextGroup, p peer.ID) {
244
r.Unlock()
245
246
} else if res.success {
242
- log.Debugf("SUCCESS worker for: %v", p, res)
247
+ log.Debugf("SUCCESS worker for: %v %s", p, res)
248
r.Lock()
249
r.result = res
250
r.Unlock()
routing/dht/routing.go
+28
-18
@@ -1,7 +1,6 @@
1
package dht
2
3
import (
4
- "fmt"
4
"math"
5
"sync"
6
@@ -66,25 +65,29 @@ func (dht *IpfsDHT) PutValue(ctx context.Context, key u.Key, value []byte) error
65
// If the search does not succeed, a multiaddr string of a closer peer is
66
// returned along with util.ErrSearchIncomplete
67
func (dht *IpfsDHT) GetValue(ctx context.Context, key u.Key) ([]byte, error) {
69
- log.Debugf("Get Value [%s]", key)
68
+ log := dht.log().Prefix("GetValue(%s)", key)
69
+ log.Debugf("start")
70
+ defer log.Debugf("end")
71
72
// If we have it local, dont bother doing an RPC!
73
val, err := dht.getLocal(key)
74
if err == nil {
74
- log.Debug("Got value locally!")
75
+ log.Debug("have it locally")
76
return val, nil
77
}
78
79
// get closest peers in the routing table
80
+ rtp := dht.routingTable.ListPeers()
81
+ log.Debugf("peers in rt: %s", len(rtp), rtp)
82
+
83
closest := dht.routingTable.NearestPeers(kb.ConvertKey(key), PoolSize)
84
if closest == nil || len(closest) == 0 {
81
- log.Warning("Got no peers back from routing table!")
85
+ log.Warning("No peers from routing table!")
86
return nil, errors.Wrap(kb.ErrLookupFailure)
87
}
88
89
// setup the Query
90
query := dht.newQuery(key, func(ctx context.Context, p peer.ID) (*dhtQueryResult, error) {
87
-
91
val, peers, err := dht.getValueOrPeers(ctx, p, key)
92
if err != nil {
93
return nil, err
@@ -117,9 +120,13 @@ func (dht *IpfsDHT) GetValue(ctx context.Context, key u.Key) ([]byte, error) {
120
121
// Provide makes this node announce that it can provide a value for the given key
122
func (dht *IpfsDHT) Provide(ctx context.Context, key u.Key) error {
120
-
123
+ log := dht.log().Prefix("Provide(%s)", key)
124
+ log.Debugf("start", key)
125
log.Event(ctx, "provideBegin", &key)
126
+ defer log.Debugf("end", key)
127
defer log.Event(ctx, "provideEnd", &key)
128
+
129
+ // add self locally
130
dht.providers.AddProvider(key, dht.self)
131
132
peers, err := dht.getClosestPeers(ctx, key)
@@ -132,6 +139,7 @@ func (dht *IpfsDHT) Provide(ctx context.Context, key u.Key) error {
139
wg.Add(1)
140
go func(p peer.ID) {
141
defer wg.Done()
142
+ log.Debugf("putProvider(%s, %s)", key, p)
143
err := dht.putProvider(ctx, p, string(key))
144
if err != nil {
145
log.Error(err)
@@ -231,9 +239,12 @@ func (dht *IpfsDHT) FindProvidersAsync(ctx context.Context, key u.Key, count int
239
}
240
241
func (dht *IpfsDHT) findProvidersAsyncRoutine(ctx context.Context, key u.Key, count int, peerOut chan peer.PeerInfo) {
242
+ log := dht.log().Prefix("FindProviders(%s)", key)
243
+
244
defer close(peerOut)
245
defer log.Event(ctx, "findProviders end", &key)
236
- log.Debugf("%s FindProviders %s", dht.self, key)
246
+ log.Debug("begin")
247
+ defer log.Debug("begin")
248
249
ps := pset.NewLimited(count)
250
provs := dht.providers.GetProviders(ctx, key)
@@ -255,25 +266,24 @@ func (dht *IpfsDHT) findProvidersAsyncRoutine(ctx context.Context, key u.Key, co
266
267
// setup the Query
268
query := dht.newQuery(key, func(ctx context.Context, p peer.ID) (*dhtQueryResult, error) {
258
-
259
- reqDesc := fmt.Sprintf("%s findProviders(%s).Query(%s): ", dht.self, key, p)
260
- log.Debugf("%s begin", reqDesc)
261
- defer log.Debugf("%s end", reqDesc)
269
+ log := log.Prefix("Query(%s)", p)
270
+ log.Debugf("begin")
271
+ defer log.Debugf("end")
272
273
pmes, err := dht.findProvidersSingle(ctx, p, key)
274
if err != nil {
275
return nil, err
276
}
277
268
- log.Debugf("%s got %d provider entries", reqDesc, len(pmes.GetProviderPeers()))
278
+ log.Debugf("%d provider entries", len(pmes.GetProviderPeers()))
279
provs := pb.PBPeersToPeerInfos(pmes.GetProviderPeers())
270
- log.Debugf("%s got %d provider entries decoded", reqDesc, len(provs))
280
+ log.Debugf("%d provider entries decoded", len(provs))
281
282
// Add unique providers from request, up to 'count'
283
for _, prov := range provs {
274
- log.Debugf("%s got provider: %s", reqDesc, prov)
284
+ log.Debugf("got provider: %s", prov)
285
if ps.TryAdd(prov.ID) {
276
- log.Debugf("%s using provider: %s", reqDesc, prov)
286
+ log.Debugf("using provider: %s", prov)
287
select {
288
case peerOut <- prov:
289
case <-ctx.Done():
@@ -282,7 +292,7 @@ func (dht *IpfsDHT) findProvidersAsyncRoutine(ctx context.Context, key u.Key, co
292
}
293
}
294
if ps.Size() >= count {
285
- log.Debugf("%s got enough providers (%d/%d)", reqDesc, ps.Size(), count)
295
+ log.Debugf("got enough providers (%d/%d)", ps.Size(), count)
296
return &dhtQueryResult{success: true}, nil
297
}
298
}
@@ -290,14 +300,14 @@ func (dht *IpfsDHT) findProvidersAsyncRoutine(ctx context.Context, key u.Key, co
300
// Give closer peers back to the query to be queried
301
closer := pmes.GetCloserPeers()
302
clpeers := pb.PBPeersToPeerInfos(closer)
293
- log.Debugf("%s got closer peers: %s", reqDesc, clpeers)
303
+ log.Debugf("got closer peers: %d %s", len(clpeers), clpeers)
304
return &dhtQueryResult{closerPeers: clpeers}, nil
305
})
306
307
peers := dht.routingTable.NearestPeers(kb.ConvertKey(key), AlphaValue)
308
_, err := query.Run(ctx, peers)
309
if err != nil {
300
- log.Errorf("FindProviders Query error: %s", err)
310
+ log.Errorf("Query error: %s", err)
311
}
312
}
313