get implementation according to kademlia spec.
Jeromy committed
Aug 15, 2014 at 09:39 UTC
b7a882be89d14392c2243db2b317e01b28a31737
4 files changed
+140
-45
routing/dht/dht.go
+39
@@ -496,6 +496,45 @@ out:
496
dht.network.Send(mes)
497
}
498
499
+func (dht *IpfsDHT) getValueOrPeers(p *peer.Peer, key u.Key, timeout time.Duration, level int) ([]byte, []*peer.Peer, error) {
500
+ pmes, err := dht.getValueSingle(p, key, timeout, level)
501
+ if err != nil {
502
+ return nil, nil, u.WrapError(err, "getValue Error")
503
+ }
504
+
505
+ if pmes.GetSuccess() {
506
+ if pmes.Value == nil { // We were given provider[s]
507
+ val, err := dht.getFromPeerList(key, timeout, pmes.GetPeers(), level)
508
+ if err != nil {
509
+ return nil, nil, err
510
+ }
511
+ return val, nil, nil
512
+ }
513
+
514
+ // Success! We were given the value
515
+ return pmes.GetValue(), nil, nil
516
+ } else {
517
+ // We were given a closer node
518
+ var peers []*peer.Peer
519
+ for _, pb := range pmes.GetPeers() {
520
+ addr, err := ma.NewMultiaddr(pb.GetAddr())
521
+ if err != nil {
522
+ u.PErr(err.Error())
523
+ continue
524
+ }
525
+
526
+ np, err := dht.network.GetConnection(peer.ID(pb.GetId()), addr)
527
+ if err != nil {
528
+ u.PErr(err.Error())
529
+ continue
530
+ }
531
+
532
+ peers = append(peers, np)
533
+ }
534
+ return nil, peers, nil
535
+ }
536
+}
537
+
538
// getValueSingle simply performs the get value RPC with the given parameters
539
func (dht *IpfsDHT) getValueSingle(p *peer.Peer, key u.Key, timeout time.Duration, level int) (*PBDHTMessage, error) {
540
pmes := DHTMessage{
routing/dht/dht_logger.go
+2
-2
@@ -31,8 +31,8 @@ func (l *logDhtRpc) EndLog() {
31
func (l *logDhtRpc) Print() {
32
b, err := json.Marshal(l)
33
if err != nil {
34
- u.POut(err.Error())
34
+ u.DOut(err.Error())
35
} else {
36
- u.POut(string(b))
36
+ u.DOut(string(b))
37
}
38
}
routing/dht/ext_test.go
+4
-8
@@ -88,13 +88,9 @@ func TestGetFailures(t *testing.T) {
88
d.Update(other)
89
90
// This one should time out
91
- _, err := d.GetValue(u.Key("test"), time.Millisecond*5)
91
+ _, err := d.GetValue(u.Key("test"), time.Millisecond*10)
92
if err != nil {
93
- nerr, ok := err.(*u.IpfsError)
94
- if !ok {
95
- t.Fatal("Got different error than we expected.")
96
- }
97
- if nerr.Inner != u.ErrTimeout {
93
+ if err != u.ErrTimeout {
94
t.Fatal("Got different error than we expected.")
95
}
96
} else {
@@ -119,10 +115,10 @@ func TestGetFailures(t *testing.T) {
115
})
116
117
// This one should fail with NotFound
122
- _, err = d.GetValue(u.Key("test"), time.Millisecond*5)
118
+ _, err = d.GetValue(u.Key("test"), time.Millisecond*1000)
119
if err != nil {
120
if err != u.ErrNotFound {
125
- t.Fatal("Expected ErrNotFound, got: %s", err)
121
+ t.Fatalf("Expected ErrNotFound, got: %s", err)
122
}
123
} else {
124
t.Fatal("expected error, got none.")
routing/dht/routing.go
+95
-35
@@ -5,6 +5,7 @@ import (
5
"encoding/json"
6
"errors"
7
"math/rand"
8
+ "sync"
9
"time"
10
11
proto "code.google.com/p/goprotobuf/proto"
@@ -56,6 +57,30 @@ func (s *IpfsDHT) PutValue(key u.Key, value []byte) {
57
}
58
}
59
60
+// A counter for incrementing a variable across multiple threads
61
+type counter struct {
62
+ n int
63
+ mut sync.RWMutex
64
+}
65
+
66
+func (c *counter) Increment() {
67
+ c.mut.Lock()
68
+ c.n++
69
+ c.mut.Unlock()
70
+}
71
+
72
+func (c *counter) Decrement() {
73
+ c.mut.Lock()
74
+ c.n--
75
+ c.mut.Unlock()
76
+}
77
+
78
+func (c *counter) Size() int {
79
+ c.mut.RLock()
80
+ defer c.mut.RUnlock()
81
+ return c.n
82
+}
83
+
84
// GetValue searches for the value corresponding to given Key.
85
// If the search does not succeed, a multiaddr string of a closer peer is
86
// returned along with util.ErrSearchIncomplete
@@ -65,7 +90,6 @@ func (s *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
90
ll.EndLog()
91
ll.Print()
92
}()
68
- route_level := 0
93
94
// If we have it local, dont bother doing an RPC!
95
// NOTE: this might not be what we want to do...
@@ -76,54 +100,90 @@ func (s *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
100
return val, nil
101
}
102
79
- p := s.routes[route_level].NearestPeer(kb.ConvertKey(key))
80
- if p == nil {
103
+ route_level := 0
104
+ closest := s.routes[route_level].NearestPeers(kb.ConvertKey(key), PoolSize)
105
+ if closest == nil || len(closest) == 0 {
106
return nil, kb.ErrLookupFailure
107
}
108
84
- for route_level < len(s.routes) && p != nil {
85
- ll.RpcCount++
86
- pmes, err := s.getValueSingle(p, key, timeout, route_level)
87
- if err != nil {
88
- return nil, u.WrapError(err, "getValue Error")
89
- }
109
+ val_chan := make(chan []byte)
110
+ npeer_chan := make(chan *peer.Peer, 30)
111
+ proc_peer := make(chan *peer.Peer, 30)
112
+ err_chan := make(chan error)
113
+ after := time.After(timeout)
114
91
- if pmes.GetSuccess() {
92
- if pmes.Value == nil { // We were given provider[s]
93
- ll.RpcCount++
94
- return s.getFromPeerList(key, timeout, pmes.GetPeers(), route_level)
95
- }
115
+ for _, p := range closest {
116
+ npeer_chan <- p
117
+ }
118
97
- // Success! We were given the value
98
- ll.Success = true
99
- return pmes.GetValue(), nil
100
- } else {
101
- // We were given a closer node
102
- closers := pmes.GetPeers()
103
- if len(closers) > 0 {
104
- if peer.ID(closers[0].GetId()).Equal(s.self.ID) {
105
- u.DOut("Got myself back as a closer peer.")
106
- return nil, u.ErrNotFound
119
+ c := counter{}
120
+
121
+ // This limit value is referred to as k in the kademlia paper
122
+ limit := 20
123
+ count := 0
124
+ go func() {
125
+ for {
126
+ select {
127
+ case p := <-npeer_chan:
128
+ count++
129
+ if count >= limit {
130
+ break
131
}
108
- maddr, err := ma.NewMultiaddr(closers[0].GetAddr())
109
- if err != nil {
110
- // ??? Move up route level???
111
- panic("not yet implemented")
132
+ c.Increment()
133
+ proc_peer <- p
134
+ default:
135
+ if c.Size() == 0 {
136
+ err_chan <- u.ErrNotFound
137
}
138
+ }
139
+ }
140
+ }()
141
114
- np, err := s.network.GetConnection(peer.ID(closers[0].GetId()), maddr)
142
+ process := func() {
143
+ for {
144
+ select {
145
+ case p, ok := <-proc_peer:
146
+ if !ok || p == nil {
147
+ c.Decrement()
148
+ return
149
+ }
150
+ val, peers, err := s.getValueOrPeers(p, key, timeout/4, route_level)
151
if err != nil {
116
- u.PErr("[%s] Failed to connect to: %s", s.self.ID.Pretty(), closers[0].GetAddr())
117
- route_level++
152
+ u.DErr(err.Error())
153
+ c.Decrement()
154
continue
155
}
120
- p = np
121
- } else {
122
- route_level++
156
+ if val != nil {
157
+ val_chan <- val
158
+ c.Decrement()
159
+ return
160
+ }
161
+
162
+ for _, np := range peers {
163
+ // TODO: filter out peers that arent closer
164
+ npeer_chan <- np
165
+ }
166
+ c.Decrement()
167
}
168
}
169
}
126
- return nil, u.ErrNotFound
170
+
171
+ concurFactor := 3
172
+ for i := 0; i < concurFactor; i++ {
173
+ go process()
174
+ }
175
+
176
+ select {
177
+ case val := <-val_chan:
178
+ close(npeer_chan)
179
+ return val, nil
180
+ case err := <-err_chan:
181
+ close(npeer_chan)
182
+ return nil, err
183
+ case <-after:
184
+ close(npeer_chan)
185
+ return nil, u.ErrTimeout
186
+ }
187
}
188
189
// Value provider layer of indirection.