update comments and reintroduce test
Jeromy committed
May 12, 2015 at 23:50 UTC
8443b99c1d8060f626079114f0bea9f0e2784a2f
2 files changed
+40
-13
exchange/bitswap/bitswap_test.go
+23
@@ -13,6 +13,7 @@ import (
13
blocks "github.com/ipfs/go-ipfs/blocks"
14
blocksutil "github.com/ipfs/go-ipfs/blocks/blocksutil"
15
tn "github.com/ipfs/go-ipfs/exchange/bitswap/testnet"
16
+ p2ptestutil "github.com/ipfs/go-ipfs/p2p/test/util"
17
mockrouting "github.com/ipfs/go-ipfs/routing/mock"
18
delay "github.com/ipfs/go-ipfs/thirdparty/delay"
19
u "github.com/ipfs/go-ipfs/util"
@@ -35,6 +36,28 @@ func TestClose(t *testing.T) {
36
bitswap.Exchange.GetBlock(context.Background(), block.Key())
37
}
38
39
+func TestProviderForKeyButNetworkCannotFind(t *testing.T) { // TODO revisit this
40
+
41
+ rs := mockrouting.NewServer()
42
+ net := tn.VirtualNetwork(rs, delay.Fixed(kNetworkDelay))
43
+ g := NewTestSessionGenerator(net)
44
+ defer g.Close()
45
+
46
+ block := blocks.NewBlock([]byte("block"))
47
+ pinfo := p2ptestutil.RandTestBogusIdentityOrFatal(t)
48
+ rs.Client(pinfo).Provide(context.Background(), block.Key()) // but not on network
49
+
50
+ solo := g.Next()
51
+ defer solo.Exchange.Close()
52
+
53
+ ctx, _ := context.WithTimeout(context.Background(), time.Nanosecond)
54
+ _, err := solo.Exchange.GetBlock(ctx, block.Key())
55
+
56
+ if err != context.DeadlineExceeded {
57
+ t.Fatal("Expected DeadlineExceeded error")
58
+ }
59
+}
60
+
61
func TestGetBlockFromPeerAfterPeerAnnounces(t *testing.T) {
62
63
net := tn.VirtualNetwork(mockrouting.NewServer(), delay.Fixed(kNetworkDelay))
exchange/bitswap/peermanager.go
+17
-13
@@ -46,8 +46,8 @@ type cancellation struct {
46
type msgQueue struct {
47
p peer.ID
48
49
- lk sync.Mutex
50
- wlmsg bsmsg.BitSwapMessage
49
+ outlk sync.Mutex
50
+ out bsmsg.BitSwapMessage
51
52
work chan struct{}
53
done chan struct{}
@@ -106,11 +106,11 @@ func (pm *PeerManager) runQueue(mq *msgQueue) {
106
// TODO: cant connect, what now?
107
}
108
109
- // grab messages from queue
110
- mq.lk.Lock()
111
- wlm := mq.wlmsg
112
- mq.wlmsg = nil
113
- mq.lk.Unlock()
109
+ // grab outgoin message
110
+ mq.outlk.Lock()
111
+ wlm := mq.out
112
+ mq.out = nil
113
+ mq.outlk.Unlock()
114
115
if wlm != nil && !wlm.Empty() {
116
// send wantlist updates
@@ -178,26 +178,30 @@ func (pm *PeerManager) Run(ctx context.Context) {
178
}
179
180
func (mq *msgQueue) addMessage(msg bsmsg.BitSwapMessage) {
181
- mq.lk.Lock()
181
+ mq.outlk.Lock()
182
defer func() {
183
- mq.lk.Unlock()
183
+ mq.outlk.Unlock()
184
select {
185
case mq.work <- struct{}{}:
186
default:
187
}
188
}()
189
190
- if mq.wlmsg == nil || msg.Full() {
191
- mq.wlmsg = msg
190
+ // if we have no message held, or the one we are given is full
191
+ // overwrite the one we are holding
192
+ if mq.out == nil || msg.Full() {
193
+ mq.out = msg
194
return
195
}
196
197
// TODO: add a msg.Combine(...) method
198
+ // otherwise, combine the one we are holding with the
199
+ // one passed in
200
for _, e := range msg.Wantlist() {
201
if e.Cancel {
198
- mq.wlmsg.Cancel(e.Key)
202
+ mq.out.Cancel(e.Key)
203
} else {
200
- mq.wlmsg.AddEntry(e.Key, e.Priority)
204
+ mq.out.AddEntry(e.Key, e.Priority)
205
}
206
}
207
}