1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
|
// Copyright 2018 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// +build withserver
package stream
import (
"bytes"
"context"
"errors"
"fmt"
"io"
"sync"
"testing"
"time"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/protocols"
"github.com/ethereum/go-ethereum/p2p/simulations"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/swarm/log"
"github.com/ethereum/go-ethereum/swarm/network/simulation"
"github.com/ethereum/go-ethereum/swarm/state"
"github.com/ethereum/go-ethereum/swarm/storage"
)
/*
The tests in this file need to be executed with
-tags=withserver
Also, they will stall if executed stand-alone, because they wait
for the visualization frontend to send a POST /runsim message.
*/
//setup the sim, evaluate nodeCount and chunkCount and create the sim
func setupSim(serviceMap map[string]simulation.ServiceFunc) (int, int, *simulation.Simulation) {
nodeCount := *nodes
chunkCount := *chunks
if nodeCount == 0 || chunkCount == 0 {
nodeCount = 32
chunkCount = 1
}
//setup the simulation with server, which means the sim won't run
//until it receives a POST /runsim from the frontend
sim := simulation.New(serviceMap).WithServer(":8888")
return nodeCount, chunkCount, sim
}
//watch for disconnections and wait for healthy
func watchSim(sim *simulation.Simulation) (context.Context, context.CancelFunc) {
ctx, cancelSimRun := context.WithTimeout(context.Background(), 1*time.Minute)
if _, err := sim.WaitTillHealthy(ctx); err != nil {
panic(err)
}
disconnections := sim.PeerEvents(
context.Background(),
sim.NodeIDs(),
simulation.NewPeerEventsFilter().Drop(),
)
go func() {
for d := range disconnections {
log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
panic("unexpected disconnect")
cancelSimRun()
}
}()
return ctx, cancelSimRun
}
//This test requests bogus hashes into the network
func TestNonExistingHashesWithServer(t *testing.T) {
nodeCount, _, sim := setupSim(retrievalSimServiceMap)
defer sim.Close()
err := sim.UploadSnapshot(fmt.Sprintf("testing/snapshot_%d.json", nodeCount))
if err != nil {
panic(err)
}
ctx, cancelSimRun := watchSim(sim)
defer cancelSimRun()
//in order to get some meaningful visualization, it is beneficial
//to define a minimum duration of this test
testDuration := 20 * time.Second
result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error {
//check on the node's FileStore (netstore)
id := sim.Net.GetRandomUpNode().ID()
item, ok := sim.NodeItem(id, bucketKeyFileStore)
if !ok {
t.Fatalf("No filestore")
}
fileStore := item.(*storage.FileStore)
//create a bogus hash
fakeHash := storage.GenerateRandomChunk(1000).Address()
//try to retrieve it - will propagate RetrieveRequestMsg into the network
reader, _ := fileStore.Retrieve(context.TODO(), fakeHash)
if _, err := reader.Size(ctx, nil); err != nil {
log.Debug("expected error for non-existing chunk")
}
//sleep so that the frontend can have something to display
time.Sleep(testDuration)
return nil
})
if result.Error != nil {
sendSimTerminatedEvent(sim)
t.Fatal(result.Error)
}
sendSimTerminatedEvent(sim)
}
//send a termination event to the frontend
func sendSimTerminatedEvent(sim *simulation.Simulation) {
evt := &simulations.Event{
Type: EventTypeSimTerminated,
Control: false,
}
sim.Net.Events().Send(evt)
}
//This test is the same as the snapshot sync test,
//but with a HTTP server
//It also sends some custom events so that the frontend
//can visualize messages like SendOfferedMsg, WantedHashesMsg, DeliveryMsg
func TestSnapshotSyncWithServer(t *testing.T) {
//t.Skip("temporarily disabled as simulations.WaitTillHealthy cannot be trusted")
//define a wrapper object to be able to pass around data
wrapper := &netWrapper{}
nodeCount := *nodes
chunkCount := *chunks
if nodeCount == 0 || chunkCount == 0 {
nodeCount = 32
chunkCount = 1
}
log.Info(fmt.Sprintf("Running the simulation with %d nodes and %d chunks", nodeCount, chunkCount))
sim := simulation.New(map[string]simulation.ServiceFunc{
"streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) {
addr, netStore, delivery, clean, err := newNetStoreAndDeliveryWithRequestFunc(ctx, bucket, dummyRequestFromPeers)
if err != nil {
return nil, nil, err
}
r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{
Retrieval: RetrievalDisabled,
Syncing: SyncingAutoSubscribe,
SyncUpdateDelay: 3 * time.Second,
}, nil)
tr := &testRegistry{
Registry: r,
w: wrapper,
}
bucket.Store(bucketKeyRegistry, tr)
cleanup = func() {
tr.Close()
clean()
}
return tr, cleanup, nil
},
}).WithServer(":8888") //start with the HTTP server
nodeCount, chunkCount, sim := setupSim(simServiceMap)
defer sim.Close()
log.Info("Initializing test config")
conf := &synctestConfig{}
//map of discover ID to indexes of chunks expected at that ID
conf.idToChunksMap = make(map[enode.ID][]int)
//map of overlay address to discover ID
conf.addrToIDMap = make(map[string]enode.ID)
//array where the generated chunk hashes will be stored
conf.hashes = make([]storage.Address, 0)
//pass the network to the wrapper object
wrapper.setNetwork(sim.Net)
err := sim.UploadSnapshot(fmt.Sprintf("testing/snapshot_%d.json", nodeCount))
if err != nil {
panic(err)
}
ctx, cancelSimRun := watchSim(sim)
defer cancelSimRun()
//run the sim
result := runSim(conf, ctx, sim, chunkCount)
//send terminated event
evt := &simulations.Event{
Type: EventTypeSimTerminated,
Control: false,
}
go sim.Net.Events().Send(evt)
if result.Error != nil {
panic(result.Error)
}
log.Info("Simulation ended")
}
//testRegistry embeds registry
//it allows to replace the protocol run function
type testRegistry struct {
*Registry
w *netWrapper
}
//Protocols replaces the protocol's run function
func (tr *testRegistry) Protocols() []p2p.Protocol {
regProto := tr.Registry.Protocols()
//set the `stream` protocol's run function with the testRegistry's one
regProto[0].Run = tr.runProto
return regProto
}
//runProto is the new overwritten protocol's run function for this test
func (tr *testRegistry) runProto(p *p2p.Peer, rw p2p.MsgReadWriter) error {
//create a custom rw message ReadWriter
testRw := &testMsgReadWriter{
MsgReadWriter: rw,
Peer: p,
w: tr.w,
Registry: tr.Registry,
}
//now run the actual upper layer `Registry`'s protocol function
return tr.runProtocol(p, testRw)
}
//testMsgReadWriter is a custom rw
//it will allow us to re-use the message twice
type testMsgReadWriter struct {
*Registry
p2p.MsgReadWriter
*p2p.Peer
w *netWrapper
}
//netWrapper wrapper object so we can pass data around
type netWrapper struct {
net *simulations.Network
}
//set the network to the wrapper for later use (used inside the custom rw)
func (w *netWrapper) setNetwork(n *simulations.Network) {
w.net = n
}
//get he network from the wrapper (used inside the custom rw)
func (w *netWrapper) getNetwork() *simulations.Network {
return w.net
}
// ReadMsg reads a message from the underlying MsgReadWriter and emits a
// "message received" event
//we do this because we are interested in the Payload of the message for custom use
//in this test, but messages can only be consumed once (stream io.Reader)
func (ev *testMsgReadWriter) ReadMsg() (p2p.Msg, error) {
//read the message from the underlying rw
msg, err := ev.MsgReadWriter.ReadMsg()
if err != nil {
return msg, err
}
//don't do anything with message codes we actually are not needing/reading
subCodes := []uint64{1, 2, 10}
found := false
for _, c := range subCodes {
if c == msg.Code {
found = true
}
}
//just return if not a msg code we are interested in
if !found {
return msg, nil
}
//we use a io.TeeReader so that we can read the message twice
//the Payload is a io.Reader, so if we read from it, the actual protocol handler
//cannot access it anymore.
//But we need that handler to be able to consume the message as normal,
//as if we would not do anything here with that message
var buf bytes.Buffer
tee := io.TeeReader(msg.Payload, &buf)
mcp := &p2p.Msg{
Code: msg.Code,
Size: msg.Size,
ReceivedAt: msg.ReceivedAt,
Payload: tee,
}
//assign the copy for later use
msg.Payload = &buf
//now let's look into the message
var wmsg protocols.WrappedMsg
err = mcp.Decode(&wmsg)
if err != nil {
log.Error(err.Error())
return msg, err
}
//create a new message from the code
val, ok := ev.Registry.GetSpec().NewMsg(mcp.Code)
if !ok {
return msg, errors.New(fmt.Sprintf("Invalid message code: %v", msg.Code))
}
//decode it
if err := rlp.DecodeBytes(wmsg.Payload, val); err != nil {
return msg, errors.New(fmt.Sprintf("Decoding error <= %v: %v", msg, err))
}
//now for every message type we are interested in, create a custom event and send it
var evt *simulations.Event
switch val := val.(type) {
case *OfferedHashesMsg:
evt = &simulations.Event{
Type: EventTypeChunkOffered,
Node: ev.w.getNetwork().GetNode(ev.ID()),
Control: false,
Data: val.Hashes,
}
case *WantedHashesMsg:
evt = &simulations.Event{
Type: EventTypeChunkWanted,
Node: ev.w.getNetwork().GetNode(ev.ID()),
Control: false,
}
case *ChunkDeliveryMsgSyncing:
evt = &simulations.Event{
Type: EventTypeChunkDelivered,
Node: ev.w.getNetwork().GetNode(ev.ID()),
Control: false,
Data: val.Addr.String(),
}
}
if evt != nil {
//send custom event to feed; frontend will listen to it and display
ev.w.getNetwork().Events().Send(evt)
}
return msg, nil
}
|