Repository: libp2p/go-libp2p
Pull request #2184 (2023-03-15 12:35:45)

autorelay: remove unused semaphore


Patch:
diff --git a/p2p/host/autorelay/relay_finder.go b/p2p/host/autorelay/relay_finder.go
index 36db86b5..bff33003 100644
--- a/p2p/host/autorelay/relay_finder.go
+++ b/p2p/host/autorelay/relay_finder.go
@@ -465,19 +465,12 @@ func (rf *relayFinder) tryNode(ctx context.Context, pi peer.AddrInfo) (supportsR
 // This function makes sure that we only run one instance of maybeConnectToRelay at once, and buffers
 // exactly one more trigger event to run maybeConnectToRelay.
 func (rf *relayFinder) handleNewCandidates(ctx context.Context) {
-	sem := make(chan struct{}, 1)
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case <-rf.maybeConnectToRelayTrigger:
-			select {
-			case <-ctx.Done():
-				return
-			case sem <- struct{}{}:
-			}
 			rf.maybeConnectToRelay(ctx)
-			<-sem
 		}
 	}
 }
diff --git a/p2p/host/basic/basic_host.go b/p2p/host/basic/basic_host.go
index c1144797..bee068fd 100644
--- a/p2p/host/basic/basic_host.go
+++ b/p2p/host/basic/basic_host.go
@@ -635,8 +635,13 @@ func (h *BasicHost) NewStream(ctx context.Context, p peer.ID, pids ...protocol.I
 		}
 	}
 
-	s, err := h.Network().NewStream(ctx, p)
+	s, err := h.Network().NewStream(network.WithNoDial(ctx, "already dialed"), p)
 	if err != nil {
+		// TODO: It would be nicer to get the actual error from the swarm,
+		// but this will require some more work.
+		if errors.Is(err, network.ErrNoConn) {
+			return nil, errors.New("connection failed")
+		}
 		return nil, err
 	}
 
