mirror
/
go-ethereum
mirror of https://github.com/ethereum/go-ethereum
2
0
Fork 1
Code Issues Releases Wiki Activity

swarm/network/simulation: use simulations.Event instead p2p.PeerEvent (#18098)

Browse Source
pull/18127/head
Janoš Guljaš 6 years ago committed by Viktor Trón
parent b91766fe6d
commit 324027640b
6 changed files with 101 additions and 55 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 114
      swarm/network/simulation/events.go
  2. 13
      swarm/network/simulation/example_test.go
  3. 8
      swarm/network/stream/delivery_test.go
  4. 5
      swarm/network/stream/intervals_test.go
  5. 11
      swarm/network/stream/snapshot_sync_test.go
  6. 5
      swarm/network/stream/syncer_test.go

114
swarm/network/simulation/events.go
Unescape View File

@ -20,16 +20,18 @@ import (
"context"
"sync"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/simulations"
)
// PeerEvent is the type of the channel returned by Simulation.PeerEvents.
type PeerEvent struct {
// NodeID is the ID of node that the event is caught on.
NodeID enode.ID
// PeerID is the ID of the peer node that the event is caught on.
PeerID enode.ID
// Event is the event that is caught.
Event *p2p.PeerEvent
Event *simulations.Event
// Error is the error that may have happened during event watching.
Error error
}
@ -37,9 +39,13 @@ type PeerEvent struct {
// PeerEventsFilter defines a filter on PeerEvents to exclude messages with
// defined properties. Use PeerEventsFilter methods to set required options.
type PeerEventsFilter struct {
t *p2p.PeerEventType
protocol *string
msgCode *uint64
eventType simulations.EventType
connUp *bool
msgReceive *bool
protocol *string
msgCode *uint64
}
// NewPeerEventsFilter returns a new PeerEventsFilter instance.
@ -47,20 +53,48 @@ func NewPeerEventsFilter() *PeerEventsFilter {
return &PeerEventsFilter{}
}
// Type sets the filter to only one peer event type.
func (f *PeerEventsFilter) Type(t p2p.PeerEventType) *PeerEventsFilter {
f.t = &t
// Connect sets the filter to events when two nodes connect.
func (f *PeerEventsFilter) Connect() *PeerEventsFilter {
f.eventType = simulations.EventTypeConn
b := true
f.connUp = &b
return f
}
// Drop sets the filter to events when two nodes disconnect.
func (f *PeerEventsFilter) Drop() *PeerEventsFilter {
f.eventType = simulations.EventTypeConn
b := false
f.connUp = &b
return f
}
// ReceivedMessages sets the filter to only messages that are received.
func (f *PeerEventsFilter) ReceivedMessages() *PeerEventsFilter {
f.eventType = simulations.EventTypeMsg
b := true
f.msgReceive = &b
return f
}
// SentMessages sets the filter to only messages that are sent.
func (f *PeerEventsFilter) SentMessages() *PeerEventsFilter {
f.eventType = simulations.EventTypeMsg
b := false
f.msgReceive = &b
return f
}
// Protocol sets the filter to only one message protocol.
func (f *PeerEventsFilter) Protocol(p string) *PeerEventsFilter {
f.eventType = simulations.EventTypeMsg
f.protocol = &p
return f
}
// MsgCode sets the filter to only one msg code.
func (f *PeerEventsFilter) MsgCode(c uint64) *PeerEventsFilter {
f.eventType = simulations.EventTypeMsg
f.msgCode = &c
return f
}
@ -80,19 +114,8 @@ func (s *Simulation) PeerEvents(ctx context.Context, ids []enode.ID, filters ...
go func(id enode.ID) {
defer s.shutdownWG.Done()
client, err := s.Net.GetNode(id).Client()
if err != nil {
subsWG.Done()
eventC <- PeerEvent{NodeID: id, Error: err}
return
}
events := make(chan *p2p.PeerEvent)
sub, err := client.Subscribe(ctx, "admin", events, "peerEvents")
if err != nil {
subsWG.Done()
eventC <- PeerEvent{NodeID: id, Error: err}
return
}
events := make(chan *simulations.Event)
sub := s.Net.Events().Subscribe(events)
defer sub.Unsubscribe()
subsWG.Done()
@ -110,28 +133,55 @@ func (s *Simulation) PeerEvents(ctx context.Context, ids []enode.ID, filters ...
case <-s.Done():
return
case e := <-events:
// ignore control events
if e.Control {
continue
}
match := len(filters) == 0 // if there are no filters match all events
for _, f := range filters {
if f.t != nil && *f.t != e.Type {
continue
if f.eventType == simulations.EventTypeConn && e.Conn != nil {
if *f.connUp != e.Conn.Up {
continue
}
// all connection filter parameters matched, break the loop
match = true
break
}
if f.eventType == simulations.EventTypeMsg && e.Msg != nil {
if f.msgReceive != nil && *f.msgReceive != e.Msg.Received {
continue
}
if f.protocol != nil && *f.protocol != e.Msg.Protocol {
continue
}
if f.msgCode != nil && *f.msgCode != e.Msg.Code {
continue
}
// all message filter parameters matched, break the loop
match = true
break
}
if f.protocol != nil && *f.protocol != e.Protocol {
continue
}
var peerID enode.ID
switch e.Type {
case simulations.EventTypeConn:
peerID = e.Conn.One
if peerID == id {
peerID = e.Conn.Other
}
if f.msgCode != nil && e.MsgCode != nil && *f.msgCode != *e.MsgCode {
continue
case simulations.EventTypeMsg:
peerID = e.Msg.One
if peerID == id {
peerID = e.Msg.Other
}
// all filter parameters matched, break the loop
match = true
break
}
if match {
select {
case eventC <- PeerEvent{NodeID: id, Event: e}:
case eventC <- PeerEvent{NodeID: id, PeerID: peerID, Event: e}:
case <-ctx.Done():
if err := ctx.Err(); err != nil {
select {
case eventC <- PeerEvent{NodeID: id, Error: err}:
case eventC <- PeerEvent{NodeID: id, PeerID: peerID, Error: err}:
case <-s.Done():
}
}

13
swarm/network/simulation/example_test.go
Unescape View File

@ -24,7 +24,6 @@ import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/network/simulation"
@ -87,7 +86,7 @@ func ExampleSimulation_PeerEvents() {
log.Error("peer event", "err", e.Error)
continue
}
log.Info("peer event", "node", e.NodeID, "peer", e.Event.Peer, "msgcode", e.Event.MsgCode)
log.Info("peer event", "node", e.NodeID, "peer", e.PeerID, "type", e.Event.Type)
}
}()
}
@ -100,7 +99,7 @@ func ExampleSimulation_PeerEvents_disconnections() {
disconnections := sim.PeerEvents(
context.Background(),
sim.NodeIDs(),
simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeDrop),
simulation.NewPeerEventsFilter().Drop(),
)
go func() {
@ -109,7 +108,7 @@ func ExampleSimulation_PeerEvents_disconnections() {
log.Error("peer drop", "err", d.Error)
continue
}
log.Warn("peer drop", "node", d.NodeID, "peer", d.Event.Peer)
log.Warn("peer drop", "node", d.NodeID, "peer", d.PeerID)
}
}()
}
@ -124,8 +123,8 @@ func ExampleSimulation_PeerEvents_multipleFilters() {
context.Background(),
sim.NodeIDs(),
// Watch when bzz messages 1 and 4 are received.
simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeMsgRecv).Protocol("bzz").MsgCode(1),
simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeMsgRecv).Protocol("bzz").MsgCode(4),
simulation.NewPeerEventsFilter().ReceivedMessages().Protocol("bzz").MsgCode(1),
simulation.NewPeerEventsFilter().ReceivedMessages().Protocol("bzz").MsgCode(4),
)
go func() {
@ -134,7 +133,7 @@ func ExampleSimulation_PeerEvents_multipleFilters() {
log.Error("bzz message", "err", m.Error)
continue
}
log.Info("bzz message", "node", m.NodeID, "peer", m.Event.Peer)
log.Info("bzz message", "node", m.NodeID, "peer", m.PeerID)
}
}()
}

8
swarm/network/stream/delivery_test.go
Unescape View File

@ -565,13 +565,13 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
disconnections := sim.PeerEvents(
context.Background(),
sim.NodeIDs(),
simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeDrop),
simulation.NewPeerEventsFilter().Drop(),
)
go func() {
for d := range disconnections {
if d.Error != nil {
log.Error("peer drop", "node", d.NodeID, "peer", d.Event.Peer)
log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
t.Fatal(d.Error)
}
}
@ -697,13 +697,13 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
disconnections := sim.PeerEvents(
context.Background(),
sim.NodeIDs(),
simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeDrop),
simulation.NewPeerEventsFilter().Drop(),
)
go func() {
for d := range disconnections {
if d.Error != nil {
log.Error("peer drop", "node", d.NodeID, "peer", d.Event.Peer)
log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
b.Fatal(d.Error)
}
}

5
swarm/network/stream/intervals_test.go
Unescape View File

@ -27,7 +27,6 @@ import (
"github.com/ethereum/go-ethereum/log"
"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/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/network"
@ -154,7 +153,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
disconnections := sim.PeerEvents(
context.Background(),
sim.NodeIDs(),
simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeDrop),
simulation.NewPeerEventsFilter().Drop(),
)
err = registry.Subscribe(storer, NewStream(externalStreamName, "", live), history, Top)
@ -165,7 +164,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
go func() {
for d := range disconnections {
if d.Error != nil {
log.Error("peer drop", "node", d.NodeID, "peer", d.Event.Peer)
log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
t.Fatal(d.Error)
}
}

11
swarm/network/stream/snapshot_sync_test.go
Unescape View File

@ -27,7 +27,6 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"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/simulations"
"github.com/ethereum/go-ethereum/p2p/simulations/adapters"
@ -210,12 +209,12 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) {
disconnections := sim.PeerEvents(
context.Background(),
sim.NodeIDs(),
simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeDrop),
simulation.NewPeerEventsFilter().Drop(),
)
go func() {
for d := range disconnections {
log.Error("peer drop", "node", d.NodeID, "peer", d.Event.Peer)
log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
t.Fatal("unexpected disconnect")
cancelSimRun()
}
@ -402,12 +401,12 @@ func testSyncingViaDirectSubscribe(t *testing.T, chunkCount int, nodeCount int)
disconnections := sim.PeerEvents(
context.Background(),
sim.NodeIDs(),
simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeDrop),
simulation.NewPeerEventsFilter().Drop(),
)
go func() {
for d := range disconnections {
log.Error("peer drop", "node", d.NodeID, "peer", d.Event.Peer)
log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
t.Fatal("unexpected disconnect")
cancelSimRun()
}
@ -428,7 +427,7 @@ func testSyncingViaDirectSubscribe(t *testing.T, chunkCount int, nodeCount int)
var subscriptionCount int
filter := simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeMsgRecv).Protocol("stream").MsgCode(4)
filter := simulation.NewPeerEventsFilter().ReceivedMessages().Protocol("stream").MsgCode(4)
eventC := sim.PeerEvents(ctx, nodeIDs, filter)
for j, node := range nodeIDs {

5
swarm/network/stream/syncer_test.go
Unescape View File

@ -28,7 +28,6 @@ import (
"github.com/ethereum/go-ethereum/common"
"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/simulations/adapters"
"github.com/ethereum/go-ethereum/swarm/log"
@ -151,13 +150,13 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
disconnections := sim.PeerEvents(
context.Background(),
sim.NodeIDs(),
simulation.NewPeerEventsFilter().Type(p2p.PeerEventTypeDrop),
simulation.NewPeerEventsFilter().Drop(),
)
go func() {
for d := range disconnections {
if d.Error != nil {
log.Error("peer drop", "node", d.NodeID, "peer", d.Event.Peer)
log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
t.Fatal(d.Error)
}
}

Write Preview
Loading…
Cancel
Save