fixing some race conditions
Jeromy committed
Aug 6, 2014 at 21:36 UTC
c22b6aa316c89915ea95219584a46470b80b7f71
4 files changed
+42
-4
peer/peer.go
+14
-1
@@ -3,6 +3,7 @@ package peer
3
import (
4
"encoding/hex"
5
"time"
6
+ "sync"
7
8
u "github.com/jbenet/go-ipfs/util"
9
ma "github.com/jbenet/go-multiaddr"
@@ -31,7 +32,9 @@ type Map map[u.Key]*Peer
32
type Peer struct {
33
ID ID
34
Addresses []*ma.Multiaddr
34
- Distance time.Duration
35
+
36
+ distance time.Duration
37
+ distLock sync.RWMutex
38
}
39
40
// Key returns the ID as a Key (string) for maps.
@@ -60,3 +63,13 @@ func (p *Peer) NetAddress(n string) *ma.Multiaddr {
63
}
64
return nil
65
}
66
+
67
+func (p *Peer) GetDistance() time.Duration {
68
+ return p.distance
69
+}
70
+
71
+func (p *Peer) SetDistance(dist time.Duration) {
72
+ p.distLock.Lock()
73
+ p.distance = dist
74
+ p.distLock.Unlock()
75
+}
routing/dht/dht.go
+1
@@ -186,6 +186,7 @@ func (dht *IpfsDHT) handleMessages() {
186
187
case err := <-dht.network.Chan.Errors:
188
u.DErr("dht err: %s", err)
189
+ panic(err)
190
case <-dht.shutdown:
191
checkTimeouts.Stop()
192
return
routing/dht/routing.go
+8
-2
@@ -48,6 +48,8 @@ func (s *IpfsDHT) PutValue(key u.Key, value []byte) error {
48
}
49
50
// GetValue searches for the value corresponding to given Key.
51
+// If the search does not succeed, a multiaddr string of a closer peer is
52
+// returned along with util.ErrSearchIncomplete
53
func (s *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
54
var p *peer.Peer
55
p = s.routes.NearestPeer(convertKey(key))
@@ -77,7 +79,11 @@ func (s *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
79
if err != nil {
80
return nil,err
81
}
80
- return pmes_out.GetValue(), nil
82
+ if pmes_out.GetSuccess() {
83
+ return pmes_out.GetValue(), nil
84
+ } else {
85
+ return pmes_out.GetValue(), u.ErrSearchIncomplete
86
+ }
87
}
88
}
89
@@ -225,7 +231,7 @@ func (dht *IpfsDHT) Ping(p *peer.Peer, timeout time.Duration) error {
231
select {
232
case <-response_chan:
233
roundtrip := time.Since(before)
228
- p.Distance = roundtrip //TODO: This isnt threadsafe
234
+ p.SetDistance(roundtrip)
235
u.POut("Ping took %s.", roundtrip.String())
236
return nil
237
case <-tout:
routing/dht/table.go
+19
-1
@@ -3,8 +3,10 @@ package dht
3
import (
4
"container/list"
5
"sort"
6
+ "sync"
7
8
peer "github.com/jbenet/go-ipfs/peer"
9
+ u "github.com/jbenet/go-ipfs/util"
10
)
11
12
// RoutingTable defines the routing table.
@@ -13,6 +15,9 @@ type RoutingTable struct {
15
// ID of the local peer
16
local ID
17
18
+ // Blanket lock, refine later for better performance
19
+ tabLock sync.RWMutex
20
+
21
// kBuckets define all the fingers to other nodes.
22
Buckets []*Bucket
23
bucketsize int
@@ -29,6 +34,8 @@ func NewRoutingTable(bucketsize int, local_id ID) *RoutingTable {
34
// Update adds or moves the given peer to the front of its respective bucket
35
// If a peer gets removed from a bucket, it is returned
36
func (rt *RoutingTable) Update(p *peer.Peer) *peer.Peer {
37
+ rt.tabLock.Lock()
38
+ defer rt.tabLock.Unlock()
39
peer_id := convertPeerID(p.ID)
40
cpl := xor(peer_id, rt.local).commonPrefixLen()
41
@@ -88,7 +95,11 @@ func (p peerSorterArr) Less(a, b int) bool {
95
//
96
97
func copyPeersFromList(target ID, peerArr peerSorterArr, peerList *list.List) peerSorterArr {
91
- for e := peerList.Front(); e != nil; e = e.Next() {
98
+ if peerList == nil {
99
+ return peerSorterArr{}
100
+ }
101
+ e := peerList.Front()
102
+ for ; e != nil; {
103
p := e.Value.(*peer.Peer)
104
p_id := convertPeerID(p.ID)
105
pd := peerDistance{
@@ -96,6 +107,11 @@ func copyPeersFromList(target ID, peerArr peerSorterArr, peerList *list.List) pe
107
distance: xor(target, p_id),
108
}
109
peerArr = append(peerArr, &pd)
110
+ if e != nil {
111
+ u.POut("list element was nil.")
112
+ return peerArr
113
+ }
114
+ e = e.Next()
115
}
116
return peerArr
117
}
@@ -112,6 +128,8 @@ func (rt *RoutingTable) NearestPeer(id ID) *peer.Peer {
128
129
// Returns a list of the 'count' closest peers to the given ID
130
func (rt *RoutingTable) NearestPeers(id ID, count int) []*peer.Peer {
131
+ rt.tabLock.RLock()
132
+ defer rt.tabLock.RUnlock()
133
cpl := xor(id, rt.local).commonPrefixLen()
134
135
// Get bucket at cpl index or last bucket