From 9f7306f014ec64f4b8323758c8742e154eec0599 Mon Sep 17 00:00:00 2001 From: Anton Evangelatov Date: Tue, 12 Jun 2018 11:54:04 +0300 Subject: [PATCH] p2p/sim: remove SocketPipe which does not work on MS Windows --- p2p/simulations/adapters/inproc.go | 22 +---- p2p/simulations/adapters/inproc_test.go | 117 +----------------------- p2p/simulations/pipes/pipes.go | 31 ------- 3 files changed, 7 insertions(+), 163 deletions(-) diff --git a/p2p/simulations/adapters/inproc.go b/p2p/simulations/adapters/inproc.go index 99904c8766..b68d08f392 100644 --- a/p2p/simulations/adapters/inproc.go +++ b/p2p/simulations/adapters/inproc.go @@ -33,7 +33,7 @@ import ( ) // SimAdapter is a NodeAdapter which creates in-memory simulation nodes and -// connects them using net.Pipe or OS socket connections +// connects them using net.Pipe type SimAdapter struct { pipe func() (net.Conn, net.Conn, error) mtx sync.RWMutex @@ -53,18 +53,6 @@ func NewSimAdapter(services map[string]ServiceFunc) *SimAdapter { } } -// NewSocketAdapter 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 OS socketpairs for in-memory simulated network connections -func NewSocketAdapter(services map[string]ServiceFunc) *SimAdapter { - return &SimAdapter{ - pipe: pipes.SocketPipe, - nodes: make(map[discover.NodeID]*SimNode), - services: services, - } -} - func NewTCPAdapter(services map[string]ServiceFunc) *SimAdapter { return &SimAdapter{ pipe: pipes.TCPPipe, @@ -126,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 or OS socket 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 { @@ -136,7 +124,7 @@ 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) } - // SimAdapter.pipe is either net.Pipe (NewSimAdapter) or socketPipe (NewSocketAdapter) + // SimAdapter.pipe is net.Pipe (NewSimAdapter) pipe1, pipe2, err := s.pipe() if err != nil { return nil, err @@ -171,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 -// net.Pipe or OS socket 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 diff --git a/p2p/simulations/adapters/inproc_test.go b/p2p/simulations/adapters/inproc_test.go index 3cc3f5434a..e1e092f6e1 100644 --- a/p2p/simulations/adapters/inproc_test.go +++ b/p2p/simulations/adapters/inproc_test.go @@ -26,120 +26,7 @@ import ( "github.com/ethereum/go-ethereum/p2p/simulations/pipes" ) -func TestSocketPipe(t *testing.T) { - c1, c2, err := pipes.SocketPipe() - if err != nil { - t.Fatal(err) - } - - done := make(chan struct{}) - - go func() { - msgs := 20 - size := 8 - - // OS socket pipe is blocking (depending on buffer size on OS), 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 TestSocketPipeBidirections(t *testing.T) { - c1, c2, err := pipes.SocketPipe() - if err != nil { - t.Fatal(err) - } - - done := make(chan struct{}) - - go func() { - msgs := 100 - size := 4 - - // OS socket pipe is blocking (depending on buffer size on OS), so writes are emitted asynchronously - go func() { - for i := 0; i < msgs; i++ { - msg := []byte(`ping`) - - _, err := c1.Write(msg) - if err != nil { - t.Fatal(err) - } - } - }() - - for i := 0; i < msgs; i++ { - out := make([]byte, size) - _, err := c2.Read(out) - if err != nil { - t.Fatal(err) - } - - if bytes.Equal(out, []byte(`ping`)) { - msg := []byte(`pong`) - _, err := c2.Write(msg) - if err != nil { - t.Fatal(err) - } - } - } - - for i := 0; i < msgs; i++ { - expected := []byte(`pong`) - - out := make([]byte, size) - _, err := c1.Read(out) - if err != nil { - t.Fatal(err) - } - - if !bytes.Equal(out, expected) { - t.Fatalf("expected %#v, got %#v", expected, out) - } - } - - done <- struct{}{} - }() - - select { - case <-done: - case <-time.After(5 * time.Second): - t.Fatal("test timeout") - } -} - -func TestTcpPipe(t *testing.T) { +func TestTCPPipe(t *testing.T) { c1, c2, err := pipes.TCPPipe() if err != nil { t.Fatal(err) @@ -184,7 +71,7 @@ func TestTcpPipe(t *testing.T) { } } -func TestTcpPipeBidirections(t *testing.T) { +func TestTCPPipeBidirections(t *testing.T) { c1, c2, err := pipes.TCPPipe() if err != nil { t.Fatal(err) diff --git a/p2p/simulations/pipes/pipes.go b/p2p/simulations/pipes/pipes.go index 000cae7405..8532c1bcf0 100644 --- a/p2p/simulations/pipes/pipes.go +++ b/p2p/simulations/pipes/pipes.go @@ -17,10 +17,7 @@ package pipes import ( - "crypto/rand" "net" - "os" - "syscall" ) // NetPipe wraps net.Pipe in a signature returning an error @@ -56,31 +53,3 @@ func TCPPipe() (net.Conn, net.Conn, error) { } return aconn, dconn, nil } - -// SocketPipe creates an in process full duplex pipe based on OS sockets -// credit to @lmars & Flynn -// https://github.com/flynn/flynn/blob/master/host/containerinit/init.go#L743-L749 -// using this in large simulations requires raising OS's max open file limit -func SocketPipe() (net.Conn, net.Conn, error) { - pair, err := syscall.Socketpair(syscall.AF_UNIX, syscall.SOCK_STREAM, 0) - if err != nil { - return nil, nil, err - } - nameb := make([]byte, 8) - _, err = rand.Read(nameb) - if err != nil { - return nil, nil, err - } - f1 := os.NewFile(uintptr(pair[0]), string(nameb)+".out") - f2 := os.NewFile(uintptr(pair[1]), string(nameb)+".in") - pipe1, err := net.FileConn(f1) - if err != nil { - return nil, nil, err - } - pipe2, err := net.FileConn(f2) - if err != nil { - return nil, nil, err - } - - return pipe1, pipe2, nil -}