diff --git a/p2p/transport/websocket/conn.go b/p2p/transport/websocket/conn.go
index 70d4abbd..c3918c38 100644
--- a/p2p/transport/websocket/conn.go
+++ b/p2p/transport/websocket/conn.go
@@ -31,7 +31,6 @@ func (c conn) Read(b []byte) (int, error) {
 	if err == nil && n == 0 && c.readAttempts < maxReadAttempts {
 		c.readAttempts++
 		// Nothing happened, let's read again. We reached the end of the frame
-		// we have
 		// (https://github.com/nhooyr/websocket/blob/master/netconn.go#L118).
 		// The next read will block until we get
 		// the next frame. We limit here to avoid looping in case of a bunch of
diff --git a/p2p/transport/websocket/listener.go b/p2p/transport/websocket/listener.go
index 70dbce0f..ab9a73f8 100644
--- a/p2p/transport/websocket/listener.go
+++ b/p2p/transport/websocket/listener.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"math"
 	"net"
 	"net/http"
 	"net/url"
@@ -19,6 +20,9 @@ import (
 type listener struct {
 	nl     net.Listener
 	server http.Server
+	// The Go standard library sets the http.Server.TLSConfig no matter if this is a WS or WSS,
+	// so we can't rely on checking if server.TLSConfig is set.
+	isWss bool
 
 	laddr ma.Multiaddr
 
@@ -80,6 +84,7 @@ func newListener(a ma.Multiaddr, tlsConf *tls.Config) (*listener, error) {
 	}
 	ln.server = http.Server{Handler: ln}
 	if parsed.isWSS {
+		ln.isWss = true
 		ln.server.TLSConfig = tlsConf
 	}
 	return ln, nil
@@ -87,7 +92,7 @@ func newListener(a ma.Multiaddr, tlsConf *tls.Config) (*listener, error) {
 
 func (l *listener) serve() {
 	defer close(l.closed)
-	if l.server.TLSConfig == nil {
+	if !l.isWss {
 		l.server.Serve(l.nl)
 	} else {
 		l.server.ServeTLS(l.nl, "", "")
@@ -96,7 +101,7 @@ func (l *listener) serve() {
 
 func (l *listener) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	scheme := "ws"
-	if l.server.TLSConfig != nil {
+	if l.isWss {
 		scheme = "wss"
 	}
 
@@ -109,6 +114,10 @@ func (l *listener) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// Set an arbitrarily large read limit since we don't actually want to limit the message size here.
+	// See https://github.com/nhooyr/websocket/issues/382 for details.
+	c.SetReadLimit(math.MaxInt64 - 1) // -1 because the library adds a byte for the fin frame
+
 	select {
 	case l.incoming <- conn{
 		Conn: ws.NetConn(context.Background(), c, ws.MessageBinary),
diff --git a/p2p/transport/websocket/websocket.go b/p2p/transport/websocket/websocket.go
index f3a3c870..a78add97 100644
--- a/p2p/transport/websocket/websocket.go
+++ b/p2p/transport/websocket/websocket.go
@@ -5,6 +5,7 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"math"
 	"net"
 	"net/http"
 	"net/url"
@@ -250,6 +251,8 @@ func (t *WebsocketTransport) maDial(ctx context.Context, raddr ma.Multiaddr) (ma
 		return nil, fmt.Errorf("failed to get local address")
 	}
 
+	// Set an arbitrarily large read limit since we don't actually want to limit the message size here.
+	wscon.SetReadLimit(math.MaxInt64 - 1) // -1 because the library adds a byte for the fin frame
 	mnc, err := manet.WrapNetConn(
 		conn{
 			Conn:       ws.NetConn(context.Background(), wscon, ws.MessageBinary),


Test patch:
diff --git a/p2p/test/websocket/websocket_test.go b/p2p/test/websocket/websocket_test.go
new file mode 100644
index 00000000..ccadb6ff
--- /dev/null
+++ b/p2p/test/websocket/websocket_test.go
@@ -0,0 +1,66 @@
+package websocket_test
+
+import (
+	"context"
+	"io"
+	"testing"
+
+	"github.com/libp2p/go-libp2p"
+	"github.com/libp2p/go-libp2p/core/network"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/stretchr/testify/require"
+)
+
+func TestReadLimit(t *testing.T) {
+	h1, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0/ws"))
+	require.NoError(t, err)
+	defer h1.Close()
+
+	ctx := context.Background()
+	h2, err := libp2p.New(libp2p.NoListenAddrs)
+	require.NoError(t, err)
+	defer h2.Close()
+
+	err = h2.Connect(ctx, peer.AddrInfo{ID: h1.ID(), Addrs: h1.Addrs()})
+	require.NoError(t, err)
+
+	buf := make([]byte, 256<<10)
+	buf2 := make([]byte, 256<<10)
+	copyBuf := make([]byte, 8<<10)
+
+	errCh := make(chan error, 1)
+	// TODO perf would be perfect here, but not yet merged.
+	h1.SetStreamHandler("/big-blocks", func(s network.Stream) {
+		defer s.Close()
+		_, err := io.CopyBuffer(io.Discard, s, copyBuf)
+		if err != nil {
+			errCh <- err
+			return
+		}
+		_, err = s.Write(buf)
+		if err != nil {
+			errCh <- err
+			return
+		}
+		errCh <- nil
+	})
+
+	allocs := testing.AllocsPerRun(100, func() {
+		s, err := h2.NewStream(ctx, h1.ID(), "/big-blocks")
+		require.NoError(t, err)
+		defer s.Close()
+		_, err = s.Write(buf2)
+		require.NoError(t, err)
+		require.NoError(t, s.CloseWrite())
+
+		_, err = io.ReadFull(s, buf2)
+		require.NoError(t, err)
+
+		_, err = s.Read([]byte{0})
+		require.ErrorIs(t, err, io.EOF)
+		require.NoError(t, <-errCh)
+	})
+
+	// Make sure we aren't doing some crazy allocs when transferring big blocks
+	require.Less(t, allocs, 8*1024.0)
+}
diff --git a/p2p/transport/websocket/websocket_test.go b/p2p/transport/websocket/websocket_test.go
index 0c7fc5f9..10878397 100644
--- a/p2p/transport/websocket/websocket_test.go
+++ b/p2p/transport/websocket/websocket_test.go
@@ -308,6 +308,16 @@ func TestWebsocketTransport(t *testing.T) {
 	ttransport.SubtestTransport(t, ta, tb, "/ip4/127.0.0.1/tcp/0/ws", "peerA")
 }
 
+func isWSS(addr ma.Multiaddr) bool {
+	if _, err := addr.ValueForProtocol(ma.P_WSS); err == nil {
+		return true
+	}
+	if _, err := addr.ValueForProtocol(ma.P_WS); err == nil {
+		return false
+	}
+	panic("not a WebSocket address")
+}
+
 func connectAndExchangeData(t *testing.T, laddr ma.Multiaddr, secure bool) {
 	var opts []Option
 	var tlsConf *tls.Config
@@ -339,6 +349,8 @@ func connectAndExchangeData(t *testing.T, laddr ma.Multiaddr, secure bool) {
 		require.NoError(t, err)
 		c, err := tpt.Dial(context.Background(), l.Multiaddr(), server)
 		require.NoError(t, err)
+		require.Equal(t, secure, isWSS(c.LocalMultiaddr()))
+		require.Equal(t, secure, isWSS(c.RemoteMultiaddr()))
 		str, err := c.OpenStream(context.Background())
 		require.NoError(t, err)
 		defer str.Close()
@@ -349,6 +361,8 @@ func connectAndExchangeData(t *testing.T, laddr ma.Multiaddr, secure bool) {
 	c, err := l.Accept()
 	require.NoError(t, err)
 	defer c.Close()
+	require.Equal(t, secure, isWSS(c.LocalMultiaddr()))
+	require.Equal(t, secure, isWSS(c.RemoteMultiaddr()))
 	str, err := c.AcceptStream()
 	require.NoError(t, err)
 	defer str.Close()