bitswap virtual test net code should send messages in order
License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>
Jeromy committed
Feb 12, 2018 at 23:40 UTC
0dd0f252710eca9c6de191b39fc2932eca2b9abf
1 file changed
+55
-6
exchange/bitswap/testnet/virtual.go
+55
-6
@@ -4,6 +4,7 @@ import (
4
"context"
5
"errors"
6
"sync"
7
+ "time"
8
9
bsmsg "github.com/ipfs/go-ipfs/exchange/bitswap/message"
10
bsnet "github.com/ipfs/go-ipfs/exchange/bitswap/network"
@@ -22,7 +23,7 @@ var log = logging.Logger("bstestnet")
23
24
func VirtualNetwork(rs mockrouting.Server, d delay.D) Network {
25
return &network{
25
- clients: make(map[peer.ID]bsnet.Receiver),
26
+ clients: make(map[peer.ID]*receiverQueue),
27
delay: d,
28
routingserver: rs,
29
conns: make(map[string]struct{}),
@@ -31,12 +32,28 @@ func VirtualNetwork(rs mockrouting.Server, d delay.D) Network {
32
33
type network struct {
34
mu sync.Mutex
34
- clients map[peer.ID]bsnet.Receiver
35
+ clients map[peer.ID]*receiverQueue
36
routingserver mockrouting.Server
37
delay delay.D
38
conns map[string]struct{}
39
}
40
41
+type message struct {
42
+ from peer.ID
43
+ msg bsmsg.BitSwapMessage
44
+ shouldSend time.Time
45
+}
46
+
47
+// receiverQueue queues up a set of messages to be sent, and sends them *in
48
+// order* with their delays respected as much as sending them in order allows
49
+// for
50
+type receiverQueue struct {
51
+ receiver bsnet.Receiver
52
+ queue []*message
53
+ active bool
54
+ lk sync.Mutex
55
+}
56
+
57
func (n *network) Adapter(p testutil.Identity) bsnet.BitSwapNetwork {
58
n.mu.Lock()
59
defer n.mu.Unlock()
@@ -46,7 +63,7 @@ func (n *network) Adapter(p testutil.Identity) bsnet.BitSwapNetwork {
63
network: n,
64
routing: n.routingserver.Client(p),
65
}
49
- n.clients[p.ID()] = client
66
+ n.clients[p.ID()] = &receiverQueue{receiver: client}
67
return client
68
}
69
@@ -64,7 +81,7 @@ func (n *network) SendMessage(
81
ctx context.Context,
82
from peer.ID,
83
to peer.ID,
67
- message bsmsg.BitSwapMessage) error {
84
+ mes bsmsg.BitSwapMessage) error {
85
86
n.mu.Lock()
87
defer n.mu.Unlock()
@@ -77,7 +94,12 @@ func (n *network) SendMessage(
94
// nb: terminate the context since the context wouldn't actually be passed
95
// over the network in a real scenario
96
80
- go n.deliver(receiver, from, message)
97
+ msg := &message{
98
+ from: from,
99
+ msg: mes,
100
+ shouldSend: time.Now().Add(n.delay.Get()),
101
+ }
102
+ receiver.enqueue(msg)
103
104
return nil
105
}
@@ -191,11 +213,38 @@ func (nc *networkClient) ConnectTo(_ context.Context, p peer.ID) error {
213
214
// TODO: add handling for disconnects
215
194
- otherClient.PeerConnected(nc.local)
216
+ otherClient.receiver.PeerConnected(nc.local)
217
nc.Receiver.PeerConnected(p)
218
return nil
219
}
220
221
+func (rq *receiverQueue) enqueue(m *message) {
222
+ rq.lk.Lock()
223
+ defer rq.lk.Unlock()
224
+ rq.queue = append(rq.queue, m)
225
+ if !rq.active {
226
+ rq.active = true
227
+ go rq.process()
228
+ }
229
+}
230
+
231
+func (rq *receiverQueue) process() {
232
+ for {
233
+ rq.lk.Lock()
234
+ if len(rq.queue) == 0 {
235
+ rq.active = false
236
+ rq.lk.Unlock()
237
+ return
238
+ }
239
+ m := rq.queue[0]
240
+ rq.queue = rq.queue[1:]
241
+ rq.lk.Unlock()
242
+
243
+ time.Sleep(time.Until(m.shouldSend))
244
+ rq.receiver.ReceiveMessage(context.TODO(), m.from, m.msg)
245
+ }
246
+}
247
+
248
func tagForPeers(a, b peer.ID) string {
249
if a < b {
250
return string(a + b)