net/conn: io, not channels

This commit changes the connections to use io.ReadWriters
instead of channels (+ async workers). This is a pretty
big change -- away from csp -- in the name of performance
(and predictable flow control).

It also uses the brand new secio, which is spipe's successor.

net/conn/conn.go

@@ -18,9 +18,6 @@ import (
 var log = u.Logger("conn")
 
 const (
-	// ChanBuffer is the size of the buffer in the Conn Chan
-	ChanBuffer = 10
-
 	// MaxMessageSize is the size of the largest single message. (4MB)
 	MaxMessageSize = 1 << 22
 
@@ -35,25 +32,12 @@ func ReleaseBuffer(b []byte) {
 	mpool.ByteSlicePool.Put(uint32(cap(b)), b)
 }
 
-// msgioPipe is a pipe using msgio channels.
-type msgioPipe struct {
-	outgoing *msgio.Chan
-	incoming *msgio.Chan
-}
-
-func newMsgioPipe(size int) *msgioPipe {
-	return &msgioPipe{
-		outgoing: msgio.NewChan(size),
-		incoming: msgio.NewChan(size),
-	}
-}
-
 // singleConn represents a single connection to another Peer (IPFS Node).
 type singleConn struct {
 	local  peer.Peer
 	remote peer.Peer
 	maconn manet.Conn
-	msgio  *msgioPipe
+	msgrw  msgio.ReadWriteCloser
 
 	ctxc.ContextCloser
 }
@@ -66,24 +50,12 @@ func newSingleConn(ctx context.Context, local, remote peer.Peer,
 		local:  local,
 		remote: remote,
 		maconn: maconn,
-		msgio:  newMsgioPipe(10),
+		msgrw:  msgio.NewReadWriter(maconn),
 	}
 
 	conn.ContextCloser = ctxc.NewContextCloser(ctx, conn.close)
 
-	log.Debugf("newSingleConn: %v to %v", local, remote)
-
-	// setup the various io goroutines
-	conn.Children().Add(1)
-	go func() {
-		conn.msgio.outgoing.WriteTo(maconn)
-		conn.Children().Done()
-	}()
-	conn.Children().Add(1)
-	go func() {
-		conn.msgio.incoming.ReadFromWithPool(maconn, &mpool.ByteSlicePool)
-		conn.Children().Done()
-	}()
+	log.Debugf("newSingleConn %p: %v to %v", conn, local, remote)
 
 	// version handshake
 	ctxT, _ := context.WithTimeout(ctx, HandshakeTimeout)
@@ -92,6 +64,7 @@ func newSingleConn(ctx context.Context, local, remote peer.Peer,
 		return nil, fmt.Errorf("Handshake1 failed: %s", err)
 	}
 
+	log.Debugf("newSingleConn %p: %v to %v finished", conn, local, remote)
 	return conn, nil
 }
 
@@ -100,20 +73,7 @@ func (c *singleConn) close() error {
 	log.Debugf("%s closing Conn with %s", c.local, c.remote)
 
 	// close underlying connection
-	err := c.maconn.Close()
-	c.msgio.outgoing.Close()
-	return err
-}
-
-func (c *singleConn) GetError() error {
-	select {
-	case err := <-c.msgio.incoming.ErrChan:
-		return err
-	case err := <-c.msgio.outgoing.ErrChan:
-		return err
-	default:
-		return nil
-	}
+	return c.msgrw.Close()
 }
 
 // ID is an identifier unique to this connection.
@@ -145,14 +105,29 @@ func (c *singleConn) RemotePeer() peer.Peer {
 	return c.remote
 }
 
-// In returns a readable message channel
-func (c *singleConn) In() <-chan []byte {
-	return c.msgio.incoming.MsgChan
+// Read reads data, net.Conn style
+func (c *singleConn) Read(buf []byte) (int, error) {
+	return c.msgrw.Read(buf)
+}
+
+// Write writes data, net.Conn style
+func (c *singleConn) Write(buf []byte) (int, error) {
+	return c.msgrw.Write(buf)
+}
+
+// ReadMsg reads data, net.Conn style
+func (c *singleConn) ReadMsg() ([]byte, error) {
+	return c.msgrw.ReadMsg()
+}
+
+// WriteMsg writes data, net.Conn style
+func (c *singleConn) WriteMsg(buf []byte) error {
+	return c.msgrw.WriteMsg(buf)
 }
 
-// Out returns a writable message channel
-func (c *singleConn) Out() chan<- []byte {
-	return c.msgio.outgoing.MsgChan
+// ReleaseMsg releases a buffer
+func (c *singleConn) ReleaseMsg(m []byte) {
+	c.msgrw.ReleaseMsg(m)
 }
 
 // ID returns the ID of a given Conn.
@@ -167,6 +142,6 @@ func ID(c Conn) string {
 
 // String returns the user-friendly String representation of a conn
 func String(c Conn, typ string) string {
-	return fmt.Sprintf("%s (%s) <-- %s --> (%s) %s",
-		c.LocalPeer(), c.LocalMultiaddr(), typ, c.RemoteMultiaddr(), c.RemotePeer())
+	return fmt.Sprintf("%s (%s) <-- %s %p --> (%s) %s",
+		c.LocalPeer(), c.LocalMultiaddr(), typ, c, c.RemoteMultiaddr(), c.RemotePeer())
 }

net/conn/conn_test.go

@@ -100,18 +100,24 @@ func TestCloseLeak(t *testing.T) {
 		c1, c2 := setupConn(t, ctx, "/ip4/127.0.0.1/tcp/"+a1, "/ip4/127.0.0.1/tcp/"+a2)
 
 		for i := 0; i < num; i++ {
-			b1 := []byte("beep")
-			c1.Out() <- b1
-			b2 := <-c2.In()
+			b1 := []byte(fmt.Sprintf("beep%d", i))
+			c1.WriteMsg(b1)
+			b2, err := c2.ReadMsg()
+			if err != nil {
+				panic(err)
+			}
 			if !bytes.Equal(b1, b2) {
-				panic("bytes not equal")
+				panic(fmt.Errorf("bytes not equal: %s != %s", b1, b2))
 			}
 
-			b2 = []byte("boop")
-			c2.Out() <- b2
-			b1 = <-c1.In()
+			b2 = []byte(fmt.Sprintf("boop%d", i))
+			c2.WriteMsg(b2)
+			b1, err = c1.ReadMsg()
+			if err != nil {
+				panic(err)
+			}
 			if !bytes.Equal(b1, b2) {
-				panic("bytes not equal")
+				panic(fmt.Errorf("bytes not equal: %s != %s", b1, b2))
 			}
 
 			<-time.After(time.Microsecond * 5)
@@ -123,7 +129,7 @@ func TestCloseLeak(t *testing.T) {
 		wg.Done()
 	}
 
-	var cons = 20
+	var cons = 1
 	var msgs = 100
 	fmt.Printf("Running %d connections * %d msgs.\n", cons, msgs)
 	for i := 0; i < cons; i++ {

net/conn/dial.go

@@ -72,5 +72,6 @@ func (d *Dialer) DialAddr(ctx context.Context, raddr ma.Multiaddr, remote peer.P
 		return nil, err
 	}
 
+	// return c, nil
 	return newSecureConn(ctx, c, d.Peerstore)
 }

net/conn/dial_test.go

 package conn
 
 import (
+	"io"
 	"testing"
 
 	ci "github.com/jbenet/go-ipfs/crypto"
@@ -42,14 +43,7 @@ func echoListen(ctx context.Context, listener Listener) {
 }
 
 func echo(ctx context.Context, c Conn) {
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case m := <-c.In():
-			c.Out() <- m
-		}
-	}
+	io.Copy(c, c)
 }
 
 func setupConn(t *testing.T, ctx context.Context, a1, a2 string) (a, b Conn) {
@@ -137,17 +131,25 @@ func TestDialer(t *testing.T) {
 	}
 
 	// fmt.Println("sending")
-	c.Out() <- []byte("beep")
-	c.Out() <- []byte("boop")
+	c.WriteMsg([]byte("beep"))
+	c.WriteMsg([]byte("boop"))
+
+	out, err := c.ReadMsg()
+	if err != nil {
+		t.Fatal(err)
+	}
 
-	out := <-c.In()
 	// fmt.Println("recving", string(out))
 	data := string(out)
 	if data != "beep" {
 		t.Error("unexpected conn output", data)
 	}
 
-	out = <-c.In()
+	out, err = c.ReadMsg()
+	if err != nil {
+		t.Fatal(err)
+	}
+
 	data = string(out)
 	if string(out) != "boop" {
 		t.Error("unexpected conn output", data)
@@ -207,17 +209,24 @@ func TestDialAddr(t *testing.T) {
 	}
 
 	// fmt.Println("sending")
-	c.Out() <- []byte("beep")
-	c.Out() <- []byte("boop")
+	c.WriteMsg([]byte("beep"))
+	c.WriteMsg([]byte("boop"))
 
-	out := <-c.In()
+	out, err := c.ReadMsg()
+	if err != nil {
+		t.Fatal(err)
+	}
 	// fmt.Println("recving", string(out))
 	data := string(out)
 	if data != "beep" {
 		t.Error("unexpected conn output", data)
 	}
 
-	out = <-c.In()
+	out, err = c.ReadMsg()
+	if err != nil {
+		t.Fatal(err)
+	}
+
 	data = string(out)
 	if string(out) != "boop" {
 		t.Error("unexpected conn output", data)

net/conn/handshake.go

 package conn
 
 import (
-	"errors"
 	"fmt"
 
 	handshake "github.com/jbenet/go-ipfs/net/handshake"
@@ -25,29 +24,22 @@ func Handshake1(ctx context.Context, c Conn) error {
 		return err
 	}
 
-	c.Out() <- myVerBytes
-	log.Debugf("Sent my version (%s) to %s", localH, rpeer)
-
-	select {
-	case <-ctx.Done():
-		return ctx.Err()
-
-	case <-c.Closing():
-		return errors.New("remote closed connection during version exchange")
-
-	case data, ok := <-c.In():
-		if !ok {
-			return fmt.Errorf("error retrieving from conn: %v", rpeer)
-		}
+	if err := CtxWriteMsg(ctx, c, myVerBytes); err != nil {
+		return err
+	}
+	log.Debugf("%p sent my version (%s) to %s", c, localH, rpeer)
 
-		remoteH = new(hspb.Handshake1)
-		err = proto.Unmarshal(data, remoteH)
-		if err != nil {
-			return fmt.Errorf("could not decode remote version: %q", err)
-		}
+	data, err := CtxReadMsg(ctx, c)
+	if err != nil {
+		return err
+	}
 
-		log.Debugf("Received remote version (%s) from %s", remoteH, rpeer)
+	remoteH = new(hspb.Handshake1)
+	err = proto.Unmarshal(data, remoteH)
+	if err != nil {
+		return fmt.Errorf("could not decode remote version: %q", err)
 	}
+	log.Debugf("%p received remote version (%s) from %s", c, remoteH, rpeer)
 
 	if err := handshake.Handshake1Compatible(localH, remoteH); err != nil {
 		log.Infof("%s (%s) incompatible version with %s (%s)", lpeer, localH, rpeer, remoteH)
@@ -71,31 +63,25 @@ func Handshake3(ctx context.Context, c Conn) (*handshake.Handshake3Result, error
 		return nil, err
 	}
 
-	c.Out() <- localB
+	if err := CtxWriteMsg(ctx, c, localB); err != nil {
+		return nil, err
+	}
 	log.Debugf("Handshake1: sent to %s", rpeer)
 
 	// wait + listen for response
-	select {
-	case <-ctx.Done():
-		return nil, ctx.Err()
-
-	case <-c.Closing():
-		return nil, errors.New("Handshake3: error remote connection closed")
-
-	case remoteB, ok := <-c.In():
-		if !ok {
-			return nil, fmt.Errorf("Handshake3 error receiving from conn: %v", rpeer)
-		}
-
-		remoteH = new(hspb.Handshake3)
-		err = proto.Unmarshal(remoteB, remoteH)
-		if err != nil {
-			return nil, fmt.Errorf("Handshake3 could not decode remote msg: %q", err)
-		}
+	remoteB, err := CtxReadMsg(ctx, c)
+	if err != nil {
+		return nil, err
+	}
 
-		log.Debugf("Handshake3 received from %s", rpeer)
+	remoteH = new(hspb.Handshake3)
+	err = proto.Unmarshal(remoteB, remoteH)
+	if err != nil {
+		return nil, fmt.Errorf("Handshake3 could not decode remote msg: %q", err)
 	}
 
+	log.Debugf("Handshake3 received from %s", rpeer)
+
 	// actually update our state based on the new knowledge
 	res, err := handshake.Handshake3Update(lpeer, rpeer, remoteH)
 	if err != nil {

net/conn/interface.go

 package conn
 
 import (
+	"errors"
+
 	peer "github.com/jbenet/go-ipfs/peer"
 	u "github.com/jbenet/go-ipfs/util"
 	ctxc "github.com/jbenet/go-ipfs/util/ctxcloser"
 
+	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
+	msgio "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-msgio"
 	ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
 )
 
@@ -31,15 +35,8 @@ type Conn interface {
 	// RemotePeer is the Peer on the remote side
 	RemotePeer() peer.Peer
 
-	// In returns a readable message channel
-	In() <-chan []byte
-
-	// Out returns a writable message channel
-	Out() chan<- []byte
-
-	// Get an error from this conn if one is available
-	// TODO: implement a better error handling system
-	GetError() error
+	msgio.Reader
+	msgio.Writer
 }
 
 // Dialer is an object that can open connections. We could have a "convenience"
@@ -77,3 +74,89 @@ type Listener interface {
 	// Any blocked Accept operations will be unblocked and return errors.
 	Close() error
 }
+
+// CtxRead is a function that Reads from a connection while respecting a
+// Context. Though it cannot cancel the read per-se (as not all Connections
+// implement SetTimeout, and a CancelFunc can't be predicted), at least it
+// doesn't hang. The Read will eventually return and the goroutine will exit.
+func CtxRead(ctx context.Context, c Conn, buf []byte) (n int, err error) {
+	done := make(chan struct{})
+	go func() {
+		n, err = c.Read(buf)
+		close(done)
+	}()
+
+	select {
+	case <-ctx.Done():
+		return 0, ctx.Err()
+
+	case <-c.Closing():
+		return 0, errors.New("remote connection closed")
+
+	case <-done:
+		return n, err
+	}
+}
+
+// CtxReadMsg is a function that Reads from a connection while respecting a
+// Context. See CtxRead.
+func CtxReadMsg(ctx context.Context, c Conn) (msg []byte, err error) {
+	done := make(chan struct{})
+	go func() {
+		msg, err = c.ReadMsg()
+		close(done)
+	}()
+
+	select {
+	case <-ctx.Done():
+		return msg, ctx.Err()
+
+	case <-c.Closing():
+		return msg, errors.New("remote connection closed")
+
+	case <-done:
+		return msg, err
+	}
+}
+
+// CtxWrite is a function that Writes to a connection while respecting a
+// Context. See CtxRead.
+func CtxWrite(ctx context.Context, c Conn, buf []byte) (n int, err error) {
+	done := make(chan struct{})
+	go func() {
+		n, err = c.Read(buf)
+		close(done)
+	}()
+
+	select {
+	case <-ctx.Done():
+		return 0, ctx.Err()
+
+	case <-c.Closing():
+		return 0, errors.New("remote connection closed")
+
+	case <-done:
+		return n, err
+	}
+}
+
+// CtxWriteMsg is a function that Writes to a connection while respecting a
+// Context. See CtxRead.
+func CtxWriteMsg(ctx context.Context, c Conn, buf []byte) (err error) {
+	done := make(chan struct{})
+	go func() {
+		err = c.WriteMsg(buf)
+		close(done)
+	}()
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+
+	case <-c.Closing():
+		return errors.New("remote connection closed")
+
+	case <-done:
+		return err
+	}
+}

net/conn/listen.go

@@ -65,12 +65,15 @@ func (l *listener) listen() {
 			return
 		}
 
+		// if insecure:
+		// l.conns <- c
+
+		// if secure
 		sc, err := newSecureConn(l.ctx, c, l.peers)
 		if err != nil {
 			log.Errorf("Error securing connection: %v", err)
 			return
 		}
-
 		l.conns <- sc
 	}
 

net/conn/multiconn.go

 package conn
 
 import (
+	"errors"
 	"sync"
 
 	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
@@ -14,12 +15,6 @@ import (
 // MultiConnMap is for shorthand
 type MultiConnMap map[u.Key]*MultiConn
 
-// Duplex is a simple duplex channel
-type Duplex struct {
-	In  chan []byte
-	Out chan []byte
-}
-
 // MultiConn represents a single connection to another Peer (IPFS Node).
 type MultiConn struct {
 
@@ -30,8 +25,8 @@ type MultiConn struct {
 	local  peer.Peer
 	remote peer.Peer
 
-	// fan-in/fan-out
-	duplex Duplex
+	// fan-in
+	fanIn chan []byte
 
 	// for adding/removing connections concurrently
 	sync.RWMutex
@@ -45,10 +40,7 @@ func NewMultiConn(ctx context.Context, local, remote peer.Peer, conns []Conn) (*
 		local:  local,
 		remote: remote,
 		conns:  map[string]Conn{},
-		duplex: Duplex{
-			In:  make(chan []byte, 10),
-			Out: make(chan []byte, 10),
-		},
+		fanIn:  make(chan []byte),
 	}
 
 	// must happen before Adds / fanOut
@@ -58,8 +50,6 @@ func NewMultiConn(ctx context.Context, local, remote peer.Peer, conns []Conn) (*
 		c.Add(conns...)
 	}
 
-	c.Children().Add(1)
-	go c.fanOut()
 	return c, nil
 }
 
@@ -135,38 +125,8 @@ func CloseConns(conns ...Conn) {
 	wg.Wait()
 }
 
-// fanOut is the multiplexor out -- it sends outgoing messages over the
-// underlying single connections.
-func (c *MultiConn) fanOut() {
-	defer c.Children().Done()
-
-	i := 0
-	for {
-		select {
-		case <-c.Closing():
-			return
-
-		// send data out through our "best connection"
-		case m, more := <-c.duplex.Out:
-			if !more {
-				log.Debugf("%s out channel closed", c)
-				return
-			}
-			sc := c.BestConn()
-			if sc == nil {
-				// maybe this should be a logged error, not a panic.
-				panic("sending out multiconn without any live connection")
-			}
-
-			i++
-			log.Debugf("%s sending (%d)", sc, i)
-			sc.Out() <- m
-		}
-	}
-}
-
-// fanInSingle is a multiplexor in -- it receives incoming messages over the
-// underlying single connections.
+// fanInSingle Reads from a connection, and sends to the fanIn.
+// waits for child to close and reclaims resources
 func (c *MultiConn) fanInSingle(child Conn) {
 	// cleanup all data associated with this child Connection.
 	defer func() {
@@ -186,8 +146,13 @@ func (c *MultiConn) fanInSingle(child Conn) {
 		}
 	}()
 
-	i := 0
 	for {
+		msg, err := child.ReadMsg()
+		if err != nil {
+			log.Warning(err)
+			return
+		}
+
 		select {
 		case <-c.Closing(): // multiconn closing
 			return
@@ -195,18 +160,7 @@ func (c *MultiConn) fanInSingle(child Conn) {
 		case <-child.Closing(): // child closing
 			return
 
-		case m, more := <-child.In(): // receiving data
-			if !more {
-				log.Debugf("%s in channel closed", child)
-				err := c.GetError()
-				if err != nil {
-					log.Errorf("Found error on connection: %s", err)
-				}
-				return // closed
-			}
-			i++
-			log.Debugf("%s received (%d)", child, i)
-			c.duplex.In <- m
+		case c.fanIn <- msg:
 		}
 	}
 }
@@ -296,24 +250,38 @@ func (c *MultiConn) RemotePeer() peer.Peer {
 	return c.remote
 }
 
-// In returns a readable message channel
-func (c *MultiConn) In() <-chan []byte {
-	return c.duplex.In
+// Read reads data, net.Conn style
+func (c *MultiConn) Read(buf []byte) (int, error) {
+	return 0, errors.New("multiconn does not support Read. use ReadMsg")
 }
 
-// Out returns a writable message channel
-func (c *MultiConn) Out() chan<- []byte {
-	return c.duplex.Out
+// Write writes data, net.Conn style
+func (c *MultiConn) Write(buf []byte) (int, error) {
+	bc := c.BestConn()
+	if bc == nil {
+		return 0, errors.New("no best connection")
+	}
+	return bc.Write(buf)
 }
 
-func (c *MultiConn) GetError() error {
-	c.RLock()
-	defer c.RUnlock()
-	for _, sub := range c.conns {
-		err := sub.GetError()
-		if err != nil {
-			return err
-		}
+// ReadMsg reads data, net.Conn style
+func (c *MultiConn) ReadMsg() ([]byte, error) {
+	next := <-c.fanIn
+	return next, nil
+}
+
+// WriteMsg writes data, net.Conn style
+func (c *MultiConn) WriteMsg(buf []byte) error {
+	bc := c.BestConn()
+	if bc == nil {
+		return errors.New("no best connection")
+	}
+	return bc.WriteMsg(buf)
+}
+
+// ReleaseMsg releases a buffer
+func (c *MultiConn) ReleaseMsg(m []byte) {
+	for _, c := range c.getConns() {
+		c.ReleaseMsg(m)
 	}
-	return nil
 }

net/conn/multiconn_test.go

@@ -178,7 +178,7 @@ func TestMulticonnSend(t *testing.T) {
 
 		for _, m := range msgs.msgs {
 			log.Info("send: %s", m.payload)
-			c.Out() <- []byte(m.payload)
+			c.WriteMsg([]byte(m.payload))
 			msgs.Sent(t, m.payload)
 			<-time.After(time.Microsecond * 10)
 		}
@@ -189,16 +189,20 @@ func TestMulticonnSend(t *testing.T) {
 
 		for {
 			select {
-			case payload := <-c.In():
-				msgs.Received(t, string(payload))
-				log.Info("recv: %s", payload)
-				if msgs.recv == len(msgs.msgs) {
-					return
-				}
-
+			default:
 			case <-ctx.Done():
 				return
+			}
+
+			payload, err := c.ReadMsg()
+			if err != nil {
+				panic(err)
+			}
 
+			msgs.Received(t, string(payload))
+			log.Info("recv: %s", payload)
+			if msgs.recv == len(msgs.msgs) {
+				return
 			}
 		}
 
@@ -252,11 +256,11 @@ func TestMulticonnSendUnderlying(t *testing.T) {
 			log.Info("send: %s", m.payload)
 			switch i % 3 {
 			case 0:
-				conns[0].Out() <- []byte(m.payload)
+				conns[0].WriteMsg([]byte(m.payload))
 			case 1:
-				conns[1].Out() <- []byte(m.payload)
+				conns[1].WriteMsg([]byte(m.payload))
 			case 2:
-				c.Out() <- []byte(m.payload)
+				c.WriteMsg([]byte(m.payload))
 			}
 			msgs.Sent(t, m.payload)
 			<-time.After(time.Microsecond * 10)
@@ -269,16 +273,20 @@ func TestMulticonnSendUnderlying(t *testing.T) {
 
 		for {
 			select {
-			case payload := <-c.In():
-				msgs.Received(t, string(payload))
-				log.Info("recv: %s", payload)
-				if msgs.recv == len(msgs.msgs) {
-					return
-				}
-
+			default:
 			case <-ctx.Done():
 				return
+			}
+
+			payload, err := c.ReadMsg()
+			if err != nil {
+				panic(err)
+			}
 
+			msgs.Received(t, string(payload))
+			log.Info("recv: %s", payload)
+			if msgs.recv == len(msgs.msgs) {
+				return
 			}
 		}
 

net/conn/secure_conn.go

 package conn
 
 import (
-	"errors"
-
 	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
+	msgio "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-msgio"
 	ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
 
-	spipe "github.com/jbenet/go-ipfs/crypto/spipe"
+	secio "github.com/jbenet/go-ipfs/crypto/secio"
 	peer "github.com/jbenet/go-ipfs/peer"
 	ctxc "github.com/jbenet/go-ipfs/util/ctxcloser"
-	"github.com/jbenet/go-ipfs/util/pipes"
 )
 
 // secureConn wraps another Conn object with an encrypted channel.
@@ -18,8 +16,11 @@ type secureConn struct {
 	// the wrapped conn
 	insecure Conn
 
-	// secure pipe, wrapping insecure
-	secure *spipe.SecurePipe
+	// secure io (wrapping insecure)
+	secure msgio.ReadWriteCloser
+
+	// secure Session
+	session secio.Session
 
 	ctxc.ContextCloser
 }
@@ -27,74 +28,30 @@ type secureConn struct {
 // newConn constructs a new connection
 func newSecureConn(ctx context.Context, insecure Conn, peers peer.Peerstore) (Conn, error) {
 
+	// NewSession performs the secure handshake, which takes multiple RTT
+	sessgen := secio.SessionGenerator{Local: insecure.LocalPeer(), Peerstore: peers}
+	session, err := sessgen.NewSession(ctx, insecure)
+	if err != nil {
+		return nil, err
+	}
+
 	conn := &secureConn{
 		insecure: insecure,
+		session:  session,
+		secure:   session.ReadWriter(),
 	}
 	conn.ContextCloser = ctxc.NewContextCloser(ctx, conn.close)
-
-	log.Debugf("newSecureConn: %v to %v", insecure.LocalPeer(), insecure.RemotePeer())
-	// perform secure handshake before returning this connection.
-	if err := conn.secureHandshake(peers); err != nil {
-		conn.Close()
-		return nil, err
-	}
-	log.Debugf("newSecureConn: %v to %v handshake success!", insecure.LocalPeer(), insecure.RemotePeer())
-
+	log.Debugf("newSecureConn: %v to %v handshake success!", conn.LocalPeer(), conn.RemotePeer())
 	return conn, nil
 }
 
-// secureHandshake performs the spipe secure handshake.
-func (c *secureConn) secureHandshake(peers peer.Peerstore) error {
-	if c.secure != nil {
-		return errors.New("Conn is already secured or being secured.")
-	}
-
-	// ok to panic here if this type assertion fails. Interface hack.
-	// when we support wrapping other Conns, we'll need to change
-	// spipe to do something else.
-	insecureSC := c.insecure.(*singleConn)
-
-	// setup a Duplex pipe for spipe
-	insecureD := pipes.Duplex{
-		In:  insecureSC.msgio.incoming.MsgChan,
-		Out: insecureSC.msgio.outgoing.MsgChan,
-	}
-
-	// spipe performs the secure handshake, which takes multiple RTT
-	sp, err := spipe.NewSecurePipe(c.Context(), 10, c.LocalPeer(), peers, insecureD)
-	if err != nil {
-		return err
-	}
-
-	// assign it into the conn object
-	c.secure = sp
-
-	// if we do not know RemotePeer, get it from secure chan (who identifies it)
-	if insecureSC.remote == nil {
-		insecureSC.remote = c.secure.RemotePeer()
-
-	} else if insecureSC.remote != c.secure.RemotePeer() {
-		// this panic is here because this would be an insidious programmer error
-		// that we need to ensure we catch.
-		// update: this actually might happen under normal operation-- should
-		// perhaps return an error. TBD.
-
-		log.Errorf("secureConn peer mismatch. %v != %v", insecureSC.remote, c.secure.RemotePeer())
-		log.Errorf("insecureSC.remote: %s %#v", insecureSC.remote, insecureSC.remote)
-		log.Errorf("c.secure.LocalPeer: %s %#v", c.secure.RemotePeer(), c.secure.RemotePeer())
-		panic("secureConn peer mismatch. consructed incorrectly?")
-	}
-
-	return nil
-}
-
 // close is called by ContextCloser
 func (c *secureConn) close() error {
-	err := c.insecure.Close()
-	if c.secure != nil { // may never have gotten here.
-		err = c.secure.Close()
+	if err := c.secure.Close(); err != nil {
+		c.insecure.Close()
+		return err
 	}
-	return err
+	return c.insecure.Close()
 }
 
 // ID is an identifier unique to this connection.
@@ -118,24 +75,35 @@ func (c *secureConn) RemoteMultiaddr() ma.Multiaddr {
 
 // LocalPeer is the Peer on this side
 func (c *secureConn) LocalPeer() peer.Peer {
-	return c.insecure.LocalPeer()
+	return c.session.LocalPeer()
 }
 
 // RemotePeer is the Peer on the remote side
 func (c *secureConn) RemotePeer() peer.Peer {
-	return c.insecure.RemotePeer()
+	return c.session.RemotePeer()
+}
+
+// Read reads data, net.Conn style
+func (c *secureConn) Read(buf []byte) (int, error) {
+	return c.secure.Read(buf)
+}
+
+// Write writes data, net.Conn style
+func (c *secureConn) Write(buf []byte) (int, error) {
+	return c.secure.Write(buf)
 }
 
-// In returns a readable message channel
-func (c *secureConn) In() <-chan []byte {
-	return c.secure.In
+// ReadMsg reads data, net.Conn style
+func (c *secureConn) ReadMsg() ([]byte, error) {
+	return c.secure.ReadMsg()
 }
 
-// Out returns a writable message channel
-func (c *secureConn) Out() chan<- []byte {
-	return c.secure.Out
+// WriteMsg writes data, net.Conn style
+func (c *secureConn) WriteMsg(buf []byte) error {
+	return c.secure.WriteMsg(buf)
 }
 
-func (c *secureConn) GetError() error {
-	return c.insecure.GetError()
+// ReleaseMsg releases a buffer
+func (c *secureConn) ReleaseMsg(m []byte) {
+	c.secure.ReleaseMsg(m)
 }

net/conn/secure_conn_test.go

@@ -105,6 +105,8 @@ func TestSecureCancel(t *testing.T) {
 }
 
 func TestSecureCloseLeak(t *testing.T) {
+	// t.Skip("Skipping in favor of another test")
+
 	if testing.Short() {
 		t.SkipNow()
 	}
@@ -125,15 +127,21 @@ func TestSecureCloseLeak(t *testing.T) {
 
 		for i := 0; i < num; i++ {
 			b1 := []byte("beep")
-			c1.Out() <- b1
-			b2 := <-c2.In()
+			c1.WriteMsg(b1)
+			b2, err := c2.ReadMsg()
+			if err != nil {
+				panic(err)
+			}
 			if !bytes.Equal(b1, b2) {
 				panic("bytes not equal")
 			}
 
-			b2 = []byte("boop")
-			c2.Out() <- b2
-			b1 = <-c1.In()
+			b2 = []byte("beep")
+			c2.WriteMsg(b2)
+			b1, err = c1.ReadMsg()
+			if err != nil {
+				panic(err)
+			}
 			if !bytes.Equal(b1, b2) {
 				panic("bytes not equal")
 			}

net/swarm/conn.go

@@ -200,7 +200,10 @@ func (s *Swarm) fanOut() {
 			log.Debugf("%s sent message to %s (%d)", s.local, msg.Peer(), i)
 			log.Event(context.TODO(), "sendMessage", s.local, msg)
 			// queue it in the connection's buffer
-			c.Out() <- msg.Data()
+			if err := c.WriteMsg(msg.Data()); err != nil {
+				log.Infof("%s connection failed to write: %s", c, err)
+				continue
+			}
 		}
 	}
 }
@@ -219,6 +222,9 @@ func (s *Swarm) fanInSingle(c conn.Conn) {
 		c.Children().Done() // child of Conn as well.
 	}()
 
+	// use readChan to be able to listen to Closing events
+	rchan := readChan(s.Context(), c)
+
 	i := 0
 	for {
 		select {
@@ -228,7 +234,7 @@ func (s *Swarm) fanInSingle(c conn.Conn) {
 		case <-c.Closing(): // Conn closing
 			return
 
-		case data, ok := <-c.In():
+		case data, ok := <-rchan:
 			if !ok {
 				log.Infof("%s in channel closed", c)
 				return // channel closed.
@@ -240,26 +246,30 @@ func (s *Swarm) fanInSingle(c conn.Conn) {
 	}
 }
 
-// Commenting out because it's platform specific
-// func setSocketReuse(l manet.Listener) error {
-// 	nl := l.NetListener()
-//
-// 	// for now only TCP. TODO change this when more networks.
-// 	file, err := nl.(*net.TCPListener).File()
-// 	if err != nil {
-// 		return err
-// 	}
-//
-// 	fd := file.Fd()
-// 	err = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_REUSEADDR, 1)
-// 	if err != nil {
-// 		return err
-// 	}
-//
-// 	err = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_REUSEPORT, 1)
-// 	if err != nil {
-// 		return err
-// 	}
-//
-// 	return nil
-// }
+// readChan is a temporary fixture to match the old interface. will be removed soon.
+func readChan(ctx context.Context, c conn.Conn) <-chan []byte {
+
+	ch := make(chan []byte) // no buffer. sync.
+
+	go func() {
+		defer close(ch)
+
+		for {
+			msg, err := c.ReadMsg()
+			if err != nil {
+				log.Infof("%s connection failed: %s", c, err)
+				return
+			}
+
+			select {
+			case <-c.Closing():
+				return
+			case <-ctx.Done():
+				return
+			case ch <- msg:
+			}
+		}
+	}()
+
+	return ch
+}