finish implementation of Put and Get for DHT
Jeromy committed
Aug 3, 2014 at 17:35 UTC
a85ce3fad31e66912a750e96f9d8bff87f6a1286
5 files changed
+135
-51
routing/dht/dht.go
+7
-3
@@ -21,7 +21,7 @@ import (
21
// IpfsDHT is an implementation of Kademlia with Coral and S/Kademlia modifications.
22
// It is used to implement the base IpfsRouting module.
23
type IpfsDHT struct {
24
- routes RoutingTable
24
+ routes *RoutingTable
25
26
network *swarm.Swarm
27
@@ -53,6 +53,7 @@ func NewDHT(p *peer.Peer) (*IpfsDHT, error) {
53
dht.self = p
54
dht.listeners = make(map[uint64]chan *swarm.Message)
55
dht.shutdown = make(chan struct{})
56
+ dht.routes = NewRoutingTable(20, convertPeerID(p.ID))
57
return dht, nil
58
}
59
@@ -78,14 +79,14 @@ func (dht *IpfsDHT) Connect(addr *ma.Multiaddr) error {
79
80
dht.network.StartConn(conn)
81
81
- // TODO: Add this peer to our routing table
82
+ dht.routes.Update(peer)
83
return nil
84
}
85
86
// Read in all messages from swarm and handle them appropriately
87
// NOTE: this function is just a quick sketch
88
func (dht *IpfsDHT) handleMessages() {
88
- u.DOut("Being message handling routine")
89
+ u.DOut("Begin message handling routine")
90
for {
91
select {
92
case mes := <-dht.network.Chan.Incoming:
@@ -98,6 +99,9 @@ func (dht *IpfsDHT) handleMessages() {
99
continue
100
}
101
102
+ // Update peers latest visit in routing table
103
+ dht.routes.Update(mes.Peer)
104
+
105
// Note: not sure if this is the correct place for this
106
if pmes.GetResponse() {
107
dht.listenLock.RLock()
routing/dht/dht_test.go
+59
-1
@@ -6,9 +6,13 @@ import (
6
ma "github.com/jbenet/go-multiaddr"
7
u "github.com/jbenet/go-ipfs/util"
8
9
+ "fmt"
10
+
11
"time"
12
)
13
14
+var _ = fmt.Println
15
+
16
func TestPing(t *testing.T) {
17
u.Debug = false
18
addr_a,err := ma.NewMultiaddr("/ip4/127.0.0.1/tcp/1234")
@@ -38,7 +42,6 @@ func TestPing(t *testing.T) {
42
t.Fatal(err)
43
}
44
41
-
45
dht_a.Start()
46
dht_b.Start()
47
@@ -52,4 +55,59 @@ func TestPing(t *testing.T) {
55
if err != nil {
56
t.Fatal(err)
57
}
58
+
59
+ dht_a.Halt()
60
+ dht_b.Halt()
61
+}
62
+
63
+func TestValueGetSet(t *testing.T) {
64
+ u.Debug = false
65
+ addr_a,err := ma.NewMultiaddr("/ip4/127.0.0.1/tcp/1235")
66
+ if err != nil {
67
+ t.Fatal(err)
68
+ }
69
+ addr_b,err := ma.NewMultiaddr("/ip4/127.0.0.1/tcp/5679")
70
+ if err != nil {
71
+ t.Fatal(err)
72
+ }
73
+
74
+ peer_a := new(peer.Peer)
75
+ peer_a.AddAddress(addr_a)
76
+ peer_a.ID = peer.ID([]byte("peer_a"))
77
+
78
+ peer_b := new(peer.Peer)
79
+ peer_b.AddAddress(addr_b)
80
+ peer_b.ID = peer.ID([]byte("peer_b"))
81
+
82
+ dht_a,err := NewDHT(peer_a)
83
+ if err != nil {
84
+ t.Fatal(err)
85
+ }
86
+
87
+ dht_b,err := NewDHT(peer_b)
88
+ if err != nil {
89
+ t.Fatal(err)
90
+ }
91
+
92
+ dht_a.Start()
93
+ dht_b.Start()
94
+
95
+ err = dht_a.Connect(addr_b)
96
+ if err != nil {
97
+ t.Fatal(err)
98
+ }
99
+
100
+ err = dht_a.PutValue("hello", []byte("world"))
101
+ if err != nil {
102
+ t.Fatal(err)
103
+ }
104
+
105
+ val, err := dht_a.GetValue("hello", time.Second * 2)
106
+ if err != nil {
107
+ t.Fatal(err)
108
+ }
109
+
110
+ if string(val) != "world" {
111
+ t.Fatalf("Expected 'world' got %s", string(val))
112
+ }
113
}
routing/dht/routing.go
+31
-27
@@ -4,6 +4,8 @@ import (
4
"math/rand"
5
"time"
6
7
+ proto "code.google.com/p/goprotobuf/proto"
8
+
9
peer "github.com/jbenet/go-ipfs/peer"
10
swarm "github.com/jbenet/go-ipfs/swarm"
11
u "github.com/jbenet/go-ipfs/util"
@@ -22,21 +24,20 @@ func GenerateMessageID() uint64 {
24
func (s *IpfsDHT) PutValue(key u.Key, value []byte) error {
25
var p *peer.Peer
26
p = s.routes.NearestPeer(convertKey(key))
27
+ if p == nil {
28
+ u.POut("nbuckets: %d", len(s.routes.Buckets))
29
+ u.POut("%d", s.routes.Buckets[0].Len())
30
+ panic("Table returned nil peer!")
31
+ }
32
26
- pmes_type := DHTMessage_PUT_VALUE
27
- str_key := string(key)
28
- mes_id := GenerateMessageID()
29
-
30
- pmes := new(DHTMessage)
31
- pmes.Type = &pmes_type
32
- pmes.Key = &str_key
33
- pmes.Value = value
34
- pmes.Id = &mes_id
35
-
36
- mes := new(swarm.Message)
37
- mes.Data = []byte(pmes.String())
38
- mes.Peer = p
33
+ pmes := pDHTMessage{
34
+ Type: DHTMessage_PUT_VALUE,
35
+ Key: string(key),
36
+ Value: value,
37
+ Id: GenerateMessageID(),
38
+ }
39
40
+ mes := swarm.NewMessage(p, pmes.ToProtobuf())
41
s.network.Chan.Outgoing <- mes
42
return nil
43
}
@@ -45,21 +46,19 @@ func (s *IpfsDHT) PutValue(key u.Key, value []byte) error {
46
func (s *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
47
var p *peer.Peer
48
p = s.routes.NearestPeer(convertKey(key))
49
+ if p == nil {
50
+ panic("Table returned nil peer!")
51
+ }
52
49
- str_key := string(key)
50
- mes_type := DHTMessage_GET_VALUE
51
- mes_id := GenerateMessageID()
52
- // protobuf structure
53
- pmes := new(DHTMessage)
54
- pmes.Type = &mes_type
55
- pmes.Key = &str_key
56
- pmes.Id = &mes_id
57
-
58
- mes := new(swarm.Message)
59
- mes.Data = []byte(pmes.String())
60
- mes.Peer = p
53
+ pmes := pDHTMessage{
54
+ Type: DHTMessage_GET_VALUE,
55
+ Key: string(key),
56
+ Id: GenerateMessageID(),
57
+ }
58
+ response_chan := s.ListenFor(pmes.Id)
59
62
- response_chan := s.ListenFor(*pmes.Id)
60
+ mes := swarm.NewMessage(p, pmes.ToProtobuf())
61
+ s.network.Chan.Outgoing <- mes
62
63
// Wait for either the response or a timeout
64
timeup := time.After(timeout)
@@ -68,7 +67,12 @@ func (s *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
67
// TODO: unregister listener
68
return nil, u.ErrTimeout
69
case resp := <-response_chan:
71
- return resp.Data, nil
70
+ pmes_out := new(DHTMessage)
71
+ err := proto.Unmarshal(resp.Data, pmes_out)
72
+ if err != nil {
73
+ return nil,err
74
+ }
75
+ return pmes_out.GetValue(), nil
76
}
77
}
78
routing/dht/table.go
+34
-17
@@ -49,7 +49,7 @@ func (rt *RoutingTable) Update(p *peer.Peer) *peer.Peer {
49
new_bucket := bucket.Split(b_id, rt.local)
50
rt.Buckets = append(rt.Buckets, new_bucket)
51
if new_bucket.Len() > rt.bucketsize {
52
- // This is another very rare and annoying case
52
+ // TODO: This is a very rare and annoying case
53
panic("Case not handled.")
54
}
55
@@ -87,10 +87,27 @@ func (p peerSorterArr) Less(a, b int) bool {
87
}
88
//
89
90
+func (rt *RoutingTable) copyPeersFromList(peerArr peerSorterArr, peerList *list.List) peerSorterArr {
91
+ for e := peerList.Front(); e != nil; e = e.Next() {
92
+ p := e.Value.(*peer.Peer)
93
+ p_id := convertPeerID(p.ID)
94
+ pd := peerDistance{
95
+ p: p,
96
+ distance: xor(rt.local, p_id),
97
+ }
98
+ peerArr = append(peerArr, &pd)
99
+ }
100
+ return peerArr
101
+}
102
+
103
// Returns a single peer that is nearest to the given ID
104
func (rt *RoutingTable) NearestPeer(id ID) *peer.Peer {
105
peers := rt.NearestPeers(id, 1)
93
- return peers[0]
106
+ if len(peers) > 0 {
107
+ return peers[0]
108
+ } else {
109
+ return nil
110
+ }
111
}
112
113
// Returns a list of the 'count' closest peers to the given ID
@@ -100,26 +117,26 @@ func (rt *RoutingTable) NearestPeers(id ID, count int) []*peer.Peer {
117
// Get bucket at cpl index or last bucket
118
var bucket *Bucket
119
if cpl >= len(rt.Buckets) {
103
- bucket = rt.Buckets[len(rt.Buckets) - 1]
104
- } else {
105
- bucket = rt.Buckets[cpl]
120
+ cpl = len(rt.Buckets) - 1
121
}
122
+ bucket = rt.Buckets[cpl]
123
124
+ var peerArr peerSorterArr
125
if bucket.Len() == 0 {
109
- // This can happen, very rarely.
110
- panic("Case not yet implemented.")
111
- }
126
+ // In the case of an unusual split, one bucket may be empty.
127
+ // if this happens, search both surrounding buckets for nearest peer
128
+ if cpl > 0 {
129
+ plist := (*list.List)(rt.Buckets[cpl - 1])
130
+ peerArr = rt.copyPeersFromList(peerArr, plist)
131
+ }
132
113
- var peerArr peerSorterArr
114
- plist := (*list.List)(bucket)
115
- for e := plist.Front();e != nil; e = e.Next() {
116
- p := e.Value.(*peer.Peer)
117
- p_id := convertPeerID(p.ID)
118
- pd := peerDistance{
119
- p: p,
120
- distance: xor(rt.local, p_id),
133
+ if cpl < len(rt.Buckets) - 1 {
134
+ plist := (*list.List)(rt.Buckets[cpl + 1])
135
+ peerArr = rt.copyPeersFromList(peerArr, plist)
136
}
122
- peerArr = append(peerArr, &pd)
137
+ } else {
138
+ plist := (*list.List)(bucket)
139
+ peerArr = rt.copyPeersFromList(peerArr, plist)
140
}
141
142
// Sort by distance to local peer
swarm/swarm.go
+4
-3
@@ -101,7 +101,7 @@ func (s *Swarm) Listen() error {
101
for i, addr := range s.local.Addresses {
102
err := s.connListen(addr)
103
if err != nil {
104
- if ret_err != nil {
104
+ if ret_err == nil {
105
ret_err = new(SwarmListenErr)
106
ret_err.Errors = make([]error, len(s.local.Addresses))
107
}
@@ -135,8 +135,9 @@ func (s *Swarm) connListen(maddr *ma.Multiaddr) error {
135
for {
136
nconn, err := list.Accept()
137
if err != nil {
138
- u.PErr("Failed to accept connection: %s - %s [%s]", netstr,
139
- addr, err)
138
+ e := fmt.Errorf("Failed to accept connection: %s - %s [%s]",
139
+ netstr, addr, err)
140
+ s.Chan.Errors <- e
141
return
142
}
143
go s.handleNewConn(nconn)