fix(bitswap) move mutex up to strategy from ledger
addresses concurrent access in bitswap session
Brian Tiger Chow committed
Oct 24, 2014 at 16:15 UTC
c848202c7d7c77a3e7ee5c396d3b748d311e8d00
3 files changed
+27
-42
exchange/bitswap/strategy/ledger.go
+1
-20
@@ -1,7 +1,6 @@
1
package strategy
2
3
import (
4
- "sync"
4
"time"
5
6
peer "github.com/jbenet/go-ipfs/peer"
@@ -21,9 +20,8 @@ func newLedger(p peer.Peer, strategy strategyFunc) *ledger {
20
}
21
22
// ledger stores the data exchange relationship between two peers.
23
+// NOT threadsafe
24
type ledger struct {
25
- lock sync.RWMutex
26
-
25
// Partner is the remote Peer.
26
Partner peer.Peer
27
@@ -46,25 +44,16 @@ type ledger struct {
44
}
45
46
func (l *ledger) ShouldSend() bool {
49
- l.lock.Lock()
50
- defer l.lock.Unlock()
51
-
47
return l.Strategy(l)
48
}
49
50
func (l *ledger) SentBytes(n int) {
56
- l.lock.Lock()
57
- defer l.lock.Unlock()
58
-
51
l.exchangeCount++
52
l.lastExchange = time.Now()
53
l.Accounting.BytesSent += uint64(n)
54
}
55
56
func (l *ledger) ReceivedBytes(n int) {
65
- l.lock.Lock()
66
- defer l.lock.Unlock()
67
-
57
l.exchangeCount++
58
l.lastExchange = time.Now()
59
l.Accounting.BytesRecv += uint64(n)
@@ -72,22 +61,14 @@ func (l *ledger) ReceivedBytes(n int) {
61
62
// TODO: this needs to be different. We need timeouts.
63
func (l *ledger) Wants(k u.Key) {
75
- l.lock.Lock()
76
- defer l.lock.Unlock()
77
-
64
l.wantList[k] = struct{}{}
65
}
66
67
func (l *ledger) WantListContains(k u.Key) bool {
82
- l.lock.RLock()
83
- defer l.lock.RUnlock()
84
-
68
_, ok := l.wantList[k]
69
return ok
70
}
71
72
func (l *ledger) ExchangeCount() uint64 {
90
- l.lock.RLock()
91
- defer l.lock.RUnlock()
73
return l.exchangeCount
74
}
exchange/bitswap/strategy/ledger_test.go
-22
@@ -1,23 +1 @@
1
package strategy
2
-
3
-import (
4
- "sync"
5
- "testing"
6
-)
7
-
8
-func TestRaceConditions(t *testing.T) {
9
- const numberOfExpectedExchanges = 10000
10
- l := new(ledger)
11
- var wg sync.WaitGroup
12
- for i := 0; i < numberOfExpectedExchanges; i++ {
13
- wg.Add(1)
14
- go func() {
15
- defer wg.Done()
16
- l.ReceivedBytes(1)
17
- }()
18
- }
19
- wg.Wait()
20
- if l.ExchangeCount() != numberOfExpectedExchanges {
21
- t.Fail()
22
- }
23
-}
exchange/bitswap/strategy/strategy.go
+26
@@ -2,6 +2,7 @@ package strategy
2
3
import (
4
"errors"
5
+ "sync"
6
7
bsmsg "github.com/jbenet/go-ipfs/exchange/bitswap/message"
8
"github.com/jbenet/go-ipfs/peer"
@@ -26,6 +27,7 @@ func New(nice bool) Strategy {
27
}
28
29
type strategist struct {
30
+ lock sync.RWMutex
31
ledgerMap
32
strategyFunc
33
}
@@ -38,6 +40,9 @@ type peerKey u.Key
40
41
// Peers returns a list of peers
42
func (s *strategist) Peers() []peer.Peer {
43
+ s.lock.RLock()
44
+ defer s.lock.RUnlock()
45
+
46
response := make([]peer.Peer, 0)
47
for _, ledger := range s.ledgerMap {
48
response = append(response, ledger.Partner)
@@ -46,20 +51,32 @@ func (s *strategist) Peers() []peer.Peer {
51
}
52
53
func (s *strategist) BlockIsWantedByPeer(k u.Key, p peer.Peer) bool {
54
+ s.lock.RLock()
55
+ defer s.lock.RUnlock()
56
+
57
ledger := s.ledger(p)
58
return ledger.WantListContains(k)
59
}
60
61
func (s *strategist) ShouldSendBlockToPeer(k u.Key, p peer.Peer) bool {
62
+ s.lock.RLock()
63
+ defer s.lock.RUnlock()
64
+
65
ledger := s.ledger(p)
66
return ledger.ShouldSend()
67
}
68
69
func (s *strategist) Seed(int64) {
70
+ s.lock.Lock()
71
+ defer s.lock.Unlock()
72
+
73
// TODO
74
}
75
76
func (s *strategist) MessageReceived(p peer.Peer, m bsmsg.BitSwapMessage) error {
77
+ s.lock.Lock()
78
+ defer s.lock.Unlock()
79
+
80
// TODO find a more elegant way to handle this check
81
if p == nil {
82
return errors.New("Strategy received nil peer")
@@ -85,6 +102,9 @@ func (s *strategist) MessageReceived(p peer.Peer, m bsmsg.BitSwapMessage) error
102
// send happen atomically
103
104
func (s *strategist) MessageSent(p peer.Peer, m bsmsg.BitSwapMessage) error {
105
+ s.lock.Lock()
106
+ defer s.lock.Unlock()
107
+
108
l := s.ledger(p)
109
for _, block := range m.Blocks() {
110
l.SentBytes(len(block.Data))
@@ -96,10 +116,16 @@ func (s *strategist) MessageSent(p peer.Peer, m bsmsg.BitSwapMessage) error {
116
}
117
118
func (s *strategist) NumBytesSentTo(p peer.Peer) uint64 {
119
+ s.lock.RLock()
120
+ defer s.lock.RUnlock()
121
+
122
return s.ledger(p).Accounting.BytesSent
123
}
124
125
func (s *strategist) NumBytesReceivedFrom(p peer.Peer) uint64 {
126
+ s.lock.RLock()
127
+ defer s.lock.RUnlock()
128
+
129
return s.ledger(p).Accounting.BytesRecv
130
}
131