net -> p2p/net

The net package is the next to move. It will be massaged
a bit still to fix the Network / "NetworkBackend" conflict.

net/README.md

+# Network
+
+The IPFS Network package handles all of the peer-to-peer networking. It connects to other hosts, it encrypts communications, it muxes messages between the network's client services and target hosts. It has multiple subcomponents:
+
+- `Conn` - a connection to a single Peer
+  - `MultiConn` - a set of connections to a single Peer
+  - `SecureConn` - an encrypted (tls-like) connection
+- `Swarm` - holds connections to Peers, multiplexes from/to each `MultiConn`
+- `Muxer` - multiplexes between `Services` and `Swarm`. Handles `Requet/Reply`.
+  - `Service` - connects between an outside client service and Network.
+  - `Handler` - the client service part that handles requests
+
+It looks a bit like this:
+
+<center>
+![](https://docs.google.com/drawings/d/1FvU7GImRsb9GvAWDDo1le85jIrnFJNVB_OTPXC15WwM/pub?h=480)
+</center>

net/backpressure/backpressure.go

+package backpressure_tests

net/backpressure/backpressure_test.go

+package backpressure_tests
+
+import (
+	crand "crypto/rand"
+	"io"
+	"math/rand"
+	"testing"
+	"time"
+
+	inet "github.com/jbenet/go-ipfs/p2p/net"
+	netutil "github.com/jbenet/go-ipfs/p2p/net/swarmnet/util"
+	peer "github.com/jbenet/go-ipfs/p2p/peer"
+	eventlog "github.com/jbenet/go-ipfs/util/eventlog"
+
+	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
+)
+
+var log = eventlog.Logger("backpressure")
+
+// TestBackpressureStreamHandler tests whether mux handler
+// ratelimiting works. Meaning, since the handler is sequential
+// it should block senders.
+//
+// Important note: spdystream (which peerstream uses) has a set
+// of n workers (n=spdsystream.FRAME_WORKERS) which handle new
+// frames, including those starting new streams. So all of them
+// can be in the handler at one time. Also, the sending side
+// does not rate limit unless we call stream.Wait()
+//
+//
+// Note: right now, this happens muxer-wide. the muxer should
+// learn to flow control, so handlers cant block each other.
+func TestBackpressureStreamHandler(t *testing.T) {
+	t.Skip(`Sadly, as cool as this test is, it doesn't work
+Because spdystream doesnt handle stream open backpressure
+well IMO. I'll see about rewriting that part when it becomes
+a problem.
+`)
+
+	// a number of concurrent request handlers
+	limit := 10
+
+	// our way to signal that we're done with 1 request
+	requestHandled := make(chan struct{})
+
+	// handler rate limiting
+	receiverRatelimit := make(chan struct{}, limit)
+	for i := 0; i < limit; i++ {
+		receiverRatelimit <- struct{}{}
+	}
+
+	// sender counter of successfully opened streams
+	senderOpened := make(chan struct{}, limit*100)
+
+	// sender signals it's done (errored out)
+	senderDone := make(chan struct{})
+
+	// the receiver handles requests with some rate limiting
+	receiver := func(s inet.Stream) {
+		log.Debug("receiver received a stream")
+
+		<-receiverRatelimit // acquire
+		go func() {
+			// our request handler. can do stuff here. we
+			// simulate something taking time by waiting
+			// on requestHandled
+			log.Error("request worker handling...")
+			<-requestHandled
+			log.Error("request worker done!")
+			receiverRatelimit <- struct{}{} // release
+		}()
+	}
+
+	// the sender opens streams as fast as possible
+	sender := func(net inet.Network, remote peer.ID) {
+		var s inet.Stream
+		var err error
+		defer func() {
+			t.Error(err)
+			log.Debug("sender error. exiting.")
+			senderDone <- struct{}{}
+		}()
+
+		for {
+			s, err = net.NewStream(inet.ProtocolTesting, remote)
+			if err != nil {
+				return
+			}
+
+			_ = s
+			// if err = s.SwarmStream().Stream().Wait(); err != nil {
+			// 	return
+			// }
+
+			// "count" another successfully opened stream
+			// (large buffer so shouldn't block in normal operation)
+			log.Debug("sender opened another stream!")
+			senderOpened <- struct{}{}
+		}
+	}
+
+	// count our senderOpened events
+	countStreamsOpenedBySender := func(min int) int {
+		opened := 0
+		for opened < min {
+			log.Debugf("countStreamsOpenedBySender got %d (min %d)", opened, min)
+			select {
+			case <-senderOpened:
+				opened++
+			case <-time.After(10 * time.Millisecond):
+			}
+		}
+		return opened
+	}
+
+	// count our received events
+	// waitForNReceivedStreams := func(n int) {
+	// 	for n > 0 {
+	// 		log.Debugf("waiting for %d received streams...", n)
+	// 		select {
+	// 		case <-receiverRatelimit:
+	// 			n--
+	// 		}
+	// 	}
+	// }
+
+	testStreamsOpened := func(expected int) {
+		log.Debugf("testing rate limited to %d streams", expected)
+		if n := countStreamsOpenedBySender(expected); n != expected {
+			t.Fatalf("rate limiting did not work :( -- %d != %d", expected, n)
+		}
+	}
+
+	// ok that's enough setup. let's do it!
+
+	ctx := context.Background()
+	n1 := netutil.GenNetwork(t, ctx)
+	n2 := netutil.GenNetwork(t, ctx)
+
+	// setup receiver handler
+	n1.SetHandler(inet.ProtocolTesting, receiver)
+
+	log.Debugf("dialing %s", n2.ListenAddresses())
+	if err := n1.DialPeer(ctx, n2.LocalPeer()); err != nil {
+		t.Fatalf("Failed to dial:", err)
+	}
+
+	// launch sender!
+	go sender(n2, n1.LocalPeer())
+
+	// ok, what do we expect to happen? the receiver should
+	// receive 10 requests and stop receiving, blocking the sender.
+	// we can test this by counting 10x senderOpened requests
+
+	<-senderOpened // wait for the sender to successfully open some.
+	testStreamsOpened(limit - 1)
+
+	// let's "handle" 3 requests.
+	<-requestHandled
+	<-requestHandled
+	<-requestHandled
+	// the sender should've now been able to open exactly 3 more.
+
+	testStreamsOpened(3)
+
+	// shouldn't have opened anything more
+	testStreamsOpened(0)
+
+	// let's "handle" 100 requests in batches of 5
+	for i := 0; i < 20; i++ {
+		<-requestHandled
+		<-requestHandled
+		<-requestHandled
+		<-requestHandled
+		<-requestHandled
+		testStreamsOpened(5)
+	}
+
+	// success!
+
+	// now for the sugar on top: let's tear down the receiver. it should
+	// exit the sender.
+	n1.Close()
+
+	// shouldn't have opened anything more
+	testStreamsOpened(0)
+
+	select {
+	case <-time.After(100 * time.Millisecond):
+		t.Error("receiver shutdown failed to exit sender")
+	case <-senderDone:
+		log.Info("handler backpressure works!")
+	}
+}
+
+// TestStBackpressureStreamWrite tests whether streams see proper
+// backpressure when writing data over the network streams.
+func TestStBackpressureStreamWrite(t *testing.T) {
+
+	// senderWrote signals that the sender wrote bytes to remote.
+	// the value is the count of bytes written.
+	senderWrote := make(chan int, 10000)
+
+	// sender signals it's done (errored out)
+	senderDone := make(chan struct{})
+
+	// writeStats lets us listen to all the writes and return
+	// how many happened and how much was written
+	writeStats := func() (int, int) {
+		writes := 0
+		bytes := 0
+		for {
+			select {
+			case n := <-senderWrote:
+				writes++
+				bytes = bytes + n
+			default:
+				log.Debugf("stats: sender wrote %d bytes, %d writes", bytes, writes)
+				return bytes, writes
+			}
+		}
+	}
+
+	// sender attempts to write as fast as possible, signaling on the
+	// completion of every write. This makes it possible to see how
+	// fast it's actually writing. We pair this with a receiver
+	// that waits for a signal to read.
+	sender := func(s inet.Stream) {
+		defer func() {
+			s.Close()
+			senderDone <- struct{}{}
+		}()
+
+		// ready a buffer of random data
+		buf := make([]byte, 65536)
+		crand.Read(buf)
+
+		for {
+			// send a randomly sized subchunk
+			from := rand.Intn(len(buf) / 2)
+			to := rand.Intn(len(buf) / 2)
+			sendbuf := buf[from : from+to]
+
+			n, err := s.Write(sendbuf)
+			if err != nil {
+				log.Debug("sender error. exiting:", err)
+				return
+			}
+
+			log.Debugf("sender wrote %d bytes", n)
+			senderWrote <- n
+		}
+	}
+
+	// receive a number of bytes from a stream.
+	// returns the number of bytes written.
+	receive := func(s inet.Stream, expect int) {
+		log.Debugf("receiver to read %d bytes", expect)
+		rbuf := make([]byte, expect)
+		n, err := io.ReadFull(s, rbuf)
+		if err != nil {
+			t.Error("read failed:", err)
+		}
+		if expect != n {
+			t.Error("read len differs: %d != %d", expect, n)
+		}
+	}
+
+	// ok let's do it!
+
+	// setup the networks
+	ctx := context.Background()
+	n1 := netutil.GenNetwork(t, ctx)
+	n2 := netutil.GenNetwork(t, ctx)
+
+	netutil.DivulgeAddresses(n1, n2)
+	netutil.DivulgeAddresses(n2, n1)
+
+	// setup sender handler on 1
+	n1.SetHandler(inet.ProtocolTesting, sender)
+
+	log.Debugf("dialing %s", n2.ListenAddresses())
+	if err := n1.DialPeer(ctx, n2.LocalPeer()); err != nil {
+		t.Fatalf("Failed to dial:", err)
+	}
+
+	// open a stream, from 2->1, this is our reader
+	s, err := n2.NewStream(inet.ProtocolTesting, n1.LocalPeer())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// let's make sure r/w works.
+	testSenderWrote := func(bytesE int) {
+		bytesA, writesA := writeStats()
+		if bytesA != bytesE {
+			t.Errorf("numbers failed: %d =?= %d bytes, via %d writes", bytesA, bytesE, writesA)
+		}
+	}
+
+	// 500ms rounds of lockstep write + drain
+	roundsStart := time.Now()
+	roundsTotal := 0
+	for roundsTotal < (2 << 20) {
+		// let the sender fill its buffers, it will stop sending.
+		<-time.After(300 * time.Millisecond)
+		b, _ := writeStats()
+		testSenderWrote(0)
+		testSenderWrote(0)
+
+		// drain it all, wait again
+		receive(s, b)
+		roundsTotal = roundsTotal + b
+	}
+	roundsTime := time.Now().Sub(roundsStart)
+
+	// now read continously, while we measure stats.
+	stop := make(chan struct{})
+	contStart := time.Now()
+
+	go func() {
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+				receive(s, 2<<15)
+			}
+		}
+	}()
+
+	contTotal := 0
+	for contTotal < (2 << 20) {
+		n := <-senderWrote
+		contTotal += n
+	}
+	stop <- struct{}{}
+	contTime := time.Now().Sub(contStart)
+
+	// now compare! continuous should've been faster AND larger
+	if roundsTime < contTime {
+		t.Error("continuous should have been faster")
+	}
+
+	if roundsTotal < contTotal {
+		t.Error("continuous should have been larger, too!")
+	}
+
+	// and a couple rounds more for good measure ;)
+	for i := 0; i < 3; i++ {
+		// let the sender fill its buffers, it will stop sending.
+		<-time.After(300 * time.Millisecond)
+		b, _ := writeStats()
+		testSenderWrote(0)
+		testSenderWrote(0)
+
+		// drain it all, wait again
+		receive(s, b)
+	}
+
+	// this doesn't work :(:
+	// // now for the sugar on top: let's tear down the receiver. it should
+	// // exit the sender.
+	// n1.Close()
+	// testSenderWrote(0)
+	// testSenderWrote(0)
+	// select {
+	// case <-time.After(2 * time.Second):
+	// 	t.Error("receiver shutdown failed to exit sender")
+	// case <-senderDone:
+	// 	log.Info("handler backpressure works!")
+	// }
+}

