aboutsummaryrefslogtreecommitdiffstats
path: root/p2p/simulations/adapters
diff options
context:
space:
mode:
Diffstat (limited to 'p2p/simulations/adapters')
-rw-r--r--p2p/simulations/adapters/docker.go12
-rw-r--r--p2p/simulations/adapters/exec.go60
-rw-r--r--p2p/simulations/adapters/inproc.go48
-rw-r--r--p2p/simulations/adapters/inproc_test.go259
-rw-r--r--p2p/simulations/adapters/types.go56
5 files changed, 391 insertions, 44 deletions
diff --git a/p2p/simulations/adapters/docker.go b/p2p/simulations/adapters/docker.go
index 8ef5629fb..d145c46b3 100644
--- a/p2p/simulations/adapters/docker.go
+++ b/p2p/simulations/adapters/docker.go
@@ -28,11 +28,14 @@ import (
"strings"
"github.com/docker/docker/pkg/reexec"
- "github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p/discover"
)
+var (
+ ErrLinuxOnly = errors.New("DockerAdapter can only be used on Linux as it uses the current binary (which must be a Linux binary)")
+)
+
// DockerAdapter is a NodeAdapter which runs simulation nodes inside Docker
// containers.
//
@@ -52,7 +55,7 @@ func NewDockerAdapter() (*DockerAdapter, error) {
// It is reasonable to require this because the caller can just
// compile the current binary in a Docker container.
if runtime.GOOS != "linux" {
- return nil, errors.New("DockerAdapter can only be used on Linux as it uses the current binary (which must be a Linux binary)")
+ return nil, ErrLinuxOnly
}
if err := buildDockerImage(); err != nil {
@@ -95,7 +98,10 @@ func (d *DockerAdapter) NewNode(config *NodeConfig) (Node, error) {
conf.Stack.P2P.NoDiscovery = true
conf.Stack.P2P.NAT = nil
conf.Stack.NoUSB = true
- conf.Stack.Logger = log.New("node.id", config.ID.String())
+
+ // listen on all interfaces on a given port, which we set when we
+ // initialise NodeConfig (usually a random port)
+ conf.Stack.P2P.ListenAddr = fmt.Sprintf(":%d", config.Port)
node := &DockerNode{
ExecNode: ExecNode{
diff --git a/p2p/simulations/adapters/exec.go b/p2p/simulations/adapters/exec.go
index f381c1159..e64cebc2a 100644
--- a/p2p/simulations/adapters/exec.go
+++ b/p2p/simulations/adapters/exec.go
@@ -17,6 +17,7 @@
package adapters
import (
+ "bufio"
"context"
"crypto/ecdsa"
"encoding/json"
@@ -103,9 +104,9 @@ func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) {
conf.Stack.P2P.NAT = nil
conf.Stack.NoUSB = true
- // listen on a random localhost port (we'll get the actual port after
- // starting the node through the RPC admin.nodeInfo method)
- conf.Stack.P2P.ListenAddr = "127.0.0.1:0"
+ // listen on a localhost port, which we set when we
+ // initialise NodeConfig (usually a random port)
+ conf.Stack.P2P.ListenAddr = fmt.Sprintf(":%d", config.Port)
node := &ExecNode{
ID: config.ID,
@@ -190,9 +191,23 @@ func (n *ExecNode) Start(snapshots map[string][]byte) (err error) {
n.Cmd = cmd
// read the WebSocket address from the stderr logs
- wsAddr, err := findWSAddr(stderrR, 10*time.Second)
- if err != nil {
- return fmt.Errorf("error getting WebSocket address: %s", err)
+ var wsAddr string
+ wsAddrC := make(chan string)
+ go func() {
+ s := bufio.NewScanner(stderrR)
+ for s.Scan() {
+ if strings.Contains(s.Text(), "WebSocket endpoint opened") {
+ wsAddrC <- wsAddrPattern.FindString(s.Text())
+ }
+ }
+ }()
+ select {
+ case wsAddr = <-wsAddrC:
+ if wsAddr == "" {
+ return errors.New("failed to read WebSocket address from stderr")
+ }
+ case <-time.After(10 * time.Second):
+ return errors.New("timed out waiting for WebSocket address on stderr")
}
// create the RPC client and load the node info
@@ -318,6 +333,21 @@ type execNodeConfig struct {
PeerAddrs map[string]string `json:"peer_addrs,omitempty"`
}
+// ExternalIP gets an external IP address so that Enode URL is usable
+func ExternalIP() net.IP {
+ addrs, err := net.InterfaceAddrs()
+ if err != nil {
+ log.Crit("error getting IP address", "err", err)
+ }
+ for _, addr := range addrs {
+ if ip, ok := addr.(*net.IPNet); ok && !ip.IP.IsLoopback() && !ip.IP.IsLinkLocalUnicast() {
+ return ip.IP
+ }
+ }
+ log.Warn("unable to determine explicit IP address, falling back to loopback")
+ return net.IP{127, 0, 0, 1}
+}
+
// execP2PNode starts a devp2p node when the current binary is executed with
// argv[0] being "p2p-node", reading the service / ID from argv[1] / argv[2]
// and the node config from the _P2P_NODE_CONFIG environment variable
@@ -341,25 +371,11 @@ func execP2PNode() {
conf.Stack.P2P.PrivateKey = conf.Node.PrivateKey
conf.Stack.Logger = log.New("node.id", conf.Node.ID.String())
- // use explicit IP address in ListenAddr so that Enode URL is usable
- externalIP := func() string {
- addrs, err := net.InterfaceAddrs()
- if err != nil {
- log.Crit("error getting IP address", "err", err)
- }
- for _, addr := range addrs {
- if ip, ok := addr.(*net.IPNet); ok && !ip.IP.IsLoopback() {
- return ip.IP.String()
- }
- }
- log.Crit("unable to determine explicit IP address")
- return ""
- }
if strings.HasPrefix(conf.Stack.P2P.ListenAddr, ":") {
- conf.Stack.P2P.ListenAddr = externalIP() + conf.Stack.P2P.ListenAddr
+ conf.Stack.P2P.ListenAddr = ExternalIP().String() + conf.Stack.P2P.ListenAddr
}
if conf.Stack.WSHost == "0.0.0.0" {
- conf.Stack.WSHost = externalIP()
+ conf.Stack.WSHost = ExternalIP().String()
}
// initialize the devp2p stack
diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go
index 6d90b4a9f..b68d08f39 100644
--- a/p2p/simulations/adapters/inproc.go
+++ b/p2p/simulations/adapters/inproc.go
@@ -28,12 +28,14 @@ import (
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
+ "github.com/ethereum/go-ethereum/p2p/simulations/pipes"
"github.com/ethereum/go-ethereum/rpc"
)
// SimAdapter is a NodeAdapter which creates in-memory simulation nodes and
-// connects them using in-memory net.Pipe connections
+// connects them using net.Pipe
type SimAdapter struct {
+ pipe func() (net.Conn, net.Conn, error)
mtx sync.RWMutex
nodes map[discover.NodeID]*SimNode
services map[string]ServiceFunc
@@ -42,8 +44,18 @@ type SimAdapter struct {
// NewSimAdapter creates a SimAdapter which is capable of running in-memory
// simulation nodes running any of the given services (the services to run on a
// particular node are passed to the NewNode function in the NodeConfig)
+// the adapter uses a net.Pipe for in-memory simulated network connections
func NewSimAdapter(services map[string]ServiceFunc) *SimAdapter {
return &SimAdapter{
+ pipe: pipes.NetPipe,
+ nodes: make(map[discover.NodeID]*SimNode),
+ services: services,
+ }
+}
+
+func NewTCPAdapter(services map[string]ServiceFunc) *SimAdapter {
+ return &SimAdapter{
+ pipe: pipes.TCPPipe,
nodes: make(map[discover.NodeID]*SimNode),
services: services,
}
@@ -81,7 +93,7 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
MaxPeers: math.MaxInt32,
NoDiscovery: true,
Dialer: s,
- EnableMsgEvents: true,
+ EnableMsgEvents: config.EnableMsgEvents,
},
NoUSB: true,
Logger: log.New("node.id", id.String()),
@@ -102,7 +114,7 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
}
// Dial implements the p2p.NodeDialer interface by connecting to the node using
-// an in-memory net.Pipe connection
+// an in-memory net.Pipe
func (s *SimAdapter) Dial(dest *discover.Node) (conn net.Conn, err error) {
node, ok := s.GetNode(dest.ID)
if !ok {
@@ -112,7 +124,14 @@ func (s *SimAdapter) Dial(dest *discover.Node) (conn net.Conn, err error) {
if srv == nil {
return nil, fmt.Errorf("node not running: %s", dest.ID)
}
- pipe1, pipe2 := net.Pipe()
+ // SimAdapter.pipe is net.Pipe (NewSimAdapter)
+ pipe1, pipe2, err := s.pipe()
+ if err != nil {
+ return nil, err
+ }
+ // this is simulated 'listening'
+ // asynchronously call the dialed destintion node's p2p server
+ // to set up connection on the 'listening' side
go srv.SetupConn(pipe1, 0, nil)
return pipe2, nil
}
@@ -140,8 +159,8 @@ func (s *SimAdapter) GetNode(id discover.NodeID) (*SimNode, bool) {
}
// SimNode is an in-memory simulation node which connects to other nodes using
-// an in-memory net.Pipe connection (see SimAdapter.Dial), running devp2p
-// protocols directly over that pipe
+// net.Pipe (see SimAdapter.Dial), running devp2p protocols directly over that
+// pipe
type SimNode struct {
lock sync.RWMutex
ID discover.NodeID
@@ -241,7 +260,7 @@ func (sn *SimNode) Start(snapshots map[string][]byte) error {
for _, name := range sn.config.Services {
if err := sn.node.Register(newService(name)); err != nil {
regErr = err
- return
+ break
}
}
})
@@ -314,3 +333,18 @@ func (sn *SimNode) NodeInfo() *p2p.NodeInfo {
}
return server.NodeInfo()
}
+
+func setSocketBuffer(conn net.Conn, socketReadBuffer int, socketWriteBuffer int) error {
+ switch v := conn.(type) {
+ case *net.UnixConn:
+ err := v.SetReadBuffer(socketReadBuffer)
+ if err != nil {
+ return err
+ }
+ err = v.SetWriteBuffer(socketWriteBuffer)
+ if err != nil {
+ return err
+ }
+ }
+ return nil
+}
diff --git a/p2p/simulations/adapters/inproc_test.go b/p2p/simulations/adapters/inproc_test.go
new file mode 100644
index 000000000..e1e092f6e
--- /dev/null
+++ b/p2p/simulations/adapters/inproc_test.go
@@ -0,0 +1,259 @@
+// Copyright 2017 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/>.
+
+package adapters
+
+import (
+ "bytes"
+ "encoding/binary"
+ "fmt"
+ "testing"
+ "time"
+
+ "github.com/ethereum/go-ethereum/p2p/simulations/pipes"
+)
+
+func TestTCPPipe(t *testing.T) {
+ c1, c2, err := pipes.TCPPipe()
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ done := make(chan struct{})
+
+ go func() {
+ msgs := 50
+ size := 1024
+ for i := 0; i < msgs; i++ {
+ msg := make([]byte, size)
+ _ = binary.PutUvarint(msg, uint64(i))
+
+ _, err := c1.Write(msg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+
+ for i := 0; i < msgs; i++ {
+ msg := make([]byte, size)
+ _ = binary.PutUvarint(msg, uint64(i))
+
+ out := make([]byte, size)
+ _, err := c2.Read(out)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ if !bytes.Equal(msg, out) {
+ t.Fatalf("expected %#v, got %#v", msg, out)
+ }
+ }
+ done <- struct{}{}
+ }()
+
+ select {
+ case <-done:
+ case <-time.After(5 * time.Second):
+ t.Fatal("test timeout")
+ }
+}
+
+func TestTCPPipeBidirections(t *testing.T) {
+ c1, c2, err := pipes.TCPPipe()
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ done := make(chan struct{})
+
+ go func() {
+ msgs := 50
+ size := 7
+ for i := 0; i < msgs; i++ {
+ msg := []byte(fmt.Sprintf("ping %02d", i))
+
+ _, err := c1.Write(msg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+
+ for i := 0; i < msgs; i++ {
+ expected := []byte(fmt.Sprintf("ping %02d", i))
+
+ out := make([]byte, size)
+ _, err := c2.Read(out)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ if !bytes.Equal(expected, out) {
+ t.Fatalf("expected %#v, got %#v", out, expected)
+ } else {
+ msg := []byte(fmt.Sprintf("pong %02d", i))
+ _, err := c2.Write(msg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+ }
+
+ for i := 0; i < msgs; i++ {
+ expected := []byte(fmt.Sprintf("pong %02d", i))
+
+ out := make([]byte, size)
+ _, err := c1.Read(out)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ if !bytes.Equal(expected, out) {
+ t.Fatalf("expected %#v, got %#v", out, expected)
+ }
+ }
+ done <- struct{}{}
+ }()
+
+ select {
+ case <-done:
+ case <-time.After(5 * time.Second):
+ t.Fatal("test timeout")
+ }
+}
+
+func TestNetPipe(t *testing.T) {
+ c1, c2, err := pipes.NetPipe()
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ done := make(chan struct{})
+
+ go func() {
+ msgs := 50
+ size := 1024
+ // netPipe is blocking, so writes are emitted asynchronously
+ go func() {
+ for i := 0; i < msgs; i++ {
+ msg := make([]byte, size)
+ _ = binary.PutUvarint(msg, uint64(i))
+
+ _, err := c1.Write(msg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+ }()
+
+ for i := 0; i < msgs; i++ {
+ msg := make([]byte, size)
+ _ = binary.PutUvarint(msg, uint64(i))
+
+ out := make([]byte, size)
+ _, err := c2.Read(out)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ if !bytes.Equal(msg, out) {
+ t.Fatalf("expected %#v, got %#v", msg, out)
+ }
+ }
+
+ done <- struct{}{}
+ }()
+
+ select {
+ case <-done:
+ case <-time.After(5 * time.Second):
+ t.Fatal("test timeout")
+ }
+}
+
+func TestNetPipeBidirections(t *testing.T) {
+ c1, c2, err := pipes.NetPipe()
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ done := make(chan struct{})
+
+ go func() {
+ msgs := 1000
+ size := 8
+ pingTemplate := "ping %03d"
+ pongTemplate := "pong %03d"
+
+ // netPipe is blocking, so writes are emitted asynchronously
+ go func() {
+ for i := 0; i < msgs; i++ {
+ msg := []byte(fmt.Sprintf(pingTemplate, i))
+
+ _, err := c1.Write(msg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+ }()
+
+ // netPipe is blocking, so reads for pong are emitted asynchronously
+ go func() {
+ for i := 0; i < msgs; i++ {
+ expected := []byte(fmt.Sprintf(pongTemplate, i))
+
+ out := make([]byte, size)
+ _, err := c1.Read(out)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ if !bytes.Equal(expected, out) {
+ t.Fatalf("expected %#v, got %#v", expected, out)
+ }
+ }
+
+ done <- struct{}{}
+ }()
+
+ // expect to read pings, and respond with pongs to the alternate connection
+ for i := 0; i < msgs; i++ {
+ expected := []byte(fmt.Sprintf(pingTemplate, i))
+
+ out := make([]byte, size)
+ _, err := c2.Read(out)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ if !bytes.Equal(expected, out) {
+ t.Fatalf("expected %#v, got %#v", expected, out)
+ } else {
+ msg := []byte(fmt.Sprintf(pongTemplate, i))
+
+ _, err := c2.Write(msg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+ }
+ }()
+
+ select {
+ case <-done:
+ case <-time.After(5 * time.Second):
+ t.Fatal("test timeout")
+ }
+}
diff --git a/p2p/simulations/adapters/types.go b/p2p/simulations/adapters/types.go
index 5b4b47fe2..2c4b9dd8f 100644
--- a/p2p/simulations/adapters/types.go
+++ b/p2p/simulations/adapters/types.go
@@ -23,6 +23,7 @@ import (
"fmt"
"net"
"os"
+ "strconv"
"github.com/docker/docker/pkg/reexec"
"github.com/ethereum/go-ethereum/crypto"
@@ -97,24 +98,30 @@ type NodeConfig struct {
// function to sanction or prevent suggesting a peer
Reachable func(id discover.NodeID) bool
+
+ Port uint16
}
// nodeConfigJSON is used to encode and decode NodeConfig as JSON by encoding
// all fields as strings
type nodeConfigJSON struct {
- ID string `json:"id"`
- PrivateKey string `json:"private_key"`
- Name string `json:"name"`
- Services []string `json:"services"`
+ ID string `json:"id"`
+ PrivateKey string `json:"private_key"`
+ Name string `json:"name"`
+ Services []string `json:"services"`
+ EnableMsgEvents bool `json:"enable_msg_events"`
+ Port uint16 `json:"port"`
}
// MarshalJSON implements the json.Marshaler interface by encoding the config
// fields as strings
func (n *NodeConfig) MarshalJSON() ([]byte, error) {
confJSON := nodeConfigJSON{
- ID: n.ID.String(),
- Name: n.Name,
- Services: n.Services,
+ ID: n.ID.String(),
+ Name: n.Name,
+ Services: n.Services,
+ Port: n.Port,
+ EnableMsgEvents: n.EnableMsgEvents,
}
if n.PrivateKey != nil {
confJSON.PrivateKey = hex.EncodeToString(crypto.FromECDSA(n.PrivateKey))
@@ -152,6 +159,8 @@ func (n *NodeConfig) UnmarshalJSON(data []byte) error {
n.Name = confJSON.Name
n.Services = confJSON.Services
+ n.Port = confJSON.Port
+ n.EnableMsgEvents = confJSON.EnableMsgEvents
return nil
}
@@ -163,13 +172,36 @@ func RandomNodeConfig() *NodeConfig {
if err != nil {
panic("unable to generate key")
}
- var id discover.NodeID
- pubkey := crypto.FromECDSAPub(&key.PublicKey)
- copy(id[:], pubkey[1:])
+
+ id := discover.PubkeyID(&key.PublicKey)
+ port, err := assignTCPPort()
+ if err != nil {
+ panic("unable to assign tcp port")
+ }
return &NodeConfig{
- ID: id,
- PrivateKey: key,
+ ID: id,
+ Name: fmt.Sprintf("node_%s", id.String()),
+ PrivateKey: key,
+ Port: port,
+ EnableMsgEvents: true,
+ }
+}
+
+func assignTCPPort() (uint16, error) {
+ l, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ return 0, err
+ }
+ l.Close()
+ _, port, err := net.SplitHostPort(l.Addr().String())
+ if err != nil {
+ return 0, err
+ }
+ p, err := strconv.ParseInt(port, 10, 32)
+ if err != nil {
+ return 0, err
}
+ return uint16(p), nil
}
// ServiceContext is a collection of options and methods which can be utilised
generated by cgit (git 2.34.1) at 2024-10-11 06:36:43 +0800