p2p/sim: update socket pipe to use the default available buffer from OS

p2p/simulations/adapters/inproc.go

@@ -34,11 +34,6 @@ import (
 	"github.com/ethereum/go-ethereum/rpc"
 )
 
-const (
-	socketReadBuffer  = 5000 * 1024
-	socketWriteBuffer = 5000 * 1024
-)
-
 // SimAdapter is a NodeAdapter which creates in-memory simulation nodes and
 // connects them using net.Pipe or OS socket connections
 type SimAdapter struct {
@@ -378,20 +373,10 @@ func socketPipe() (net.Conn, net.Conn, error) {
 		return nil, nil, err
 	}
 
-	err = setSocketBuffer(pipe1)
-	if err != nil {
-		return nil, nil, err
-	}
-
-	err = setSocketBuffer(pipe2)
-	if err != nil {
-		return nil, nil, err
-	}
-
 	return pipe1, pipe2, nil
 }
 
-func setSocketBuffer(conn net.Conn) error {
+func setSocketBuffer(conn net.Conn, socketReadBuffer int, socketWriteBuffer int) error {
 	switch v := conn.(type) {
 	case *net.UnixConn:
 		err := v.SetReadBuffer(socketReadBuffer)

p2p/simulations/adapters/inproc_test.go

@@ -27,7 +27,7 @@ import (
 func TestSocketPipe(t *testing.T) {
 	c1, c2, err := socketPipe()
 	if err != nil {
-		t.Skip(err)
+		t.Fatal(err)
 	}
 
 	done := make(chan struct{})
@@ -35,6 +35,9 @@ func TestSocketPipe(t *testing.T) {
 	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))
@@ -44,6 +47,7 @@ func TestSocketPipe(t *testing.T) {
 					t.Fatal(err)
 				}
 			}
+		}()
 
 		for i := 0; i < msgs; i++ {
 			msg := make([]byte, size)
@@ -72,7 +76,7 @@ func TestSocketPipe(t *testing.T) {
 func TestSocketPipeBidirections(t *testing.T) {
 	c1, c2, err := socketPipe()
 	if err != nil {
-		t.Skip(err)
+		t.Fatal(err)
 	}
 
 	done := make(chan struct{})
@@ -80,6 +84,9 @@ func TestSocketPipeBidirections(t *testing.T) {
 	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`)
 
@@ -88,6 +95,7 @@ func TestSocketPipeBidirections(t *testing.T) {
 					t.Fatal(err)
 				}
 			}
+		}()
 
 		for i := 0; i < msgs; i++ {
 			out := make([]byte, size)

swarm/network/simulations/discovery/discovery_test.go

@@ -99,7 +99,7 @@ func testDiscoverySimulationExecAdapter(t *testing.T, nodes, conns int) {
 	testDiscoverySimulation(t, nodes, conns, adapters.NewExecAdapter(baseDir))
 }
 
-func XTestDiscoverySimulationSocketAdapter(t *testing.T) {
+func TestDiscoverySimulationSocketAdapter(t *testing.T) {
 	testDiscoverySimulationSocketAdapter(t, *nodeCount, *initCount)
 }