explicitly set bitswap message fullness
Jeromy committed
May 16, 2015 at 22:08 UTC
829b88420e0a50c9d03f6ef5689123baec8487fb
7 files changed
+38
-36
exchange/bitswap/bitswap.go
+1
-1
@@ -288,7 +288,7 @@ func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg
288
bs.dupBlocksRecvd++
289
}
290
bs.counterLk.Unlock()
291
- log.Debugf("got block %s from %s", block, p)
291
+ log.Debugf("got block %s from %s (%d,%d)", block, p, bs.blocksRecvd, bs.dupBlocksRecvd)
292
293
hasBlockCtx, cancel := context.WithTimeout(ctx, hasBlockTimeout)
294
if err := bs.HasBlock(hasBlockCtx, block); err != nil {
exchange/bitswap/bitswap_test.go
-1
@@ -163,7 +163,6 @@ func PerformDistributionTest(t *testing.T, numInstances, numBlocks int) {
163
}
164
for _ = range outch {
165
}
166
- log.Error("DONE")
166
}(inst)
167
}
168
wg.Wait()
exchange/bitswap/decision/engine_test.go
+4
-4
@@ -41,7 +41,7 @@ func TestConsistentAccounting(t *testing.T) {
41
// Send messages from Ernie to Bert
42
for i := 0; i < 1000; i++ {
43
44
- m := message.New()
44
+ m := message.New(false)
45
content := []string{"this", "is", "message", "i"}
46
m.AddBlock(blocks.NewBlock([]byte(strings.Join(content, " "))))
47
@@ -73,7 +73,7 @@ func TestPeerIsAddedToPeersWhenMessageReceivedOrSent(t *testing.T) {
73
sanfrancisco := newEngine(ctx, "sf")
74
seattle := newEngine(ctx, "sea")
75
76
- m := message.New()
76
+ m := message.New(true)
77
78
sanfrancisco.Engine.MessageSent(seattle.Peer, m)
79
seattle.Engine.MessageReceived(sanfrancisco.Peer, m)
@@ -164,7 +164,7 @@ func TestPartnerWantsThenCancels(t *testing.T) {
164
}
165
166
func partnerWants(e *Engine, keys []string, partner peer.ID) {
167
- add := message.New()
167
+ add := message.New(false)
168
for i, letter := range keys {
169
block := blocks.NewBlock([]byte(letter))
170
add.AddEntry(block.Key(), math.MaxInt32-i)
@@ -173,7 +173,7 @@ func partnerWants(e *Engine, keys []string, partner peer.ID) {
173
}
174
175
func partnerCancels(e *Engine, keys []string, partner peer.ID) {
176
- cancels := message.New()
176
+ cancels := message.New(false)
177
for _, k := range keys {
178
block := blocks.NewBlock([]byte(k))
179
cancels.Cancel(block.Key())
exchange/bitswap/message/message.go
+6
-16
@@ -31,12 +31,7 @@ type BitSwapMessage interface {
31
32
Empty() bool
33
34
- // Sets whether or not the contained wantlist represents the entire wantlist
35
- // true = full wantlist
36
- // false = wantlist 'patch'
37
- // default: true
38
- SetFull(isFull bool)
39
-
34
+ // A full wantlist is an authoritative copy, a 'non-full' wantlist is a patch-set
35
Full() bool
36
37
AddBlock(*blocks.Block)
@@ -56,15 +51,15 @@ type impl struct {
51
blocks map[u.Key]*blocks.Block
52
}
53
59
-func New() BitSwapMessage {
60
- return newMsg()
54
+func New(full bool) BitSwapMessage {
55
+ return newMsg(full)
56
}
57
63
-func newMsg() *impl {
58
+func newMsg(full bool) *impl {
59
return &impl{
60
blocks: make(map[u.Key]*blocks.Block),
61
wantlist: make(map[u.Key]Entry),
67
- full: true,
62
+ full: full,
63
}
64
}
65
@@ -74,8 +69,7 @@ type Entry struct {
69
}
70
71
func newMessageFromProto(pbm pb.Message) BitSwapMessage {
77
- m := newMsg()
78
- m.SetFull(pbm.GetWantlist().GetFull())
72
+ m := newMsg(pbm.GetWantlist().GetFull())
73
for _, e := range pbm.GetWantlist().GetEntries() {
74
m.addEntry(u.Key(e.GetBlock()), int(e.GetPriority()), e.GetCancel())
75
}
@@ -86,10 +80,6 @@ func newMessageFromProto(pbm pb.Message) BitSwapMessage {
80
return m
81
}
82
89
-func (m *impl) SetFull(full bool) {
90
- m.full = full
91
-}
92
-
83
func (m *impl) Full() bool {
84
return m.full
85
}
exchange/bitswap/message/message_test.go
+7
-7
@@ -13,7 +13,7 @@ import (
13
14
func TestAppendWanted(t *testing.T) {
15
const str = "foo"
16
- m := New()
16
+ m := New(true)
17
m.AddEntry(u.Key(str), 1)
18
19
if !wantlistContains(m.ToProto().GetWantlist(), str) {
@@ -44,7 +44,7 @@ func TestAppendBlock(t *testing.T) {
44
strs = append(strs, "Celeritas")
45
strs = append(strs, "Incendia")
46
47
- m := New()
47
+ m := New(true)
48
for _, str := range strs {
49
block := blocks.NewBlock([]byte(str))
50
m.AddBlock(block)
@@ -61,7 +61,7 @@ func TestAppendBlock(t *testing.T) {
61
62
func TestWantlist(t *testing.T) {
63
keystrs := []string{"foo", "bar", "baz", "bat"}
64
- m := New()
64
+ m := New(true)
65
for _, s := range keystrs {
66
m.AddEntry(u.Key(s), 1)
67
}
@@ -84,7 +84,7 @@ func TestWantlist(t *testing.T) {
84
85
func TestCopyProtoByValue(t *testing.T) {
86
const str = "foo"
87
- m := New()
87
+ m := New(true)
88
protoBeforeAppend := m.ToProto()
89
m.AddEntry(u.Key(str), 1)
90
if wantlistContains(protoBeforeAppend.GetWantlist(), str) {
@@ -93,7 +93,7 @@ func TestCopyProtoByValue(t *testing.T) {
93
}
94
95
func TestToNetFromNetPreservesWantList(t *testing.T) {
96
- original := New()
96
+ original := New(true)
97
original.AddEntry(u.Key("M"), 1)
98
original.AddEntry(u.Key("B"), 1)
99
original.AddEntry(u.Key("D"), 1)
@@ -124,7 +124,7 @@ func TestToNetFromNetPreservesWantList(t *testing.T) {
124
125
func TestToAndFromNetMessage(t *testing.T) {
126
127
- original := New()
127
+ original := New(true)
128
original.AddBlock(blocks.NewBlock([]byte("W")))
129
original.AddBlock(blocks.NewBlock([]byte("E")))
130
original.AddBlock(blocks.NewBlock([]byte("F")))
@@ -172,7 +172,7 @@ func contains(strs []string, x string) bool {
172
173
func TestDuplicates(t *testing.T) {
174
b := blocks.NewBlock([]byte("foo"))
175
- msg := New()
175
+ msg := New(true)
176
177
msg.AddEntry(b.Key(), 1)
178
msg.AddEntry(b.Key(), 1)
exchange/bitswap/testnet/network_test.go
+2
-2
@@ -31,7 +31,7 @@ func TestSendMessageAsyncButWaitForResponse(t *testing.T) {
31
fromWaiter peer.ID,
32
msgFromWaiter bsmsg.BitSwapMessage) {
33
34
- msgToWaiter := bsmsg.New()
34
+ msgToWaiter := bsmsg.New(true)
35
msgToWaiter.AddBlock(blocks.NewBlock([]byte(expectedStr)))
36
waiter.SendMessage(ctx, fromWaiter, msgToWaiter)
37
}))
@@ -55,7 +55,7 @@ func TestSendMessageAsyncButWaitForResponse(t *testing.T) {
55
}
56
}))
57
58
- messageSentAsync := bsmsg.New()
58
+ messageSentAsync := bsmsg.New(true)
59
messageSentAsync.AddBlock(blocks.NewBlock([]byte("data")))
60
errSending := waiter.SendMessage(
61
context.Background(), responderPeer.ID(), messageSentAsync)
exchange/bitswap/wantmanager.go
+18
-5
@@ -2,6 +2,7 @@ package bitswap
2
3
import (
4
"sync"
5
+ "time"
6
7
context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
8
engine "github.com/ipfs/go-ipfs/exchange/bitswap/decision"
@@ -94,9 +95,8 @@ func (pm *WantManager) SendBlock(ctx context.Context, env *engine.Envelope) {
95
// throughout the network stack
96
defer env.Sent()
97
97
- msg := bsmsg.New()
98
+ msg := bsmsg.New(false)
99
msg.AddBlock(env.Block)
99
- msg.SetFull(false)
100
err := pm.network.SendMessage(ctx, env.Peer, msg)
101
if err != nil {
102
log.Error(err)
@@ -113,11 +113,10 @@ func (pm *WantManager) startPeerHandler(p peer.ID) *msgQueue {
113
mq := newMsgQueue(p)
114
115
// new peer, we will want to give them our full wantlist
116
- fullwantlist := bsmsg.New()
116
+ fullwantlist := bsmsg.New(true)
117
for _, e := range pm.wl.Entries() {
118
fullwantlist.AddEntry(e.Key, e.Priority)
119
}
120
- fullwantlist.SetFull(true)
120
mq.out = fullwantlist
121
mq.work <- struct{}{}
122
@@ -180,6 +179,7 @@ func (pm *WantManager) Disconnected(p peer.ID) {
179
180
// TODO: use goprocess here once i trust it
181
func (pm *WantManager) Run() {
182
+ tock := time.NewTicker(rebroadcastDelay.Get())
183
for {
184
select {
185
case entries := <-pm.incoming:
@@ -198,6 +198,19 @@ func (pm *WantManager) Run() {
198
p.addMessage(entries)
199
}
200
201
+ case <-tock.C:
202
+ // resend entire wantlist every so often (REALLY SHOULDNT BE NECESSARY)
203
+ var es []*bsmsg.Entry
204
+ for _, e := range pm.wl.Entries() {
205
+ es = append(es, &bsmsg.Entry{Entry: e})
206
+ }
207
+ for _, p := range pm.peers {
208
+ p.outlk.Lock()
209
+ p.out = bsmsg.New(true)
210
+ p.outlk.Unlock()
211
+
212
+ p.addMessage(es)
213
+ }
214
case p := <-pm.connect:
215
pm.startPeerHandler(p)
216
case p := <-pm.disconnect:
@@ -230,7 +243,7 @@ func (mq *msgQueue) addMessage(entries []*bsmsg.Entry) {
243
// if we have no message held, or the one we are given is full
244
// overwrite the one we are holding
245
if mq.out == nil {
233
- mq.out = bsmsg.New()
246
+ mq.out = bsmsg.New(false)
247
}
248
249
// TODO: add a msg.Combine(...) method