fix(bitswap.decision.Engine) enqueue only the freshest messages
Before, the engine worker would pop a task and block on send to the bitswap worker even if the bitswap worker wasn't to receive. Since the task could have been invalidated during this blocking send, a small number of stale (already acquired) blocks would be send to partners. Now, tasks are only popped off of the queue when bitswap is ready to send them over the wire. This is accomplished by removing the outboxChanBuffer and implementing a two-phase communication sequence.
Brian Tiger Chow committed
Jan 18, 2015 at 23:37 UTC
60e288ed4b44e21d3e3911eda8f4dc0d18662291
3 files changed
+81
-49
exchange/bitswap/bitswap.go
+7
-4
@@ -277,10 +277,13 @@ func (bs *bitswap) taskWorker(ctx context.Context) {
277
case <-ctx.Done():
278
log.Debugf("exiting")
279
return
280
- case envelope := <-bs.engine.Outbox():
281
- log.Debugf("message to %s sending...", envelope.Peer)
282
- bs.send(ctx, envelope.Peer, envelope.Message)
283
- log.Debugf("message to %s sent", envelope.Peer)
280
+ case nextEnvelope := <-bs.engine.Outbox():
281
+ select {
282
+ case <-ctx.Done():
283
+ return
284
+ case envelope := <-nextEnvelope:
285
+ bs.send(ctx, envelope.Peer, envelope.Message)
286
+ }
287
}
288
}
289
}
exchange/bitswap/decision/engine.go
+39
-27
@@ -44,7 +44,8 @@ import (
44
var log = eventlog.Logger("engine")
45
46
const (
47
- sizeOutboxChan = 4
47
+ // outboxChanBuffer must be 0 to prevent stale messages from being sent
48
+ outboxChanBuffer = 0
49
)
50
51
// Envelope contains a message for a Peer
@@ -68,8 +69,9 @@ type Engine struct {
69
// that case, no lock would be required.
70
workSignal chan struct{}
71
71
- // outbox contains outgoing messages to peers
72
- outbox chan Envelope
72
+ // outbox contains outgoing messages to peers. This is owned by the
73
+ // taskWorker goroutine
74
+ outbox chan (<-chan Envelope)
75
76
bs bstore.Blockstore
77
@@ -83,7 +85,7 @@ func NewEngine(ctx context.Context, bs bstore.Blockstore) *Engine {
85
ledgerMap: make(map[peer.ID]*ledger),
86
bs: bs,
87
peerRequestQueue: newPRQ(),
86
- outbox: make(chan Envelope, sizeOutboxChan),
88
+ outbox: make(chan (<-chan Envelope), outboxChanBuffer),
89
workSignal: make(chan struct{}),
90
}
91
go e.taskWorker(ctx)
@@ -91,45 +93,55 @@ func NewEngine(ctx context.Context, bs bstore.Blockstore) *Engine {
93
}
94
95
func (e *Engine) taskWorker(ctx context.Context) {
94
- log := log.Prefix("bitswap.Engine.taskWorker")
96
+ defer close(e.outbox) // because taskWorker uses the channel exclusively
97
+ for {
98
+ oneTimeUse := make(chan Envelope, 1) // buffer to prevent blocking
99
+ select {
100
+ case <-ctx.Done():
101
+ return
102
+ case e.outbox <- oneTimeUse:
103
+ }
104
+ // receiver is ready for an outoing envelope. let's prepare one. first,
105
+ // we must acquire a task from the PQ...
106
+ envelope, err := e.nextEnvelope(ctx)
107
+ if err != nil {
108
+ close(oneTimeUse)
109
+ return // ctx cancelled
110
+ }
111
+ oneTimeUse <- *envelope // buffered. won't block
112
+ close(oneTimeUse)
113
+ }
114
+}
115
+
116
+// nextEnvelope runs in the taskWorker goroutine. Returns an error if the
117
+// context is cancelled before the next Envelope can be created.
118
+func (e *Engine) nextEnvelope(ctx context.Context) (*Envelope, error) {
119
for {
120
nextTask := e.peerRequestQueue.Pop()
97
- if nextTask == nil {
98
- // No tasks in the list?
99
- // Wait until there are!
121
+ for nextTask == nil {
122
select {
123
case <-ctx.Done():
102
- log.Debugf("exiting: %s", ctx.Err())
103
- return
124
+ return nil, ctx.Err()
125
case <-e.workSignal:
105
- log.Debugf("woken up")
126
+ nextTask = e.peerRequestQueue.Pop()
127
}
107
- continue
128
}
109
- log := log.Prefix("%s", nextTask)
110
- log.Debugf("processing")
129
+
130
+ // with a task in hand, we're ready to prepare the envelope...
131
132
block, err := e.bs.Get(nextTask.Entry.Key)
133
if err != nil {
114
- log.Warning("engine: task exists to send block, but block is not in blockstore")
134
continue
135
}
117
- // construct message here so we can make decisions about any additional
118
- // information we may want to include at this time.
119
- m := bsmsg.New()
136
+
137
+ m := bsmsg.New() // TODO: maybe add keys from our wantlist?
138
m.AddBlock(block)
121
- // TODO: maybe add keys from our wantlist?
122
- log.Debugf("sending...")
123
- select {
124
- case <-ctx.Done():
125
- return
126
- case e.outbox <- Envelope{Peer: nextTask.Target, Message: m}:
127
- log.Debugf("sent")
128
- }
139
+ return &Envelope{Peer: nextTask.Target, Message: m}, nil
140
}
141
}
142
132
-func (e *Engine) Outbox() <-chan Envelope {
143
+// Outbox returns a channel of one-time use Envelope channels.
144
+func (e *Engine) Outbox() <-chan (<-chan Envelope) {
145
return e.outbox
146
}
147
exchange/bitswap/decision/engine_test.go
+35
-18
@@ -1,6 +1,8 @@
1
package decision
2
3
import (
4
+ "errors"
5
+ "fmt"
6
"math"
7
"strings"
8
"sync"
@@ -104,7 +106,8 @@ func TestOutboxClosedWhenEngineClosed(t *testing.T) {
106
var wg sync.WaitGroup
107
wg.Add(1)
108
go func() {
107
- for _ = range e.Outbox() {
109
+ for nextEnvelope := range e.Outbox() {
110
+ <-nextEnvelope
111
}
112
wg.Done()
113
}()
@@ -116,6 +119,10 @@ func TestOutboxClosedWhenEngineClosed(t *testing.T) {
119
}
120
121
func TestPartnerWantsThenCancels(t *testing.T) {
122
+ numRounds := 10
123
+ if testing.Short() {
124
+ numRounds = 1
125
+ }
126
alphabet := strings.Split("abcdefghijklmnopqrstuvwxyz", "")
127
vowels := strings.Split("aeiou", "")
128
@@ -129,23 +136,31 @@ func TestPartnerWantsThenCancels(t *testing.T) {
136
},
137
}
138
132
- for _, testcase := range testcases {
133
- set := testcase[0]
134
- cancels := testcase[1]
135
- keeps := stringsComplement(set, cancels)
136
-
137
- bs := blockstore.NewBlockstore(dssync.MutexWrap(ds.NewMapDatastore()))
138
- e := NewEngine(context.Background(), bs)
139
- partner := testutil.RandPeerIDFatal(t)
140
- for _, letter := range set {
141
- block := blocks.NewBlock([]byte(letter))
142
- bs.Put(block)
139
+ bs := blockstore.NewBlockstore(dssync.MutexWrap(ds.NewMapDatastore()))
140
+ for _, letter := range alphabet {
141
+ block := blocks.NewBlock([]byte(letter))
142
+ if err := bs.Put(block); err != nil {
143
+ t.Fatal(err)
144
}
144
- partnerWants(e, set, partner)
145
- partnerCancels(e, cancels, partner)
146
- assertPoppedInOrder(t, e, keeps)
145
}
146
147
+ for i := 0; i < numRounds; i++ {
148
+ for _, testcase := range testcases {
149
+ set := testcase[0]
150
+ cancels := testcase[1]
151
+ keeps := stringsComplement(set, cancels)
152
+
153
+ e := NewEngine(context.Background(), bs)
154
+ partner := testutil.RandPeerIDFatal(t)
155
+
156
+ partnerWants(e, set, partner)
157
+ partnerCancels(e, cancels, partner)
158
+ if err := checkHandledInOrder(t, e, keeps); err != nil {
159
+ t.Logf("run #%d of %d", i, numRounds)
160
+ t.Fatal(err)
161
+ }
162
+ }
163
+ }
164
}
165
166
func partnerWants(e *Engine, keys []string, partner peer.ID) {
@@ -166,15 +181,17 @@ func partnerCancels(e *Engine, keys []string, partner peer.ID) {
181
e.MessageReceived(partner, cancels)
182
}
183
169
-func assertPoppedInOrder(t *testing.T, e *Engine, keys []string) {
184
+func checkHandledInOrder(t *testing.T, e *Engine, keys []string) error {
185
for _, k := range keys {
171
- envelope := <-e.Outbox()
186
+ next := <-e.Outbox()
187
+ envelope := <-next
188
received := envelope.Message.Blocks()[0]
189
expected := blocks.NewBlock([]byte(k))
190
if received.Key() != expected.Key() {
175
- t.Fatal("received", string(received.Data), "expected", string(expected.Data))
191
+ return errors.New(fmt.Sprintln("received", string(received.Data), "expected", string(expected.Data)))
192
}
193
}
194
+ return nil
195
}
196
197
func stringsComplement(set, subset []string) []string {