net/conn/conn.go

+package conn
+
+import (
+	"fmt"
+	"net"
+	"time"
+
+	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"
+	mpool "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-msgio/mpool"
+	ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
+	manet "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net"
+
+	ic "github.com/jbenet/go-ipfs/p2p/crypto"
+	peer "github.com/jbenet/go-ipfs/p2p/peer"
+	u "github.com/jbenet/go-ipfs/util"
+	eventlog "github.com/jbenet/go-ipfs/util/eventlog"
+)
+
+var log = eventlog.Logger("conn")
+
+const (
+	// MaxMessageSize is the size of the largest single message. (4MB)
+	MaxMessageSize = 1 << 22
+)
+
+// ReleaseBuffer puts the given byte array back into the buffer pool,
+// first verifying that it is the correct size
+func ReleaseBuffer(b []byte) {
+	log.Debugf("Releasing buffer! (cap,size = %d, %d)", cap(b), len(b))
+	mpool.ByteSlicePool.Put(uint32(cap(b)), b)
+}
+
+// singleConn represents a single connection to another Peer (IPFS Node).
+type singleConn struct {
+	local  peer.ID
+	remote peer.ID
+	maconn manet.Conn
+	msgrw  msgio.ReadWriteCloser
+}
+
+// newConn constructs a new connection
+func newSingleConn(ctx context.Context, local, remote peer.ID, maconn manet.Conn) (Conn, error) {
+
+	conn := &singleConn{
+		local:  local,
+		remote: remote,
+		maconn: maconn,
+		msgrw:  msgio.NewReadWriter(maconn),
+	}
+	log.Debugf("newSingleConn %p: %v to %v", conn, local, remote)
+
+	// version handshake
+	if err := Handshake1(ctx, conn); err != nil {
+		conn.Close()
+		return nil, fmt.Errorf("Handshake1 failed: %s", err)
+	}
+
+	log.Debugf("newSingleConn %p: %v to %v finished", conn, local, remote)
+	return conn, nil
+}
+
+// close is the internal close function, called by ContextCloser.Close
+func (c *singleConn) Close() error {
+	log.Debugf("%s closing Conn with %s", c.local, c.remote)
+	// close underlying connection
+	return c.msgrw.Close()
+}
+
+// ID is an identifier unique to this connection.
+func (c *singleConn) ID() string {
+	return ID(c)
+}
+
+func (c *singleConn) String() string {
+	return String(c, "singleConn")
+}
+
+func (c *singleConn) LocalAddr() net.Addr {
+	return c.maconn.LocalAddr()
+}
+
+func (c *singleConn) RemoteAddr() net.Addr {
+	return c.maconn.RemoteAddr()
+}
+
+func (c *singleConn) LocalPrivateKey() ic.PrivKey {
+	return nil
+}
+
+func (c *singleConn) RemotePublicKey() ic.PubKey {
+	return nil
+}
+
+func (c *singleConn) SetDeadline(t time.Time) error {
+	return c.maconn.SetDeadline(t)
+}
+func (c *singleConn) SetReadDeadline(t time.Time) error {
+	return c.maconn.SetReadDeadline(t)
+}
+
+func (c *singleConn) SetWriteDeadline(t time.Time) error {
+	return c.maconn.SetWriteDeadline(t)
+}
+
+// LocalMultiaddr is the Multiaddr on this side
+func (c *singleConn) LocalMultiaddr() ma.Multiaddr {
+	return c.maconn.LocalMultiaddr()
+}
+
+// RemoteMultiaddr is the Multiaddr on the remote side
+func (c *singleConn) RemoteMultiaddr() ma.Multiaddr {
+	return c.maconn.RemoteMultiaddr()
+}
+
+// LocalPeer is the Peer on this side
+func (c *singleConn) LocalPeer() peer.ID {
+	return c.local
+}
+
+// RemotePeer is the Peer on the remote side
+func (c *singleConn) RemotePeer() peer.ID {
+	return c.remote
+}
+
+// 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)
+}
+
+func (c *singleConn) NextMsgLen() (int, error) {
+	return c.msgrw.NextMsgLen()
+}
+
+// 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)
+}
+
+// ReleaseMsg releases a buffer
+func (c *singleConn) ReleaseMsg(m []byte) {
+	c.msgrw.ReleaseMsg(m)
+}
+
+// ID returns the ID of a given Conn.
+func ID(c Conn) string {
+	l := fmt.Sprintf("%s/%s", c.LocalMultiaddr(), c.LocalPeer().Pretty())
+	r := fmt.Sprintf("%s/%s", c.RemoteMultiaddr(), c.RemotePeer().Pretty())
+	lh := u.Hash([]byte(l))
+	rh := u.Hash([]byte(r))
+	ch := u.XOR(lh, rh)
+	return u.Key(ch).Pretty()
+}
+
+// String returns the user-friendly String representation of a conn
+func String(c Conn, typ string) string {
+	return fmt.Sprintf("%s (%s) <-- %s %p --> (%s) %s",
+		c.LocalPeer(), c.LocalMultiaddr(), typ, c, c.RemoteMultiaddr(), c.RemotePeer())
+}

