WIP: Feat/memory transport

p2p/test/transport/transport_test.go

@@ -31,6 +31,7 @@ import (
 	"github.com/libp2p/go-libp2p/p2p/protocol/ping"
 	"github.com/libp2p/go-libp2p/p2p/security/noise"
 	tls "github.com/libp2p/go-libp2p/p2p/security/tls"
+	libp2pmemory "github.com/libp2p/go-libp2p/p2p/transport/memory"
 	libp2pwebrtc "github.com/libp2p/go-libp2p/p2p/transport/webrtc"
 	"go.uber.org/mock/gomock"
 
@@ -156,6 +157,21 @@ var transportsToTest = []TransportTestCase{
 			return h
 		},
 	},
+	{
+		Name: "Memory",
+		HostGenerator: func(t *testing.T, opts TransportTestCaseOpts) host.Host {
+			libp2pOpts := transformOpts(opts)
+			libp2pOpts = append(libp2pOpts, libp2p.Transport(libp2pmemory.NewTransport))
+			if opts.NoListen {
+				libp2pOpts = append(libp2pOpts, libp2p.NoListenAddrs)
+			} else {
+				libp2pOpts = append(libp2pOpts, libp2p.ListenAddrStrings("/memory/1234"))
+			}
+			h, err := libp2p.New(libp2pOpts...)
+			require.NoError(t, err)
+			return h
+		},
+	},
 }
 
 func TestPing(t *testing.T) {

p2p/transport/memory/conn.go

@@ -0,0 +1,125 @@
+package memory
+
+import (
+	"context"
+	"io"
+	"sync"
+	"sync/atomic"
+
+	ic "github.com/libp2p/go-libp2p/core/crypto"
+	"github.com/libp2p/go-libp2p/core/network"
+	"github.com/libp2p/go-libp2p/core/peer"
+	tpt "github.com/libp2p/go-libp2p/core/transport"
+	ma "github.com/multiformats/go-multiaddr"
+)
+
+type conn struct {
+	id int32
+
+	transport *transport
+	scope     network.ConnManagementScope
+
+	localPeer      peer.ID
+	localMultiaddr ma.Multiaddr
+
+	remotePeerID    peer.ID
+	remotePubKey    ic.PubKey
+	remoteMultiaddr ma.Multiaddr
+
+	isClosed  atomic.Bool
+	closeOnce sync.Once
+
+	mu sync.Mutex
+
+	streamC chan *stream
+
+	nextStreamID atomic.Int32
+	streams      map[int32]network.MuxedStream
+}
+
+var _ tpt.CapableConn = &conn{}
+
+func newConnection(id int32, s *stream) *conn {
+	c := &conn{
+		id:      id,
+		streamC: make(chan *stream, 1),
+		streams: make(map[int32]network.MuxedStream),
+	}
+
+	streamID := c.nextStreamID.Add(1)
+	c.addStream(streamID, s)
+
+	return c
+}
+
+func (c *conn) Close() error {
+	c.closeOnce.Do(func() {
+		c.isClosed.Store(true)
+		c.transport.removeConn(c)
+	})
+
+	return nil
+}
+
+func (c *conn) IsClosed() bool {
+	return c.isClosed.Load()
+}
+
+func (c *conn) OpenStream(ctx context.Context) (network.MuxedStream, error) {
+	id := c.nextStreamID.Add(1)
+	// TODO: Figure out how to exchange the pipes between the two streams
+	ra, wa := io.Pipe()
+
+	return newStream(id, ra, wa), nil
+}
+
+func (c *conn) AcceptStream() (network.MuxedStream, error) {
+	select {
+	case in := <-c.streamC:
+		id := c.nextStreamID.Add(1)
+		c.addStream(id, in)
+
+		return in, nil
+	}
+}
+
+func (c *conn) LocalPeer() peer.ID { return c.localPeer }
+
+// RemotePeer returns the peer ID of the remote peer.
+func (c *conn) RemotePeer() peer.ID { return c.remotePeerID }
+
+// RemotePublicKey returns the public pkey of the remote peer.
+func (c *conn) RemotePublicKey() ic.PubKey { return c.remotePubKey }
+
+// LocalMultiaddr returns the local Multiaddr associated
+func (c *conn) LocalMultiaddr() ma.Multiaddr { return c.localMultiaddr }
+
+// RemoteMultiaddr returns the remote Multiaddr associated
+func (c *conn) RemoteMultiaddr() ma.Multiaddr { return c.remoteMultiaddr }
+
+func (c *conn) Transport() tpt.Transport {
+	return c.transport
+}
+
+func (c *conn) Scope() network.ConnScope {
+	return c.scope
+}
+
+// ConnState is the state of security connection.
+func (c *conn) ConnState() network.ConnectionState {
+	return network.ConnectionState{Transport: "memory"}
+}
+
+func (c *conn) addStream(id int32, stream network.MuxedStream) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.streams[id] = stream
+}
+
+func (c *conn) removeStream(id int32) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	delete(c.streams, id)
+}

p2p/transport/memory/listener.go