net/conn/conn_test.go

+package conn
+
+import (
+	"bytes"
+	"fmt"
+	"os"
+	"runtime"
+	"sync"
+	"testing"
+	"time"
+
+	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
+)
+
+func testOneSendRecv(t *testing.T, c1, c2 Conn) {
+	log.Debugf("testOneSendRecv from %s to %s", c1.LocalPeer(), c2.LocalPeer())
+	m1 := []byte("hello")
+	if err := c1.WriteMsg(m1); err != nil {
+		t.Fatal(err)
+	}
+	m2, err := c2.ReadMsg()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(m1, m2) {
+		t.Fatal("failed to send: %s %s", m1, m2)
+	}
+}
+
+func testNotOneSendRecv(t *testing.T, c1, c2 Conn) {
+	m1 := []byte("hello")
+	if err := c1.WriteMsg(m1); err == nil {
+		t.Fatal("write should have failed", err)
+	}
+	_, err := c2.ReadMsg()
+	if err == nil {
+		t.Fatal("read should have failed", err)
+	}
+}
+
+func TestClose(t *testing.T) {
+	// t.Skip("Skipping in favor of another test")
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	c1, c2, _, _ := setupSingleConn(t, ctx)
+
+	testOneSendRecv(t, c1, c2)
+	testOneSendRecv(t, c2, c1)
+
+	c1.Close()
+	testNotOneSendRecv(t, c1, c2)
+
+	c2.Close()
+	testNotOneSendRecv(t, c2, c1)
+	testNotOneSendRecv(t, c1, c2)
+}
+
+func TestCloseLeak(t *testing.T) {
+	// t.Skip("Skipping in favor of another test")
+	if testing.Short() {
+		t.SkipNow()
+	}
+
+	if os.Getenv("TRAVIS") == "true" {
+		t.Skip("this doesn't work well on travis")
+	}
+
+	var wg sync.WaitGroup
+
+	runPair := func(num int) {
+		ctx, cancel := context.WithCancel(context.Background())
+		c1, c2, _, _ := setupSingleConn(t, ctx)
+
+		for i := 0; i < num; i++ {
+			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(fmt.Errorf("bytes not equal: %s != %s", b1, b2))
+			}
+
+			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(fmt.Errorf("bytes not equal: %s != %s", b1, b2))
+			}
+
+			<-time.After(time.Microsecond * 5)
+		}
+
+		c1.Close()
+		c2.Close()
+		cancel() // close the listener
+		wg.Done()
+	}
+
+	var cons = 5
+	var msgs = 50
+	log.Debugf("Running %d connections * %d msgs.\n", cons, msgs)
+	for i := 0; i < cons; i++ {
+		wg.Add(1)
+		go runPair(msgs)
+	}
+
+	log.Debugf("Waiting...\n")
+	wg.Wait()
+	// done!
+
+	<-time.After(time.Millisecond * 150)
+	if runtime.NumGoroutine() > 20 {
+		// panic("uncomment me to debug")
+		t.Fatal("leaking goroutines:", runtime.NumGoroutine())
+	}
+}

net/conn/dial.go

+package conn
+
+import (
+	"fmt"
+	"strings"
+
+	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
+	ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
+	manet "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net"
+
+	peer "github.com/jbenet/go-ipfs/p2p/peer"
+	debugerror "github.com/jbenet/go-ipfs/util/debugerror"
+)
+
+// String returns the string rep of d.
+func (d *Dialer) String() string {
+	return fmt.Sprintf("<Dialer %s %s ...>", d.LocalPeer, d.LocalAddrs[0])
+}
+
+// Dial connects to a peer over a particular address
+// Ensures raddr is part of peer.Addresses()
+// Example: d.DialAddr(ctx, peer.Addresses()[0], peer)
+func (d *Dialer) Dial(ctx context.Context, raddr ma.Multiaddr, remote peer.ID) (Conn, error) {
+
+	network, _, err := manet.DialArgs(raddr)
+	if err != nil {
+		return nil, err
+	}
+
+	if strings.HasPrefix(raddr.String(), "/ip4/0.0.0.0") {
+		return nil, debugerror.Errorf("Attempted to connect to zero address: %s", raddr)
+	}
+
+	var laddr ma.Multiaddr
+	if len(d.LocalAddrs) > 0 {
+		// laddr := MultiaddrNetMatch(raddr, d.LocalAddrs)
+		laddr = NetAddress(network, d.LocalAddrs)
+		if laddr == nil {
+			return nil, debugerror.Errorf("No local address for network %s", network)
+		}
+	}
+
+	// TODO: try to get reusing addr/ports to work.
+	// madialer := manet.Dialer{LocalAddr: laddr}
+	madialer := manet.Dialer{}
+
+	log.Debugf("%s dialing %s %s", d.LocalPeer, remote, raddr)
+	maconn, err := madialer.Dial(raddr)
+	if err != nil {
+		return nil, err
+	}
+
+	var connOut Conn
+	var errOut error
+	done := make(chan struct{})
+
+	// do it async to ensure we respect don contexteone
+	go func() {
+		defer func() { done <- struct{}{} }()
+
+		c, err := newSingleConn(ctx, d.LocalPeer, remote, maconn)
+		if err != nil {
+			errOut = err
+			return
+		}
+
+		if d.PrivateKey == nil {
+			log.Warning("dialer %s dialing INSECURELY %s at %s!", d, remote, raddr)
+			connOut = c
+			return
+		}
+		c2, err := newSecureConn(ctx, d.PrivateKey, c)
+		if err != nil {
+			errOut = err
+			c.Close()
+			return
+		}
+
+		connOut = c2
+	}()
+
+	select {
+	case <-ctx.Done():
+		maconn.Close()
+		return nil, ctx.Err()
+	case <-done:
+		// whew, finished.
+	}
+
+	return connOut, errOut
+}
+
+// MultiaddrProtocolsMatch returns whether two multiaddrs match in protocol stacks.
+func MultiaddrProtocolsMatch(a, b ma.Multiaddr) bool {
+	ap := a.Protocols()
+	bp := b.Protocols()
+
+	if len(ap) != len(bp) {
+		return false
+	}
+
+	for i, api := range ap {
+		if api != bp[i] {
+			return false
+		}
+	}
+
+	return true
+}
+
+// MultiaddrNetMatch returns the first Multiaddr found to match  network.
+func MultiaddrNetMatch(tgt ma.Multiaddr, srcs []ma.Multiaddr) ma.Multiaddr {
+	for _, a := range srcs {
+		if MultiaddrProtocolsMatch(tgt, a) {
+			return a
+		}
+	}
+	return nil
+}
+
+// NetAddress returns the first Multiaddr found for a given network.
+func NetAddress(n string, addrs []ma.Multiaddr) ma.Multiaddr {
+	for _, a := range addrs {
+		for _, p := range a.Protocols() {
+			if p.Name == n {
+				return a
+			}
+		}
+	}
+	return nil
+}

net/conn/dial_test.go

+package conn
+
+import (
+	"io"
+	"net"
+	"testing"
+	"time"
+
+	tu "github.com/jbenet/go-ipfs/util/testutil"
+
+	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
+)
+
+func echoListen(ctx context.Context, listener Listener) {
+	for {
+		c, err := listener.Accept()
+		if err != nil {
+
+			select {
+			case <-ctx.Done():
+				return
+			default:
+			}
+
+			if ne, ok := err.(net.Error); ok && ne.Temporary() {
+				<-time.After(time.Microsecond * 10)
+				continue
+			}
+
+			log.Debugf("echoListen: listener appears to be closing")
+			return
+		}
+
+		go echo(c.(Conn))
+	}
+}
+
+func echo(c Conn) {
+	io.Copy(c, c)
+}
+
+func setupSecureConn(t *testing.T, ctx context.Context) (a, b Conn, p1, p2 tu.PeerNetParams) {
+	return setupConn(t, ctx, true)
+}
+
+func setupSingleConn(t *testing.T, ctx context.Context) (a, b Conn, p1, p2 tu.PeerNetParams) {
+	return setupConn(t, ctx, false)
+}
+
+func setupConn(t *testing.T, ctx context.Context, secure bool) (a, b Conn, p1, p2 tu.PeerNetParams) {
+
+	p1 = tu.RandPeerNetParamsOrFatal(t)
+	p2 = tu.RandPeerNetParamsOrFatal(t)
+	laddr := p1.Addr
+
+	key1 := p1.PrivKey
+	key2 := p2.PrivKey
+	if !secure {
+		key1 = nil
+		key2 = nil
+	}
+	l1, err := Listen(ctx, laddr, p1.ID, key1)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	d2 := &Dialer{
+		LocalPeer:  p2.ID,
+		PrivateKey: key2,
+	}
+
+	var c2 Conn
+
+	done := make(chan error)
+	go func() {
+		var err error
+		c2, err = d2.Dial(ctx, p1.Addr, p1.ID)
+		if err != nil {
+			done <- err
+		}
+		close(done)
+	}()
+
+	c1, err := l1.Accept()
+	if err != nil {
+		t.Fatal("failed to accept", err)
+	}
+	if err := <-done; err != nil {
+		t.Fatal(err)
+	}
+
+	return c1.(Conn), c2, p1, p2
+}
+
+func testDialer(t *testing.T, secure bool) {
+	// t.Skip("Skipping in favor of another test")
+
+	p1 := tu.RandPeerNetParamsOrFatal(t)
+	p2 := tu.RandPeerNetParamsOrFatal(t)
+
+	key1 := p1.PrivKey
+	key2 := p2.PrivKey
+	if !secure {
+		key1 = nil
+		key2 = nil
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	l1, err := Listen(ctx, p1.Addr, p1.ID, key1)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	d2 := &Dialer{
+		LocalPeer:  p2.ID,
+		PrivateKey: key2,
+	}
+
+	go echoListen(ctx, l1)
+
+	c, err := d2.Dial(ctx, p1.Addr, p1.ID)
+	if err != nil {
+		t.Fatal("error dialing peer", err)
+	}
+
+	// fmt.Println("sending")
+	c.WriteMsg([]byte("beep"))
+	c.WriteMsg([]byte("boop"))
+
+	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, err = c.ReadMsg()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	data = string(out)
+	if string(out) != "boop" {
+		t.Error("unexpected conn output", data)
+	}
+
+	// fmt.Println("closing")
+	c.Close()
+	l1.Close()
+	cancel()
+}
+
+func TestDialerInsecure(t *testing.T) {
+	// t.Skip("Skipping in favor of another test")
+	testDialer(t, false)
+}
+
+func TestDialerSecure(t *testing.T) {
+	// t.Skip("Skipping in favor of another test")
+	testDialer(t, true)
+}

net/conn/handshake.go

+package conn
+
+import (
+	"fmt"
+
+	handshake "github.com/jbenet/go-ipfs/p2p/net/handshake"
+	hspb "github.com/jbenet/go-ipfs/p2p/net/handshake/pb"
+
+	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
+	ggprotoio "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/gogoprotobuf/io"
+)
+
+// Handshake1 exchanges local and remote versions and compares them
+// closes remote and returns an error in case of major difference
+func Handshake1(ctx context.Context, c Conn) error {
+	rpeer := c.RemotePeer()
+	lpeer := c.LocalPeer()
+
+	// setup up protobuf io
+	maxSize := 4096
+	r := ggprotoio.NewDelimitedReader(c, maxSize)
+	w := ggprotoio.NewDelimitedWriter(c)
+	localH := handshake.Handshake1Msg()
+	remoteH := new(hspb.Handshake1)
+
+	// send the outgoing handshake message
+	if err := w.WriteMsg(localH); err != nil {
+		return err
+	}
+	log.Debugf("%p sent my version (%s) to %s", c, localH, rpeer)
+	log.Event(ctx, "handshake1Sent", lpeer)
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+
+	if err := r.ReadMsg(remoteH); err != nil {
+		return fmt.Errorf("could not receive remote version: %q", err)
+	}
+	log.Debugf("%p received remote version (%s) from %s", c, remoteH, rpeer)
+	log.Event(ctx, "handshake1Received", lpeer)
+
+	if err := handshake.Handshake1Compatible(localH, remoteH); err != nil {
+		log.Infof("%s (%s) incompatible version with %s (%s)", lpeer, localH, rpeer, remoteH)
+		return err
+	}
+
+	log.Debugf("%s version handshake compatible %s", lpeer, rpeer)
+	return nil
+}

net/conn/interface.go

+package conn
+
+import (
+	"io"
+	"net"
+	"time"
+
+	ic "github.com/jbenet/go-ipfs/p2p/crypto"
+	peer "github.com/jbenet/go-ipfs/p2p/peer"
+	u "github.com/jbenet/go-ipfs/util"
+
+	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"
+)
+
+// Map maps Keys (Peer.IDs) to Connections.
+type Map map[u.Key]Conn
+
+type PeerConn interface {
+	// LocalPeer (this side) ID, PrivateKey, and Address
+	LocalPeer() peer.ID
+	LocalPrivateKey() ic.PrivKey
+	LocalMultiaddr() ma.Multiaddr
+
+	// RemotePeer ID, PublicKey, and Address
+	RemotePeer() peer.ID
+	RemotePublicKey() ic.PubKey
+	RemoteMultiaddr() ma.Multiaddr
+}
+
+// Conn is a generic message-based Peer-to-Peer connection.
+type Conn interface {
+	PeerConn
+
+	// ID is an identifier unique to this connection.
+	ID() string
+
+	// can't just say "net.Conn" cause we have duplicate methods.
+	LocalAddr() net.Addr
+	RemoteAddr() net.Addr
+	SetDeadline(t time.Time) error
+	SetReadDeadline(t time.Time) error
+	SetWriteDeadline(t time.Time) error
+
+	msgio.Reader
+	msgio.Writer
+	io.Closer
+}
+
+// Dialer is an object that can open connections. We could have a "convenience"
+// Dial function as before, but it would have many arguments, as dialing is
+// no longer simple (need a peerstore, a local peer, a context, a network, etc)
+type Dialer struct {
+
+	// LocalPeer is the identity of the local Peer.
+	LocalPeer peer.ID
+
+	// LocalAddrs is a set of local addresses to use.
+	LocalAddrs []ma.Multiaddr
+
+	// PrivateKey used to initialize a secure connection.
+	// Warning: if PrivateKey is nil, connection will not be secured.
+	PrivateKey ic.PrivKey
+}
+
+// Listener is an object that can accept connections. It matches net.Listener
+type Listener interface {
+
+	// Accept waits for and returns the next connection to the listener.
+	Accept() (net.Conn, error)
+
+	// Addr is the local address
+	Addr() net.Addr
+
+	// Multiaddr is the local multiaddr address
+	Multiaddr() ma.Multiaddr
+
+	// LocalPeer is the identity of the local Peer.
+	LocalPeer() peer.ID
+
+	// Close closes the listener.
+	// Any blocked Accept operations will be unblocked and return errors.
+	Close() error
+}

net/conn/listen.go

+package conn
+
+import (
+	"fmt"
+	"net"
+
+	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
+	ctxgroup "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-ctxgroup"
+	ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
+	manet "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net"
+
+	ic "github.com/jbenet/go-ipfs/p2p/crypto"
+	peer "github.com/jbenet/go-ipfs/p2p/peer"
+)
+
+// listener is an object that can accept connections. It implements Listener
+type listener struct {
+	manet.Listener
+
+	maddr ma.Multiaddr // Local multiaddr to listen on
+	local peer.ID      // LocalPeer is the identity of the local Peer
+	privk ic.PrivKey   // private key to use to initialize secure conns
+
+	cg ctxgroup.ContextGroup
+}
+
+func (l *listener) teardown() error {
+	defer log.Debugf("listener closed: %s %s", l.local, l.maddr)
+	return l.Listener.Close()
+}
+
+func (l *listener) Close() error {
+	log.Debugf("listener closing: %s %s", l.local, l.maddr)
+	return l.cg.Close()
+}
+
+func (l *listener) String() string {
+	return fmt.Sprintf("<Listener %s %s>", l.local, l.maddr)
+}
+
+// Accept waits for and returns the next connection to the listener.
+// Note that unfortunately this
+func (l *listener) Accept() (net.Conn, error) {
+
+	// listeners dont have contexts. given changes dont make sense here anymore
+	// note that the parent of listener will Close, which will interrupt all io.
+	// Contexts and io don't mix.
+	ctx := context.Background()
+
+	maconn, err := l.Listener.Accept()
+	if err != nil {
+		return nil, err
+	}
+
+	c, err := newSingleConn(ctx, l.local, "", maconn)
+	if err != nil {
+		return nil, fmt.Errorf("Error accepting connection: %v", err)
+	}
+
+	if l.privk == nil {
+		log.Warning("listener %s listening INSECURELY!", l)
+		return c, nil
+	}
+	sc, err := newSecureConn(ctx, l.privk, c)
+	if err != nil {
+		return nil, fmt.Errorf("Error securing connection: %v", err)
+	}
+	return sc, nil
+}
+
+func (l *listener) Addr() net.Addr {
+	return l.Listener.Addr()
+}
+
+// Multiaddr is the identity of the local Peer.
+func (l *listener) Multiaddr() ma.Multiaddr {
+	return l.maddr
+}
+
+// LocalPeer is the identity of the local Peer.
+func (l *listener) LocalPeer() peer.ID {
+	return l.local
+}
+
+func (l *listener) Loggable() map[string]interface{} {
+	return map[string]interface{}{
+		"listener": map[string]interface{}{
+			"peer":    l.LocalPeer(),
+			"address": l.Multiaddr(),
+			"secure":  (l.privk != nil),
+		},
+	}
+}
+
+// Listen listens on the particular multiaddr, with given peer and peerstore.
+func Listen(ctx context.Context, addr ma.Multiaddr, local peer.ID, sk ic.PrivKey) (Listener, error) {
+
+	ml, err := manet.Listen(addr)
+	if err != nil {
+		return nil, fmt.Errorf("Failed to listen on %s: %s", addr, err)
+	}
+
+	l := &listener{
+		Listener: ml,
+		maddr:    addr,
+		local:    local,
+		privk:    sk,
+		cg:       ctxgroup.WithContext(ctx),
+	}
+	l.cg.SetTeardown(l.teardown)
+
+	log.Infof("swarm listening on %s", l.Multiaddr())
+	log.Event(ctx, "swarmListen", l)
+	return l, nil
+}

net/conn/secure_conn.go

+package conn
+
+import (
+	"net"
+	"time"
+
+	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"
+
+	ic "github.com/jbenet/go-ipfs/p2p/crypto"
+	secio "github.com/jbenet/go-ipfs/p2p/crypto/secio"
+	peer "github.com/jbenet/go-ipfs/p2p/peer"
+	errors "github.com/jbenet/go-ipfs/util/debugerror"
+)
+
+// secureConn wraps another Conn object with an encrypted channel.
+type secureConn struct {
+
+	// the wrapped conn
+	insecure Conn
+
+	// secure io (wrapping insecure)
+	secure msgio.ReadWriteCloser
+
+	// secure Session
+	session secio.Session
+}
+
+// newConn constructs a new connection
+func newSecureConn(ctx context.Context, sk ic.PrivKey, insecure Conn) (Conn, error) {
+
+	if insecure == nil {
+		return nil, errors.New("insecure is nil")
+	}
+	if insecure.LocalPeer() == "" {
+		return nil, errors.New("insecure.LocalPeer() is nil")
+	}
+	if sk == nil {
+		panic("way")
+		return nil, errors.New("private key is nil")
+	}
+
+	// NewSession performs the secure handshake, which takes multiple RTT
+	sessgen := secio.SessionGenerator{LocalID: insecure.LocalPeer(), PrivateKey: sk}
+	session, err := sessgen.NewSession(ctx, insecure)
+	if err != nil {
+		return nil, err
+	}
+
+	conn := &secureConn{
+		insecure: insecure,
+		session:  session,
+		secure:   session.ReadWriter(),
+	}
+	log.Debugf("newSecureConn: %v to %v handshake success!", conn.LocalPeer(), conn.RemotePeer())
+	return conn, nil
+}
+
+func (c *secureConn) Close() error {
+	if err := c.secure.Close(); err != nil {
+		c.insecure.Close()
+		return err
+	}
+	return c.insecure.Close()
+}
+
+// ID is an identifier unique to this connection.
+func (c *secureConn) ID() string {
+	return ID(c)
+}
+
+func (c *secureConn) String() string {
+	return String(c, "secureConn")
+}
+
+func (c *secureConn) LocalAddr() net.Addr {
+	return c.insecure.LocalAddr()
+}
+
+func (c *secureConn) RemoteAddr() net.Addr {
+	return c.insecure.RemoteAddr()
+}
+
+func (c *secureConn) SetDeadline(t time.Time) error {
+	return c.insecure.SetDeadline(t)
+}
+
+func (c *secureConn) SetReadDeadline(t time.Time) error {
+	return c.insecure.SetReadDeadline(t)
+}
+
+func (c *secureConn) SetWriteDeadline(t time.Time) error {
+	return c.insecure.SetWriteDeadline(t)
+}
+
+// LocalMultiaddr is the Multiaddr on this side
+func (c *secureConn) LocalMultiaddr() ma.Multiaddr {
+	return c.insecure.LocalMultiaddr()
+}
+
+// RemoteMultiaddr is the Multiaddr on the remote side
+func (c *secureConn) RemoteMultiaddr() ma.Multiaddr {
+	return c.insecure.RemoteMultiaddr()
+}
+
+// LocalPeer is the Peer on this side
+func (c *secureConn) LocalPeer() peer.ID {
+	return c.session.LocalPeer()
+}
+
+// RemotePeer is the Peer on the remote side
+func (c *secureConn) RemotePeer() peer.ID {
+	return c.session.RemotePeer()
+}
+
+// LocalPrivateKey is the public key of the peer on this side
+func (c *secureConn) LocalPrivateKey() ic.PrivKey {
+	return c.session.LocalPrivateKey()
+}
+
+// RemotePubKey is the public key of the peer on the remote side
+func (c *secureConn) RemotePublicKey() ic.PubKey {
+	return c.session.RemotePublicKey()
+}
+
+// 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)
+}
+
+func (c *secureConn) NextMsgLen() (int, error) {
+	return c.secure.NextMsgLen()
+}
+
+// ReadMsg reads data, net.Conn style
+func (c *secureConn) ReadMsg() ([]byte, error) {
+	return c.secure.ReadMsg()
+}
+
+// WriteMsg writes data, net.Conn style
+func (c *secureConn) WriteMsg(buf []byte) error {
+	return c.secure.WriteMsg(buf)
+}
+
+// ReleaseMsg releases a buffer
+func (c *secureConn) ReleaseMsg(m []byte) {
+	c.secure.ReleaseMsg(m)
+}