@@ -0,0 +1,60 @@
+package memory
+
+import (
+	"context"
+	tpt "github.com/libp2p/go-libp2p/core/transport"
+	ma "github.com/multiformats/go-multiaddr"
+	"net"
+	"sync"
+)
+
+type listener struct {
+	t      *transport
+	ctx    context.Context
+	cancel context.CancelFunc
+	laddr  ma.Multiaddr
+
+	mu          sync.Mutex
+	connCh      chan *conn
+	connections map[int32]*conn
+}
+
+func (l *listener) Multiaddr() ma.Multiaddr {
+	return l.laddr
+}
+
+func newListener(t *transport, laddr ma.Multiaddr) *listener {
+	ctx, cancel := context.WithCancel(context.Background())
+	return &listener{
+		t:      t,
+		ctx:    ctx,
+		cancel: cancel,
+		laddr:  laddr,
+		connCh: make(chan *conn, listenerQueueSize),
+	}
+}
+
+// Accept accepts new connections.
+func (l *listener) Accept() (tpt.CapableConn, error) {
+	select {
+	case c := <-l.connCh:
+		l.mu.Lock()
+		defer l.mu.Unlock()
+
+		l.connections[c.id] = c
+		return c, nil
+	case <-l.ctx.Done():
+		return nil, l.ctx.Err()
+	}
+}
+
+// Close closes the listener.
+func (l *listener) Close() error {
+	l.cancel()
+	return nil
+}
+
+// Addr returns the address of this listener.
+func (l *listener) Addr() net.Addr {
+	return nil
+}

p2p/transport/memory/stream.go

@@ -0,0 +1,108 @@
+package memory
+
+import (
+	"errors"
+	"io"
+	"sync/atomic"
+	"time"
+
+	"github.com/libp2p/go-libp2p/core/network"
+)
+
+type stream struct {
+	id int32
+
+	r *io.PipeReader
+	w *io.PipeWriter
+
+	readCloseC  chan struct{}
+	writeCloseC chan struct{}
+
+	closed atomic.Bool
+}
+
+func newStream(id int32, r *io.PipeReader, w *io.PipeWriter) *stream {
+	return &stream{
+		id:          id,
+		r:           r,
+		w:           w,
+		readCloseC:  make(chan struct{}, 1),
+		writeCloseC: make(chan struct{}, 1),
+	}
+}
+
+func (s *stream) Read(b []byte) (int, error) {
+	if s.closed.Load() {
+		return 0, network.ErrReset
+	}
+
+	select {
+	case <-s.readCloseC:
+		return 0, network.ErrReset
+	default:
+		return s.r.Read(b)
+	}
+}
+
+func (s *stream) Write(b []byte) (int, error) {
+	if s.closed.Load() {
+		return 0, network.ErrReset
+	}
+
+	select {
+	case <-s.writeCloseC:
+		return 0, network.ErrReset
+	default:
+		return s.w.Write(b)
+	}
+}
+
+func (s *stream) Reset() error {
+	if err := s.CloseWrite(); err != nil {
+		return err
+	}
+	if err := s.CloseRead(); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (s *stream) Close() error {
+	if err := s.CloseRead(); err != nil {
+		return err
+	}
+
+	s.closed.Store(true)
+	return nil
+}
+
+func (s *stream) CloseRead() error {
+	select {
+	case s.readCloseC <- struct{}{}:
+	default:
+		return errors.New("failed to close stream read")
+	}
+
+	return nil
+}
+
+func (s *stream) CloseWrite() error {
+	select {
+	case s.writeCloseC <- struct{}{}:
+	default:
+		return errors.New("failed to close stream write")
+	}
+
+	return nil
+}
+
+func (s *stream) SetDeadline(_ time.Time) error {
+	return nil
+}
+
+func (s *stream) SetReadDeadline(_ time.Time) error {
+	return nil
+}
+func (s *stream) SetWriteDeadline(_ time.Time) error {
+	return nil
+}

p2p/transport/memory/stream_test.go

@@ -0,0 +1,55 @@
+package memory
+
+import (
+	"github.com/stretchr/testify/require"
+	"io"
+	"testing"
+)
+
+func TestStreamSimpleReadWriteClose(t *testing.T) {
+	//client, server := getDetachedDataChannels(t)
+	ra, wb := io.Pipe()
+	rb, wa := io.Pipe()
+
+	clientStr := newStream(0, ra, wa)
+	serverStr := newStream(0, rb, wb)
+
+	// send a foobar from the client
+	n, err := clientStr.Write([]byte("foobar"))
+	require.NoError(t, err)
+	require.Equal(t, 6, n)
+	require.NoError(t, clientStr.CloseWrite())
+	// writing after closing should error
+	_, err = clientStr.Write([]byte("foobar"))
+	require.Error(t, err)
+	//require.False(t, clientDone.Load())
+
+	// now read all the data on the server side
+	b, err := io.ReadAll(serverStr)
+	require.NoError(t, err)
+	require.Equal(t, []byte("foobar"), b)
+	// reading again should give another io.EOF
+	n, err = serverStr.Read(make([]byte, 10))
+	require.Zero(t, n)
+	require.ErrorIs(t, err, io.EOF)
+	//require.False(t, serverDone.Load())
+
+	// send something back
+	_, err = serverStr.Write([]byte("lorem ipsum"))
+	require.NoError(t, err)
+	require.NoError(t, serverStr.CloseWrite())
+
+	// and read it at the client
+	//require.False(t, clientDone.Load())
+	b, err = io.ReadAll(clientStr)
+	require.NoError(t, err)
+	require.Equal(t, []byte("lorem ipsum"), b)
+
+	// stream is only cleaned up on calling Close or Reset
+	clientStr.Close()
+	serverStr.Close()
+	//require.Eventually(t, func() bool { return clientDone.Load() }, 5*time.Second, 100*time.Millisecond)
+	// Need to call Close for cleanup. Otherwise the FIN_ACK is never read
+	require.NoError(t, serverStr.Close())
+	//require.Eventually(t, func() bool { return serverDone.Load() }, 5*time.Second, 100*time.Millisecond)
+}