net/conn/secure_conn_test.go

+package conn
+
+import (
+	"bytes"
+	"os"
+	"runtime"
+	"sync"
+	"testing"
+	"time"
+
+	ic "github.com/jbenet/go-ipfs/p2p/crypto"
+
+	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
+)
+
+func upgradeToSecureConn(t *testing.T, ctx context.Context, sk ic.PrivKey, c Conn) (Conn, error) {
+	if c, ok := c.(*secureConn); ok {
+		return c, nil
+	}
+
+	// shouldn't happen, because dial + listen already return secure conns.
+	s, err := newSecureConn(ctx, sk, c)
+	if err != nil {
+		return nil, err
+	}
+	return s, nil
+}
+
+func secureHandshake(t *testing.T, ctx context.Context, sk ic.PrivKey, c Conn, done chan error) {
+	_, err := upgradeToSecureConn(t, ctx, sk, c)
+	done <- err
+}
+
+func TestSecureSimple(t *testing.T) {
+	// t.Skip("Skipping in favor of another test")
+
+	numMsgs := 100
+	if testing.Short() {
+		numMsgs = 10
+	}
+
+	ctx := context.Background()
+	c1, c2, p1, p2 := setupSingleConn(t, ctx)
+
+	done := make(chan error)
+	go secureHandshake(t, ctx, p1.PrivKey, c1, done)
+	go secureHandshake(t, ctx, p2.PrivKey, c2, done)
+
+	for i := 0; i < 2; i++ {
+		if err := <-done; err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	for i := 0; i < numMsgs; i++ {
+		testOneSendRecv(t, c1, c2)
+		testOneSendRecv(t, c2, c1)
+	}
+
+	c1.Close()
+	c2.Close()
+}
+
+func TestSecureClose(t *testing.T) {
+	// t.Skip("Skipping in favor of another test")
+
+	ctx := context.Background()
+	c1, c2, p1, p2 := setupSingleConn(t, ctx)
+
+	done := make(chan error)
+	go secureHandshake(t, ctx, p1.PrivKey, c1, done)
+	go secureHandshake(t, ctx, p2.PrivKey, c2, done)
+
+	for i := 0; i < 2; i++ {
+		if err := <-done; err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	testOneSendRecv(t, c1, c2)
+
+	c1.Close()
+	testNotOneSendRecv(t, c1, c2)
+
+	c2.Close()
+	testNotOneSendRecv(t, c1, c2)
+	testNotOneSendRecv(t, c2, c1)
+
+}
+
+func TestSecureCancelHandshake(t *testing.T) {
+	// t.Skip("Skipping in favor of another test")
+
+	ctx, cancel := context.WithCancel(context.Background())
+	c1, c2, p1, p2 := setupSingleConn(t, ctx)
+
+	done := make(chan error)
+	go secureHandshake(t, ctx, p1.PrivKey, c1, done)
+	<-time.After(time.Millisecond)
+	cancel() // cancel ctx
+	go secureHandshake(t, ctx, p2.PrivKey, c2, done)
+
+	for i := 0; i < 2; i++ {
+		if err := <-done; err == nil {
+			t.Error("cancel should've errored out")
+		}
+	}
+}
+
+func TestSecureHandshakeFailsWithWrongKeys(t *testing.T) {
+	// t.Skip("Skipping in favor of another test")
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	c1, c2, p1, p2 := setupSingleConn(t, ctx)
+
+	done := make(chan error)
+	go secureHandshake(t, ctx, p2.PrivKey, c1, done)
+	go secureHandshake(t, ctx, p1.PrivKey, c2, done)
+
+	for i := 0; i < 2; i++ {
+		if err := <-done; err == nil {
+			t.Fatal("wrong keys should've errored out.")
+		}
+	}
+}
+
+func TestSecureCloseLeak(t *testing.T) {
+	// t.Skip("Skipping in favor of another test")
+
+	if testing.Short() {
+		t.SkipNow()
+	}
+	if os.Getenv("TRAVIS") == "true" {
+		t.Skip("this doesn't work well on travis")
+	}
+
+	runPair := func(c1, c2 Conn, num int) {
+		log.Debugf("runPair %d", num)
+
+		for i := 0; i < num; i++ {
+			log.Debugf("runPair iteration %d", i)
+			b1 := []byte("beep")
+			c1.WriteMsg(b1)
+			b2, err := c2.ReadMsg()
+			if err != nil {
+				panic(err)
+			}
+			if !bytes.Equal(b1, b2) {
+				panic("bytes not equal")
+			}
+
+			b2 = []byte("beep")
+			c2.WriteMsg(b2)
+			b1, err = c1.ReadMsg()
+			if err != nil {
+				panic(err)
+			}
+			if !bytes.Equal(b1, b2) {
+				panic("bytes not equal")
+			}
+
+			<-time.After(time.Microsecond * 5)
+		}
+	}
+
+	var cons = 5
+	var msgs = 50
+	log.Debugf("Running %d connections * %d msgs.\n", cons, msgs)
+
+	var wg sync.WaitGroup
+	for i := 0; i < cons; i++ {
+		wg.Add(1)
+
+		ctx, cancel := context.WithCancel(context.Background())
+		c1, c2, _, _ := setupSecureConn(t, ctx)
+		go func(c1, c2 Conn) {
+
+			defer func() {
+				c1.Close()
+				c2.Close()
+				cancel()
+				wg.Done()
+			}()
+
+			runPair(c1, c2, msgs)
+		}(c1, c2)
+	}
+
+	log.Debugf("Waiting...\n")
+	wg.Wait()
+	// done!
+
+	<-time.After(time.Millisecond * 150)
+	if runtime.NumGoroutine() > 20 {
+		// panic("uncomment me to debug")
+		t.Fatal("leaking goroutines:", runtime.NumGoroutine())
+	}
+}

net/handshake/README.md

+# IFPS Handshake
+
+The IPFS Protocol Handshake is divided into three sequential steps
+
+1. Version Handshake (`Hanshake1`)
+2. Secure Channel (`NewSecureConn`)
+3. Services (`Handshake3`)
+
+Currently these parts currently happen sequentially (costing an awful 5 RTT),
+but can be optimized to 2 RTT.
+
+### Version Handshake
+
+The Version Handshake ensures that nodes speaking to each other can interoperate.
+They send each other protocol versions and ensure there is a match on the major
+version (semver).
+
+### Secure Channel
+
+The second part exchanges keys and establishes a secure comm channel. This
+follows ECDHE TLS, but *isn't* TLS. (why will be written up elsewhere).
+
+### Services
+
+The Services portion sends any additional information on nodes needed
+by the nodes, e.g. Listen Address (the received address could be a Dial addr),
+and later on can include Service listing (dht, exchange, ipns, etc).

net/handshake/doc.go

+/*
+package handshake implements the ipfs handshake protocol
+
+IPFS Handshake
+
+The IPFS Protocol Handshake is divided into three sequential steps
+
+1. Version Handshake (`Hanshake1`)
+
+2. Secure Channel (`NewSecureConn`)
+
+3. Services (`Handshake3`)
+
+Currently these parts currently happen sequentially (costing an awful 5 RTT),
+but can be optimized to 2 RTT.
+
+Version Handshake
+
+The Version Handshake ensures that nodes speaking to each other can interoperate.
+They send each other protocol versions and ensure there is a match on the major
+version (semver).
+
+Secure Channel
+
+The second part exchanges keys and establishes a secure comm channel. This
+follows ECDHE TLS, but *isn't* TLS. (why will be written up elsewhere).
+
+Services
+
+The Services portion sends any additional information on nodes needed
+by the nodes, e.g. Listen Address (the received address could be a Dial addr),
+and later on can include Service listing (dht, exchange, ipns, etc).
+*/
+package handshake

net/handshake/handshake1.go

+package handshake
+
+import (
+	"errors"
+	"fmt"
+
+	config "github.com/jbenet/go-ipfs/config"
+	pb "github.com/jbenet/go-ipfs/p2p/net/handshake/pb"
+	u "github.com/jbenet/go-ipfs/util"
+
+	semver "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/coreos/go-semver/semver"
+)
+
+var log = u.Logger("handshake")
+
+// IpfsVersion holds the current protocol version for a client running this code
+var IpfsVersion *semver.Version
+var ClientVersion = "go-ipfs/" + config.CurrentVersionNumber
+
+func init() {
+	var err error
+	IpfsVersion, err = semver.NewVersion("0.0.1")
+	if err != nil {
+		panic(fmt.Errorf("invalid protocol version: %v", err))
+	}
+}
+
+// Handshake1Msg returns the current protocol version as a protobuf message
+func Handshake1Msg() *pb.Handshake1 {
+	return NewHandshake1(IpfsVersion.String(), ClientVersion)
+}
+
+// ErrVersionMismatch is returned when two clients don't share a protocol version
+var ErrVersionMismatch = errors.New("protocol missmatch")
+
+// Handshake1Compatible checks whether two versions are compatible
+// returns nil if they are fine
+func Handshake1Compatible(handshakeA, handshakeB *pb.Handshake1) error {
+	a, err := semver.NewVersion(*handshakeA.ProtocolVersion)
+	if err != nil {
+		return err
+	}
+	b, err := semver.NewVersion(*handshakeB.ProtocolVersion)
+	if err != nil {
+		return err
+	}
+
+	if a.Major != b.Major {
+		return ErrVersionMismatch
+	}
+
+	return nil
+}
+
+// NewHandshake1 creates a new Handshake1 from the two strings
+func NewHandshake1(protoVer, agentVer string) *pb.Handshake1 {
+	if protoVer == "" {
+		protoVer = IpfsVersion.String()
+	}
+	if agentVer == "" {
+		agentVer = ClientVersion
+	}
+
+	return &pb.Handshake1{
+		ProtocolVersion: &protoVer,
+		AgentVersion:    &agentVer,
+	}
+}

net/handshake/handshake1_test.go

+package handshake
+
+import "testing"
+
+func TestH1Compatible(t *testing.T) {
+	tcases := []struct {
+		a, b     string
+		expected error
+	}{
+		{"0.0.0", "0.0.0", nil},
+		{"1.0.0", "1.1.0", nil},
+		{"1.0.0", "1.0.1", nil},
+		{"0.0.0", "1.0.0", ErrVersionMismatch},
+		{"1.0.0", "0.0.0", ErrVersionMismatch},
+	}
+
+	for i, tcase := range tcases {
+
+		if Handshake1Compatible(NewHandshake1(tcase.a, ""), NewHandshake1(tcase.b, "")) != tcase.expected {
+			t.Fatalf("case[%d] failed", i)
+		}
+	}
+}

net/handshake/pb/Makefile

+
+PB = $(wildcard *.proto)
+GO = $(PB:.proto=.pb.go)
+
+all: $(GO)
+
+%.pb.go: %.proto
+	protoc --gogo_out=. --proto_path=../../../../../../:/usr/local/opt/protobuf/include:. $<
+
+clean:
+	rm *.pb.go

net/handshake/pb/handshake.pb.go

+// Code generated by protoc-gen-gogo.
+// source: handshake.proto
+// DO NOT EDIT!
+
+/*
+Package handshake_pb is a generated protocol buffer package.
+
+It is generated from these files:
+	handshake.proto
+
+It has these top-level messages:
+	Handshake1
+	Handshake3
+*/
+package handshake_pb
+
+import proto "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/gogoprotobuf/proto"
+import json "encoding/json"
+import math "math"
+
+// Reference proto, json, and math imports to suppress error if they are not otherwise used.
+var _ = proto.Marshal
+var _ = &json.SyntaxError{}
+var _ = math.Inf
+
+// Handshake1 is delivered _before_ the secure channel is initialized
+type Handshake1 struct {
+	// protocolVersion determines compatibility between peers
+	ProtocolVersion *string `protobuf:"bytes,1,opt,name=protocolVersion" json:"protocolVersion,omitempty"`
+	// agentVersion is like a UserAgent string in browsers, or client version in bittorrent
+	// includes the client name and client. e.g.   "go-ipfs/0.1.0"
+	AgentVersion     *string `protobuf:"bytes,2,opt,name=agentVersion" json:"agentVersion,omitempty"`
+	XXX_unrecognized []byte  `json:"-"`
+}
+
+func (m *Handshake1) Reset()         { *m = Handshake1{} }
+func (m *Handshake1) String() string { return proto.CompactTextString(m) }
+func (*Handshake1) ProtoMessage()    {}
+
+func (m *Handshake1) GetProtocolVersion() string {
+	if m != nil && m.ProtocolVersion != nil {
+		return *m.ProtocolVersion
+	}
+	return ""
+}
+
+func (m *Handshake1) GetAgentVersion() string {
+	if m != nil && m.AgentVersion != nil {
+		return *m.AgentVersion
+	}
+	return ""
+}
+
+// Handshake3 is delivered _after_ the secure channel is initialized
+type Handshake3 struct {
+	// can include all the values in handshake1, for protocol version, etc.
+	H1 *Handshake1 `protobuf:"bytes,5,opt,name=h1" json:"h1,omitempty"`
+	// publicKey is this node's public key (which also gives its node.ID)
+	// - may not need to be sent, as secure channel implies it has been sent.
+	// - then again, if we change / disable secure channel, may still want it.
+	PublicKey []byte `protobuf:"bytes,1,opt,name=publicKey" json:"publicKey,omitempty"`
+	// listenAddrs are the multiaddrs the sender node listens for open connections on
+	ListenAddrs [][]byte `protobuf:"bytes,2,rep,name=listenAddrs" json:"listenAddrs,omitempty"`
+	// protocols are the services this node is running
+	Protocols []string `protobuf:"bytes,3,rep,name=protocols" json:"protocols,omitempty"`
+	// oservedAddr is the multiaddr of the remote endpoint that the sender node perceives
+	// this is useful information to convey to the other side, as it helps the remote endpoint
+	// determine whether its connection to the local peer goes through NAT.
+	ObservedAddr     []byte `protobuf:"bytes,4,opt,name=observedAddr" json:"observedAddr,omitempty"`
+	XXX_unrecognized []byte `json:"-"`
+}
+
+func (m *Handshake3) Reset()         { *m = Handshake3{} }
+func (m *Handshake3) String() string { return proto.CompactTextString(m) }
+func (*Handshake3) ProtoMessage()    {}
+
+func (m *Handshake3) GetH1() *Handshake1 {
+	if m != nil {
+		return m.H1
+	}
+	return nil
+}
+
+func (m *Handshake3) GetPublicKey() []byte {
+	if m != nil {
+		return m.PublicKey
+	}
+	return nil
+}
+
+func (m *Handshake3) GetListenAddrs() [][]byte {
+	if m != nil {
+		return m.ListenAddrs
+	}
+	return nil
+}
+
+func (m *Handshake3) GetProtocols() []string {
+	if m != nil {
+		return m.Protocols
+	}
+	return nil
+}
+
+func (m *Handshake3) GetObservedAddr() []byte {
+	if m != nil {
+		return m.ObservedAddr
+	}
+	return nil
+}
+
+func init() {
+}

net/handshake/pb/handshake.proto

+package handshake.pb;
+
+//import "github.com/jbenet/go-ipfs/net/mux/mux.proto";
+
+// Handshake1 is delivered _before_ the secure channel is initialized
+message Handshake1 {
+  // protocolVersion determines compatibility between peers
+  optional string protocolVersion = 1; // semver
+
+  // agentVersion is like a UserAgent string in browsers, or client version in bittorrent
+  // includes the client name and client. e.g.   "go-ipfs/0.1.0"
+  optional string agentVersion = 2; // semver
+
+  // we'll have more fields here later.
+}
+
+// Handshake3 is delivered _after_ the secure channel is initialized
+message Handshake3 {
+
+  // can include all the values in handshake1, for protocol version, etc.
+  optional Handshake1 h1 = 5;
+
+  // publicKey is this node's public key (which also gives its node.ID)
+  // - may not need to be sent, as secure channel implies it has been sent.
+  // - then again, if we change / disable secure channel, may still want it.
+  optional bytes publicKey = 1;
+
+  // listenAddrs are the multiaddrs the sender node listens for open connections on
+  repeated bytes listenAddrs = 2;
+
+  // protocols are the services this node is running
+  repeated string protocols = 3;
+
+  // oservedAddr is the multiaddr of the remote endpoint that the sender node perceives
+  // this is useful information to convey to the other side, as it helps the remote endpoint
+  // determine whether its connection to the local peer goes through NAT.
+  optional bytes observedAddr = 4;
+}

net/interface.go

+package net
+
+import (
+	"io"
+
+	conn "github.com/jbenet/go-ipfs/p2p/net/conn"
+	// swarm "github.com/jbenet/go-ipfs/p2p/net/swarm2"
+	peer "github.com/jbenet/go-ipfs/p2p/peer"
+
+	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
+	ctxgroup "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-ctxgroup"
+	ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
+)
+
+// ProtocolID is an identifier used to write protocol headers in streams.
+type ProtocolID string
+
+// These are the ProtocolIDs of the protocols running. It is useful
+// to keep them in one place.
+const (
+	ProtocolTesting  ProtocolID = "/ipfs/testing"
+	ProtocolBitswap  ProtocolID = "/ipfs/bitswap"
+	ProtocolDHT      ProtocolID = "/ipfs/dht"
+	ProtocolIdentify ProtocolID = "/ipfs/id"
+	ProtocolDiag     ProtocolID = "/ipfs/diagnostics"
+	ProtocolRelay    ProtocolID = "/ipfs/relay"
+)
+
+// MessageSizeMax is a soft (recommended) maximum for network messages.
+// One can write more, as the interface is a stream. But it is useful
+// to bunch it up into multiple read/writes when the whole message is
+// a single, large serialized object.
+const MessageSizeMax = 2 << 22 // 4MB
+
+// Stream represents a bidirectional channel between two agents in
+// the IPFS network. "agent" is as granular as desired, potentially
+// being a "request -> reply" pair, or whole protocols.
+// Streams are backed by SPDY streams underneath the hood.
+type Stream interface {
+	io.Reader
+	io.Writer
+	io.Closer
+
+	// Conn returns the connection this stream is part of.
+	Conn() Conn
+}
+
+// StreamHandler is the function protocols who wish to listen to
+// incoming streams must implement.
+type StreamHandler func(Stream)
+
+type StreamHandlerMap map[ProtocolID]StreamHandler
+
+// Conn is a connection to a remote peer. It multiplexes streams.
+// Usually there is no need to use a Conn directly, but it may
+// be useful to get information about the peer on the other side:
+//  stream.Conn().RemotePeer()
+type Conn interface {
+	conn.PeerConn
+
+	// NewStreamWithProtocol constructs a new Stream over this conn.
+	NewStreamWithProtocol(pr ProtocolID) (Stream, error)
+}
+
+// Network is the interface IPFS uses for connecting to the world.
+// It dials and listens for connections. it uses a Swarm to pool
+// connnections (see swarm pkg, and peerstream.Swarm). Connections
+// are encrypted with a TLS-like protocol.
+type Network interface {
+	Dialer
+	io.Closer
+
+	// SetHandler sets the protocol handler on the Network's Muxer.
+	// This operation is threadsafe.
+	SetHandler(ProtocolID, StreamHandler)
+
+	// Protocols returns the list of protocols this network currently
+	// has registered handlers for.
+	Protocols() []ProtocolID
+
+	// NewStream returns a new stream to given peer p.
+	// If there is no connection to p, attempts to create one.
+	// If ProtocolID is "", writes no header.
+	NewStream(ProtocolID, peer.ID) (Stream, error)
+
+	// BandwidthTotals returns the total number of bytes passed through
+	// the network since it was instantiated
+	BandwidthTotals() (uint64, uint64)
+
+	// ListenAddresses returns a list of addresses at which this network listens.
+	ListenAddresses() []ma.Multiaddr
+
+	// InterfaceListenAddresses returns a list of addresses at which this network
+	// listens. It expands "any interface" addresses (/ip4/0.0.0.0, /ip6/::) to
+	// use the known local interfaces.
+	InterfaceListenAddresses() ([]ma.Multiaddr, error)
+
+	// CtxGroup returns the network's contextGroup
+	CtxGroup() ctxgroup.ContextGroup
+}
+
+// Dialer represents a service that can dial out to peers
+// (this is usually just a Network, but other services may not need the whole
+// stack, and thus it becomes easier to mock)
+type Dialer interface {
+
+	// Peerstore returns the internal peerstore
+	// This is useful to tell the dialer about a new address for a peer.
+	// Or use one of the public keys found out over the network.
+	Peerstore() peer.Peerstore
+
+	// LocalPeer returns the local peer associated with this network
+	LocalPeer() peer.ID
+
+	// DialPeer attempts to establish a connection to a given peer
+	DialPeer(context.Context, peer.ID) error
+
+	// ClosePeer closes the connection to a given peer
+	ClosePeer(peer.ID) error
+
+	// Connectedness returns a state signaling connection capabilities
+	Connectedness(peer.ID) Connectedness
+
+	// Peers returns the peers connected
+	Peers() []peer.ID
+
+	// Conns returns the connections in this Netowrk
+	Conns() []Conn
+
+	// ConnsToPeer returns the connections in this Netowrk for given peer.
+	ConnsToPeer(p peer.ID) []Conn
+}
+
+// Connectedness signals the capacity for a connection with a given node.
+// It is used to signal to services and other peers whether a node is reachable.
+type Connectedness int
+
+const (
+	// NotConnected means no connection to peer, and no extra information (default)
+	NotConnected Connectedness = iota
+
+	// Connected means has an open, live connection to peer
+	Connected
+
+	// CanConnect means recently connected to peer, terminated gracefully
+	CanConnect
+
+	// CannotConnect means recently attempted connecting but failed to connect.
+	// (should signal "made effort, failed")
+	CannotConnect
+)