Merge pull request #1438 from weaveworks/fdp

Fast datapath

prog/weaver/main.go

@@ -13,6 +13,7 @@ import (
 	"github.com/davecheney/profile"
 	"github.com/docker/docker/pkg/mflag"
 	"github.com/gorilla/mux"
+	"github.com/weaveworks/go-odp/odp"
 
 	. "github.com/weaveworks/weave/common"
 	"github.com/weaveworks/weave/common/docker"
@@ -35,8 +36,13 @@ func main() {
 	runtime.GOMAXPROCS(procs)
 
 	var (
+		// flags that cause immediate exit
+		justVersion          bool
+		createDatapath       bool
+		deleteDatapath       bool
+		addDatapathInterface string
+
 		config                    weave.Config
-		justVersion               bool
 		protocolMinVersion        int
 		ifaceName                 string
 		routerName                string
@@ -60,9 +66,14 @@ func main() {
 		dnsClientTimeout          time.Duration
 		dnsEffectiveListenAddress string
 		iface                     *net.Interface
+		datapathName              string
 	)
 
 	mflag.BoolVar(&justVersion, []string{"#version", "-version"}, false, "print version and exit")
+	mflag.BoolVar(&createDatapath, []string{"-create-datapath"}, false, "create ODP datapath and exit")
+	mflag.BoolVar(&deleteDatapath, []string{"-delete-datapath"}, false, "delete ODP datapath and exit")
+	mflag.StringVar(&addDatapathInterface, []string{"-add-datapath-iface"}, "", "add a network interface to the ODP datapath and exit")
+
 	mflag.IntVar(&config.Port, []string{"#port", "-port"}, weave.Port, "router port")
 	mflag.IntVar(&protocolMinVersion, []string{"-min-protocol-version"}, weave.ProtocolMinVersion, "minimum weave protocol version")
 	mflag.StringVar(&ifaceName, []string{"#iface", "-iface"}, "", "name of interface to capture/inject from (disabled if blank)")
@@ -86,6 +97,7 @@ func main() {
 	mflag.IntVar(&dnsTTL, []string{"-dns-ttl"}, nameserver.DefaultTTL, "TTL for DNS request from our domain")
 	mflag.DurationVar(&dnsClientTimeout, []string{"-dns-fallback-timeout"}, nameserver.DefaultClientTimeout, "timeout for fallback DNS requests")
 	mflag.StringVar(&dnsEffectiveListenAddress, []string{"-dns-effective-listen-address"}, "", "address DNS will actually be listening, after Docker port mapping")
+	mflag.StringVar(&datapathName, []string{"-datapath"}, "", "ODP datapath name")
 
 	// crude way of detecting that we probably have been started in a
 	// container, with `weave launch` --> suppress misleading paths in
@@ -94,14 +106,37 @@ func main() {
 		os.Args[0] = "weave"
 		mflag.CommandLine.Init("weave", mflag.ExitOnError)
 	}
+
 	mflag.Parse()
 
 	peers = mflag.Args()
 
 	SetLogLevel(logLevel)
-	if justVersion {
+
+	switch {
+	case justVersion:
 		fmt.Printf("weave router %s\n", version)
 		os.Exit(0)
+
+	case createDatapath:
+		err := weave.CreateDatapath(datapathName)
+		if odp.IsKernelLacksODPError(err) {
+			// When the kernel lacks ODP support, exit
+			// with a special status to distinguish it for
+			// the weave script.
+			os.Exit(17)
+		}
+
+		checkFatal(err)
+		os.Exit(0)
+
+	case deleteDatapath:
+		checkFatal(weave.DeleteDatapath(datapathName))
+		os.Exit(0)
+
+	case addDatapathInterface != "":
+		checkFatal(weave.AddDatapathInterface(datapathName, addDatapathInterface))
+		os.Exit(0)
 	}
 
 	Log.Println("Command line options:", options())
@@ -112,21 +147,58 @@ func main() {
 	}
 	config.ProtocolMinVersion = byte(protocolMinVersion)
 
-	var err error
+	var fastDPOverlay weave.Overlay
+	if datapathName != "" {
+		// A datapath name implies that "Bridge" and "Overlay"
+		// packet handling use fast datapath, although other
+		// options can override that below.  Even if both
+		// things are overridden, we might need bridging on
+		// the datapath.
+		fastdp, err := weave.NewFastDatapath(weave.FastDatapathConfig{
+			DatapathName: datapathName,
+			Port:         config.Port,
+		})
+
+		checkFatal(err)
+		config.Bridge = fastdp.Bridge()
+		fastDPOverlay = fastdp.Overlay()
+	}
 
 	if ifaceName != "" {
+		// -iface can coexist with -datapath, because
+		// pcap-based packet capture is a bit more efficient
+		// than capture via ODP misses, even when using an
+		// ODP-based bridge.  So when using weave encyption,
+		// it's preferable to use -iface.
 		iface, err := weavenet.EnsureInterface(ifaceName)
-		if err != nil {
-			Log.Fatal(err)
-		}
+		checkFatal(err)
 
 		// bufsz flag is in MB
 		config.Bridge, err = weave.NewPcap(iface, bufSzMB*1024*1024)
-		if err != nil {
-			Log.Fatal(err)
-		}
+		checkFatal(err)
 	}
 
+	if password == "" {
+		password = os.Getenv("WEAVE_PASSWORD")
+	}
+
+	if password == "" {
+		Log.Println("Communication between peers is unencrypted.")
+	} else {
+		config.Password = []byte(password)
+		Log.Println("Communication between peers is encrypted.")
+
+		// fastdp doesn't support encryption
+		fastDPOverlay = nil
+	}
+
+	overlays := weave.NewOverlaySwitch()
+	if fastDPOverlay != nil {
+		overlays.Add("fastdp", fastDPOverlay)
+	}
+	overlays.Add("sleeve", weave.NewSleeveOverlay(config.Port))
+	config.Overlay = overlays
+
 	if routerName == "" {
 		if iface == nil {
 			Log.Fatal("Either an interface must be specified with --iface or a name with -name")
@@ -135,25 +207,11 @@ func main() {
 	}
 
 	name, err := weave.PeerNameFromUserInput(routerName)
-	if err != nil {
-		Log.Fatal(err)
-	}
+	checkFatal(err)
 
 	if nickName == "" {
 		nickName, err = os.Hostname()
-		if err != nil {
-			Log.Fatal(err)
-		}
-	}
-
-	if password == "" {
-		password = os.Getenv("WEAVE_PASSWORD")
-	}
-	if password == "" {
-		Log.Println("Communication between peers is unencrypted.")
-	} else {
-		config.Password = []byte(password)
-		Log.Println("Communication between peers is encrypted.")
+		checkFatal(err)
 	}
 
 	if prof != "" {
@@ -164,7 +222,6 @@ func main() {
 	}
 
 	config.PeerDiscovery = !noDiscovery
-	config.Overlay = weave.NewSleeveOverlay(config.Port)
 
 	if pktdebug {
 		config.PacketLogging = packetLogging{}
@@ -285,9 +342,8 @@ func (nopPacketLogging) LogForwardPacket(string, weave.ForwardPacketKey) {
 
 func parseAndCheckCIDR(cidrStr string) address.CIDR {
 	_, cidr, err := address.ParseCIDR(cidrStr)
-	if err != nil {
-		Log.Fatal(err)
-	}
+	checkFatal(err)
+
 	if cidr.Size() < ipam.MinSubnetSize {
 		Log.Fatalf("Allocation range smaller than minimum size %d: %s", ipam.MinSubnetSize, cidrStr)
 	}
@@ -346,3 +402,9 @@ func listenAndServeHTTP(httpAddr string, muxRouter *mux.Router) {
 		Log.Fatal("Unable to create http server", err)
 	}
 }
+
+func checkFatal(e error) {
+	if e != nil {
+		Log.Fatal(e)
+	}
+}

proxy/common.go

@@ -24,15 +24,28 @@ var (
 
 func callWeave(args ...string) ([]byte, []byte, error) {
 	args = append([]string{"--local"}, args...)
-	Log.Debug("Calling weave", args)
 	cmd := exec.Command("./weave", args...)
 	cmd.Env = []string{
 		"PATH=/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
 		"PROCFS=/hostproc",
 	}
-	if bridge := os.Getenv("DOCKER_BRIDGE"); bridge != "" {
-		cmd.Env = append(cmd.Env, fmt.Sprintf("DOCKER_BRIDGE=%s", bridge))
+
+	propagateEnv := func(key string) {
+		if val := os.Getenv(key); val != "" {
+			cmd.Env = append(cmd.Env, fmt.Sprintf("%s=%s", key, val))
+		}
 	}
+
+	propagateEnv("DOCKER_BRIDGE")
+
+	// Propogage WEAVE_DEBUG, to make debugging easier.
+	propagateEnv("WEAVE_DEBUG")
+
+	// This prevents the code coverage contortions in our
+	// integration test suite breaking things.
+	propagateEnv("COVERAGE")
+
+	Log.Debug("Calling weave args: ", args, "env: ", cmd.Env)
 	var stdout, stderr bytes.Buffer
 	cmd.Stdout = &stdout
 	cmd.Stderr = &stderr

proxy/proxy.go

@@ -436,13 +436,15 @@ func (proxy *Proxy) getDNSDomain() (domain string) {
 	}
 
 	weaveContainer, err := proxy.client.InspectContainer("weave")
-	if err != nil ||
-		weaveContainer.NetworkSettings == nil ||
-		weaveContainer.NetworkSettings.IPAddress == "" {
-		return
+	var weaveIP string
+	if err == nil && weaveContainer.NetworkSettings != nil {
+		weaveIP = weaveContainer.NetworkSettings.IPAddress
+	}
+	if weaveIP == "" {
+		weaveIP = "127.0.0.1"
 	}
 
-	url := fmt.Sprintf("http://%s:%d/domain", weaveContainer.NetworkSettings.IPAddress, router.HTTPPort)
+	url := fmt.Sprintf("http://%s:%d/domain", weaveIP, router.HTTPPort)
 	resp, err := http.Get(url)
 	if err != nil || resp.StatusCode != http.StatusOK {
 		return

router/connection.go

@@ -115,8 +115,6 @@ func (conn *LocalConnection) BreakTie(dupConn Connection) ConnectionTieBreak {
 }
 
 func (conn *LocalConnection) Established() bool {
-	conn.RLock()
-	defer conn.RUnlock()
 	return conn.established
 }
 
@@ -201,6 +199,7 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch
 		ConnUID:            conn.uid,
 		Crypto:             conn.forwarderCrypto(),
 		SendControlMessage: conn.sendOverlayControlMessage,
+		Features:           intro.Features,
 	}
 	if conn.forwarder, err = conn.Router.Overlay.MakeForwarder(params); err != nil {
 		return
@@ -214,17 +213,16 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch
 	}
 	conn.Router.ConnectionMaker.ConnectionCreated(conn)
 
-	// SetListener has the side-effect of telling the forwarder
-	// that the connection is confirmed.  This comes after
-	// AddConnection, because only after that completes do we know
-	// the connection is valid: in particular that it is not a
-	// duplicate connection to the same peer. Sending heartbeats
-	// on a duplicate connection can trip up crypto at the other
-	// end, since the associated UDP packets may get decoded by
-	// the other connection. It is also generally wasteful to
-	// engage in any interaction with the remote on a connection
-	// that turns out to be invalid.
-	conn.forwarder.SetListener(ConnectionAsForwarderListener{conn})
+	// Forwarder confirmation comes after AddConnection, because
+	// only after that completes do we know the connection is
+	// valid: in particular that it is not a duplicate connection
+	// to the same peer. Sending heartbeats on a duplicate
+	// connection can trip up crypto at the other end, since the
+	// associated UDP packets may get decoded by the other
+	// connection. It is also generally wasteful to engage in any
+	// interaction with the remote on a connection that turns out
+	// to be invalid.
+	conn.forwarder.Confirm()
 
 	// receiveTCP must follow also AddConnection. In the absence
 	// of any indirect connectivity to the remote peer, the first
@@ -251,13 +249,16 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch
 }
 
 func (conn *LocalConnection) makeFeatures() map[string]string {
-	return map[string]string{
+	features := map[string]string{
 		"PeerNameFlavour": PeerNameFlavour,
 		"Name":            conn.local.Name.String(),
 		"NickName":        conn.local.NickName,
+		"ShortID":         fmt.Sprint(conn.local.ShortID),
 		"UID":             fmt.Sprint(conn.local.UID),
 		"ConnID":          fmt.Sprint(conn.uid),
 	}
+	conn.Router.Overlay.AddFeaturesTo(features)
+	return features
 }
 
 type features map[string]string
@@ -276,7 +277,7 @@ func (features features) Get(key string) string {
 }
 
 func (conn *LocalConnection) parseFeatures(features features) (*Peer, error) {
-	if err := features.MustHave([]string{"PeerNameFlavour", "Name", "NickName", "UID", "ConnID"}); err != nil {
+	if err := features.MustHave([]string{"PeerNameFlavour", "Name", "NickName", "ShortID", "UID", "ConnID"}); err != nil {
 		return nil, err
 	}
 
@@ -292,6 +293,12 @@ func (conn *LocalConnection) parseFeatures(features features) (*Peer, error) {
 
 	nickName := features.Get("NickName")
 
+	shortID, err := strconv.ParseUint(features.Get("ShortID"), 10,
+		PeerShortIDBits)
+	if err != nil {
+		return nil, err
+	}
+
 	uid, err := ParsePeerUID(features.Get("UID"))
 	if err != nil {
 		return nil, err
@@ -303,7 +310,7 @@ func (conn *LocalConnection) parseFeatures(features features) (*Peer, error) {
 	}
 
 	conn.uid ^= remoteConnID
-	return NewPeer(name, nickName, uid, 0), nil
+	return NewPeer(name, nickName, uid, 0, PeerShortID(shortID)), nil
 }
 
 func (conn *LocalConnection) registerRemote(remote *Peer, acceptNewPeer bool) error {
@@ -328,12 +335,23 @@ func (conn *LocalConnection) registerRemote(remote *Peer, acceptNewPeer bool) er
 }
 
 func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction) (err error) {
+	fwdErrorChan := conn.forwarder.ErrorChannel()
+	fwdEstablishedChan := conn.forwarder.EstablishedChannel()
+
 	for err == nil {
 		select {
 		case action := <-actionChan:
 			err = action()
+
 		case <-conn.heartbeatTCP.C:
 			err = conn.sendSimpleProtocolMsg(ProtocolHeartbeat)
+
+		case <-fwdEstablishedChan:
+			conn.established = true
+			fwdEstablishedChan = nil
+			conn.Router.Ourself.ConnectionEstablished(conn)
+
+		case err = <-fwdErrorChan:
 		}
 	}
 	return
@@ -377,27 +395,8 @@ func (conn *LocalConnection) forwarderCrypto() *OverlayCrypto {
 	}
 }
 
-func (conn *LocalConnection) sendOverlayControlMessage(tag ProtocolTag, msg []byte) error {
-	return conn.sendProtocolMsg(ProtocolMsg{tag, msg})
-}
-
-type ConnectionAsForwarderListener struct{ conn *LocalConnection }
-
-func (l ConnectionAsForwarderListener) Established() {
-	l.conn.sendAction(func() error {
-		old := l.conn.established
-		l.conn.Lock()
-		l.conn.established = true
-		l.conn.Unlock()
-		if !old {
-			l.conn.Router.Ourself.ConnectionEstablished(l.conn)
-		}
-		return nil
-	})
-}
-
-func (l ConnectionAsForwarderListener) Error(err error) {
-	l.conn.sendAction(func() error { return err })
+func (conn *LocalConnection) sendOverlayControlMessage(tag byte, msg []byte) error {
+	return conn.sendProtocolMsg(ProtocolMsg{ProtocolTag(tag), msg})
 }
 
 // Helpers
@@ -434,7 +433,7 @@ func (conn *LocalConnection) handleProtocolMsg(tag ProtocolTag, payload []byte)
 	switch tag {
 	case ProtocolHeartbeat:
 	case ProtocolConnectionEstablished, ProtocolFragmentationReceived, ProtocolPMTUVerified, ProtocolOverlayControlMsg:
-		conn.forwarder.ControlMessage(tag, payload)
+		conn.forwarder.ControlMessage(byte(tag), payload)
 	case ProtocolGossipUnicast, ProtocolGossipBroadcast, ProtocolGossip:
 		return conn.Router.handleGossip(tag, payload)
 	default:

router/consts.go

@@ -6,12 +6,16 @@ import (
 )
 
 const (
-	Port             = 6783
-	HTTPPort         = Port + 1
-	MaxUDPPacketSize = 65535
-	ChannelSize      = 16
-	TCPHeartbeat     = 30 * time.Second
-	GossipInterval   = 30 * time.Second
-	MaxDuration      = time.Duration(math.MaxInt64)
-	MaxTCPMsgSize    = 10 * 1024 * 1024
+	Port                = 6783
+	HTTPPort            = Port + 1
+	MaxUDPPacketSize    = 65535
+	ChannelSize         = 16
+	TCPHeartbeat        = 30 * time.Second
+	GossipInterval      = 30 * time.Second
+	MaxDuration         = time.Duration(math.MaxInt64)
+	MaxTCPMsgSize       = 10 * 1024 * 1024
+	FastHeartbeat       = 500 * time.Millisecond
+	SlowHeartbeat       = 10 * time.Second
+	MaxMissedHeartbeats = 6
+	HeartbeatTimeout    = MaxMissedHeartbeats * SlowHeartbeat
 )

router/flow.go

@@ -34,12 +34,18 @@ type MultiFlowOp struct {
 	ops       []FlowOp
 }
 
-func NewMultiFlowOp(broadcast bool) *MultiFlowOp {
-	return &MultiFlowOp{broadcast: broadcast}
+func NewMultiFlowOp(broadcast bool, ops ...FlowOp) *MultiFlowOp {
+	mfop := &MultiFlowOp{broadcast: broadcast}
+	mfop.Add(ops...)
+	return mfop
 }
 
 func (mfop *MultiFlowOp) Add(ops ...FlowOp) {
-	mfop.ops = append(mfop.ops, ops...)
+	for _, op := range ops {
+		if op != nil {
+			mfop.ops = append(mfop.ops, op)
+		}
+	}
 }
 
 func (mfop *MultiFlowOp) Process(frame []byte, dec *EthernetDecoder,
@@ -48,3 +54,23 @@ func (mfop *MultiFlowOp) Process(frame []byte, dec *EthernetDecoder,
 		op.Process(frame, dec, mfop.broadcast)
 	}
 }
+
+func FlattenFlowOp(fop FlowOp) []FlowOp {
+	return collectFlowOps(nil, fop)
+}
+
+func collectFlowOps(into []FlowOp, fop FlowOp) []FlowOp {
+	if fop == nil {
+		return into
+	}
+
+	if mfop, ok := fop.(*MultiFlowOp); ok {
+		for _, op := range mfop.ops {
+			into = collectFlowOps(into, op)
+		}
+
+		return into
+	}
+
+	return append(into, fop)
+}

router/local_peer.go

@@ -19,7 +19,8 @@ type LocalPeerAction func()
 func NewLocalPeer(name PeerName, nickName string, router *Router) *LocalPeer {
 	actionChan := make(chan LocalPeerAction, ChannelSize)
 	peer := &LocalPeer{
-		Peer:       NewPeer(name, nickName, randomPeerUID(), 0),
+		Peer: NewPeer(name, nickName, randomPeerUID(), 0,
+			randomPeerShortID()),
 		router:     router,
 		actionChan: actionChan,
 	}
@@ -160,7 +161,10 @@ func (peer *LocalPeer) handleAddConnection(conn Connection) error {
 		conn.Log("connection added (new peer)")
 		peer.router.SendAllGossipDown(conn)
 	}
+
+	peer.router.Routes.Recalculate()
 	peer.broadcastPeerUpdate(conn.Remote())
+
 	return nil
 }
 
@@ -174,6 +178,8 @@ func (peer *LocalPeer) handleConnectionEstablished(conn Connection) {
 	}
 	peer.connectionEstablished(conn)
 	conn.Log("connection fully established")
+
+	peer.router.Routes.Recalculate()
 	peer.broadcastPeerUpdate()
 }
 
@@ -193,14 +199,20 @@ func (peer *LocalPeer) handleDeleteConnection(conn Connection) {
 	// Must do garbage collection first to ensure we don't send out an
 	// update with unreachable peers (can cause looping)
 	peer.router.Peers.GarbageCollect()
+	peer.router.Routes.Recalculate()
 	peer.broadcastPeerUpdate()
 }
 
 // helpers
 
 func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) {
-	peer.router.Routes.Recalculate()
-	peer.router.TopologyGossip.GossipBroadcast(NewTopologyGossipData(peer.router.Peers, append(peers, peer.Peer)...))
+	// Some tests run without a router.  This should be fixed so
+	// that the relevant part of Router can be easily run in the
+	// context of a test, but that will involve significant
+	// reworking of tests.
+	if peer.router != nil {
+		peer.router.BroadcastTopologyUpdate(append(peers, peer.Peer))
+	}
 }
 
 func (peer *LocalPeer) checkConnectionLimit() error {
@@ -236,3 +248,10 @@ func (peer *LocalPeer) connectionCount() int {
 	defer peer.RUnlock()
 	return len(peer.connections)
 }
+
+func (peer *LocalPeer) setShortID(shortID PeerShortID) {
+	peer.Lock()
+	defer peer.Unlock()
+	peer.ShortID = shortID
+	peer.Version++
+}

router/mocks_test.go

@@ -70,7 +70,7 @@ func checkEqualConns(t *testing.T, ourName PeerName, got, wanted map[PeerName]Co
 // Get all the peers from a Peers in a slice
 func (peers *Peers) allPeers() []*Peer {
 	var res []*Peer
-	for _, peer := range peers.table {
+	for _, peer := range peers.byName {
 		res = append(res, peer)
 	}
 	return res

router/odp.go

@@ -0,0 +1,60 @@
+package router
+
+import (
+	"github.com/weaveworks/go-odp/odp"
+)
+
+// ODP admin functionality
+
+func CreateDatapath(dpname string) error {
+	dpif, err := odp.NewDpif()
+	if err != nil {
+		return err
+	}
+
+	defer dpif.Close()
+
+	_, err = dpif.CreateDatapath(dpname)
+	if err != nil && !odp.IsDatapathNameAlreadyExistsError(err) {
+		return err
+	}
+
+	return nil
+}
+
+func DeleteDatapath(dpname string) error {
+	dpif, err := odp.NewDpif()
+	if err != nil {
+		return err
+	}
+
+	defer dpif.Close()
+
+	dp, err := dpif.LookupDatapath(dpname)
+	if err != nil {
+		if odp.IsNoSuchDatapathError(err) {
+			return nil
+		}
+
+		return err
+	}
+
+	return dp.Delete()
+}
+
+func AddDatapathInterface(dpname string, ifname string) error {
+	dpif, err := odp.NewDpif()
+	if err != nil {
+		return err
+	}
+
+	defer dpif.Close()
+
+	dp, err := dpif.LookupDatapath(dpname)
+	if err != nil {
+		return err
+	}
+
+	_, err = dp.CreateVport(odp.NewNetdevVportSpec(ifname))
+	return err
+}

router/overlay.go

@@ -15,6 +15,12 @@ type Overlay interface {
 	// The routes have changed, so any cached information should
 	// be discarded.
 	InvalidateRoutes()
+
+	// A mapping of a short id to a peer has changed
+	InvalidateShortIDs()
+
+	// Enhance a features map with overlay-related features
+	AddFeaturesTo(map[string]string)
 }
 
 type ForwarderParams struct {
@@ -38,7 +44,10 @@ type ForwarderParams struct {
 
 	// Function to send a control message to the counterpart
 	// forwarder.
-	SendControlMessage func(tag ProtocolTag, msg []byte) error
+	SendControlMessage func(tag byte, msg []byte) error
+
+	// Features passed at connection initiation
+	Features map[string]string
 }
 
 // When a consumer is called, the decoder will already have been used
@@ -54,27 +63,31 @@ type OverlayCrypto struct {
 
 // All of the machinery to forward packets to a particular peer
 type OverlayForwarder interface {
-	// Register a callback for forwarder state changes.
-	// side-effect, calling this confirms that the connection is
-	// really wanted, and so the provider should activate it.
-	// However, Forward might be called before this is called
-	// (e.g. on another thread).
-	SetListener(OverlayForwarderListener)
-
-	// Forward a packet across the connection.
+	// Forward a packet across the connection.  May be called as
+	// soon as the forwarder is created, in particular before
+	// Confirm().
 	Forward(ForwardPacketKey) FlowOp
 
+	// Confirm that the connection is really wanted, and so the
+	// Overlay should begin heartbeats etc. to verify the
+	// operation of the forwarder.
+	Confirm()
+
+	// A channel indicating that the forwarder is established,
+	// i.e. its operation has been confirmed.
+	EstablishedChannel() <-chan struct{}
+
+	// A channel indicating an error from the forwarder.  The
+	// forwarder is not expected to be operational after the first
+	// error, so the channel only needs to buffer a single error.
+	ErrorChannel() <-chan error
+
 	Stop()
 
 	// Handle a message from the peer.  'tag' exists for
 	// compatibility, and should always be
 	// ProtocolOverlayControlMessage for non-sleeve overlays.
-	ControlMessage(tag ProtocolTag, msg []byte)
-}
-
-type OverlayForwarderListener interface {
-	Established()
-	Error(error)
+	ControlMessage(tag byte, msg []byte)
 }
 
 type NullOverlay struct{}
@@ -90,7 +103,21 @@ func (NullOverlay) MakeForwarder(ForwarderParams) (OverlayForwarder, error) {
 func (NullOverlay) InvalidateRoutes() {
 }
 
-func (NullOverlay) SetListener(OverlayForwarderListener) {
+func (NullOverlay) InvalidateShortIDs() {
+}
+
+func (NullOverlay) Confirm() {
+}
+
+func (NullOverlay) EstablishedChannel() <-chan struct{} {
+	return nil
+}
+
+func (NullOverlay) ErrorChannel() <-chan error {
+	return nil
+}
+
+func (NullOverlay) AddFeaturesTo(map[string]string) {
 }
 
 func (NullOverlay) Forward(ForwardPacketKey) FlowOp {
@@ -100,5 +127,5 @@ func (NullOverlay) Forward(ForwardPacketKey) FlowOp {
 func (NullOverlay) Stop() {
 }
 
-func (NullOverlay) ControlMessage(ProtocolTag, []byte) {
+func (NullOverlay) ControlMessage(byte, []byte) {
 }

router/overlay_switch.go

@@ -0,0 +1,395 @@
+package router
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+)
+
+// OverlaySwitch selects which overlay to use, from a set of
+// subsidiary overlays.  First, it passes a list of supported overlays
+// in the connection features, and uses that to determine which
+// overlays are in common.  Then it tries those common overlays, and
+// uses the best one that seems to be working.
+
+type OverlaySwitch struct {
+	overlays     map[string]Overlay
+	overlayNames []string
+}
+
+func NewOverlaySwitch() *OverlaySwitch {
+	return &OverlaySwitch{overlays: make(map[string]Overlay)}
+}
+
+func (osw *OverlaySwitch) Add(name string, overlay Overlay) {
+	// check for repeated names
+	if _, present := osw.overlays[name]; present {
+		log.Fatal("OverlaySwitch: repeated overlay name")
+	}
+
+	osw.overlays[name] = overlay
+	osw.overlayNames = append(osw.overlayNames, name)
+}
+
+func (osw *OverlaySwitch) AddFeaturesTo(features map[string]string) {
+	features["Overlays"] = strings.Join(osw.overlayNames, " ")
+}
+
+func (osw *OverlaySwitch) InvalidateRoutes() {
+	for _, overlay := range osw.overlays {
+		overlay.InvalidateRoutes()
+	}
+}
+
+func (osw *OverlaySwitch) InvalidateShortIDs() {
+	for _, overlay := range osw.overlays {
+		overlay.InvalidateShortIDs()
+	}
+}
+
+func (osw *OverlaySwitch) StartConsumingPackets(localPeer *Peer, peers *Peers,
+	consumer OverlayConsumer) error {
+	for _, overlay := range osw.overlays {
+		if err := overlay.StartConsumingPackets(localPeer, peers,
+			consumer); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+type namedOverlay struct {
+	Overlay
+	name string
+}
+
+// Find the common set of overlays supported by both sides, with the
+// ordering being the same on both sides too.
+func (osw *OverlaySwitch) commonOverlays(params ForwarderParams) ([]namedOverlay, error) {
+	var peerOverlays []string
+	if overlaysFeature, present := params.Features["Overlays"]; present {
+		peerOverlays = strings.Split(overlaysFeature, " ")
+	}
+
+	common := make(map[string]Overlay)
+	for _, name := range peerOverlays {
+		if overlay := osw.overlays[name]; overlay != nil {
+			common[name] = overlay
+		}
+	}
+
+	if len(common) == 0 {
+		return nil, fmt.Errorf("no overlays in common with peer")
+	}
+
+	// we order them according to the connecting node
+	ordering := osw.overlayNames
+	if params.RemoteAddr == nil {
+		// we are the connectee
+		ordering = peerOverlays
+	}
+
+	res := make([]namedOverlay, 0, len(common))
+	for _, name := range ordering {
+		overlay := common[name]
+		if overlay != nil {
+			res = append(res, namedOverlay{overlay, name})
+		}
+	}
+
+	// we use bytes to represent forwarder indices in control
+	// messages, so just in case:
+	if len(res) > 256 {
+		res = res[:256]
+	}
+
+	return res, nil
+}
+
+type overlaySwitchForwarder struct {
+	remotePeer *Peer
+
+	lock sync.Mutex
+
+	// the forwarder to send on
+	best OverlayForwarder
+
+	// the subsidiary forwarders
+	forwarders []subForwarder
+
+	// closed to tell the main goroutine to stop
+	stopChan chan<- struct{}
+
+	alreadyEstablished bool
+	establishedChan    chan struct{}
+	errorChan          chan error
+}
+
+// A subsidiary forwarder
+type subForwarder struct {
+	fwd         OverlayForwarder
+	overlayName string
+
+	// Has the forwarder signalled that it is established?
+	established bool
+
+	// closed to tell the forwarder monitor goroutine to stop
+	stopChan chan<- struct{}
+}
+
+// An event from a subsidiary forwarder
+type subForwarderEvent struct {
+	// the index of the forwarder
+	index int
+
+	// is this an "established" event?
+	established bool
+
+	// is this an error event?
+	err error
+}
+
+func (osw *OverlaySwitch) MakeForwarder(
+	params ForwarderParams) (OverlayForwarder, error) {
+	overlays, err := osw.commonOverlays(params)
+	if err != nil {
+		return nil, err
+	}
+
+	// channel to carry events from the subforwarder monitors to
+	// the main goroutine
+	eventsChan := make(chan subForwarderEvent)
+
+	// channel to stop the main goroutine
+	stopChan := make(chan struct{})
+
+	fwd := &overlaySwitchForwarder{
+		remotePeer: params.RemotePeer,
+
+		forwarders: make([]subForwarder, len(overlays)),
+		stopChan:   stopChan,
+
+		establishedChan: make(chan struct{}),
+		errorChan:       make(chan error, 1),
+	}
+
+	origSendControlMessage := params.SendControlMessage
+	for i, overlay := range overlays {
+		// Prefix control messages to indicate the relevant forwarder
+		index := i
+		params.SendControlMessage = func(tag byte, msg []byte) error {
+			xmsg := make([]byte, len(msg)+2)
+			xmsg[0] = byte(index)
+			xmsg[1] = tag
+			copy(xmsg[2:], msg)
+			return origSendControlMessage(ProtocolOverlayControlMsg,
+				xmsg)
+		}
+
+		subFwd, err := overlay.MakeForwarder(params)
+		if err != nil {
+			fwd.stopFrom(0)
+			return nil, err
+		}
+
+		subStopChan := make(chan struct{})
+		go monitorForwarder(i, eventsChan, subStopChan, subFwd)
+		fwd.forwarders[i] = subForwarder{
+			fwd:         subFwd,
+			overlayName: overlay.name,
+			stopChan:    subStopChan,
+		}
+	}
+
+	fwd.chooseBest()
+	go fwd.run(eventsChan, stopChan)
+	return fwd, nil
+}
+
+func monitorForwarder(index int, eventsChan chan<- subForwarderEvent,
+	stopChan <-chan struct{}, fwd OverlayForwarder) {
+	establishedChan := fwd.EstablishedChannel()
+loop:
+	for {
+		e := subForwarderEvent{index: index}
+
+		select {
+		case <-establishedChan:
+			e.established = true
+			establishedChan = nil
+
+		case err := <-fwd.ErrorChannel():
+			e.err = err
+
+		case <-stopChan:
+			break loop
+		}
+
+		select {
+		case eventsChan <- e:
+		case <-stopChan:
+			break loop
+		}
+
+		if e.err != nil {
+			break loop
+		}
+	}
+
+	fwd.Stop()
+}
+
+func (fwd *overlaySwitchForwarder) run(eventsChan <-chan subForwarderEvent,
+	stopChan <-chan struct{}) {
+loop:
+	for {
+		select {
+		case <-stopChan:
+			break loop
+
+		case e := <-eventsChan:
+			switch {
+			case e.established:
+				fwd.established(e.index)
+			case e.err != nil:
+				fwd.error(e.index, e.err)
+			}
+		}
+	}
+
+	fwd.lock.Lock()
+	defer fwd.lock.Unlock()
+	fwd.stopFrom(0)
+}
+
+func (fwd *overlaySwitchForwarder) established(index int) {
+	fwd.lock.Lock()
+	defer fwd.lock.Unlock()
+
+	fwd.forwarders[index].established = true
+
+	// less preferred forwarders are no longer needed
+	fwd.stopFrom(index + 1)
+
+	if !fwd.alreadyEstablished {
+		fwd.alreadyEstablished = true
+		close(fwd.establishedChan)
+	}
+
+	fwd.chooseBest()
+}
+
+func (fwd *overlaySwitchForwarder) logPrefix() string {
+	return fmt.Sprintf("overlay_switch ->[%s] ", fwd.remotePeer)
+}
+
+func (fwd *overlaySwitchForwarder) error(index int, err error) {
+	fwd.lock.Lock()
+	defer fwd.lock.Unlock()
+
+	log.Info(fwd.logPrefix(), fwd.forwarders[index].overlayName, " ", err)
+	fwd.forwarders[index].fwd = nil
+	fwd.chooseBest()
+}
+
+func (fwd *overlaySwitchForwarder) stopFrom(index int) {
+	for index < len(fwd.forwarders) {
+		subFwd := &fwd.forwarders[index]
+		if subFwd.fwd != nil {
+			subFwd.fwd = nil
+			close(subFwd.stopChan)
+		}
+		index++
+	}
+}
+
+func (fwd *overlaySwitchForwarder) chooseBest() {
+	// the most preferred established forwarder is the best
+	// otherwise, the most preferred working forwarder is the best
+	var bestEstablished, bestWorking *subForwarder
+
+	for i := range fwd.forwarders {
+		subFwd := &fwd.forwarders[i]
+		if subFwd.fwd == nil {
+			continue
+		}
+
+		if bestWorking == nil {
+			bestWorking = subFwd
+		}
+
+		if bestEstablished == nil && subFwd.established {
+			bestEstablished = subFwd
+		}
+	}
+
+	best := bestEstablished
+	if best == nil {
+		if bestWorking == nil {
+			select {
+			case fwd.errorChan <- fmt.Errorf("no working forwarders to %s", fwd.remotePeer):
+			default:
+			}
+
+			return
+		}
+
+		best = bestWorking
+	}
+
+	if fwd.best != best.fwd {
+		fwd.best = best.fwd
+		log.Info(fwd.logPrefix(), "using ", best.overlayName)
+	}
+}
+
+func (fwd *overlaySwitchForwarder) Confirm() {
+	var forwarders []OverlayForwarder
+
+	fwd.lock.Lock()
+	for _, subFwd := range fwd.forwarders {
+		if subFwd.fwd != nil {
+			forwarders = append(forwarders, subFwd.fwd)
+		}
+	}
+	fwd.lock.Unlock()
+
+	for _, subFwd := range forwarders {
+		subFwd.Confirm()
+	}
+}
+
+func (fwd *overlaySwitchForwarder) Forward(pk ForwardPacketKey) FlowOp {
+	fwd.lock.Lock()
+	best := fwd.best
+	fwd.lock.Unlock()
+
+	if best == nil {
+		return nil
+	}
+
+	return best.Forward(pk)
+}
+
+func (fwd *overlaySwitchForwarder) EstablishedChannel() <-chan struct{} {
+	return fwd.establishedChan
+}
+
+func (fwd *overlaySwitchForwarder) ErrorChannel() <-chan error {
+	return fwd.errorChan
+}
+
+func (fwd *overlaySwitchForwarder) Stop() {
+	fwd.lock.Lock()
+	defer fwd.lock.Unlock()
+	fwd.stopFrom(0)
+}
+
+func (fwd *overlaySwitchForwarder) ControlMessage(tag byte, msg []byte) {
+	fwd.lock.Lock()
+	subFwd := fwd.forwarders[msg[0]].fwd
+	fwd.lock.Unlock()
+	if subFwd != nil {
+		subFwd.ControlMessage(msg[1], msg[2:])
+	}
+}

router/peer.go

@@ -17,11 +17,24 @@ func ParsePeerUID(s string) (PeerUID, error) {
 	return PeerUID(uid), err
 }
 
+// Short IDs exist for the sake of fast datapath.  They are 12 bits,
+// randomly assigned, but we detect and recover from collisions.  This
+// does limit us to 4096 peers, but that should be sufficient for a
+// while.
+type PeerShortID uint16
+
+const PeerShortIDBits = 12
+
+func randomPeerShortID() PeerShortID {
+	return PeerShortID(randUint16() & (1<<PeerShortIDBits - 1))
+}
+
 type PeerSummary struct {
 	NameByte []byte
 	NickName string
 	UID      PeerUID
 	Version  uint64
+	ShortID  PeerShortID
 }
 
 type Peer struct {
@@ -41,12 +54,13 @@ func NewPeerFromSummary(summary PeerSummary) *Peer {
 	}
 }
 
-func NewPeer(name PeerName, nickName string, uid PeerUID, version uint64) *Peer {
+func NewPeer(name PeerName, nickName string, uid PeerUID, version uint64, shortID PeerShortID) *Peer {
 	return NewPeerFromSummary(PeerSummary{
 		NameByte: name.Bin(),
 		NickName: nickName,
 		UID:      uid,
 		Version:  version,
+		ShortID:  shortID,
 	})
 }
 

router/peers.go

@@ -4,14 +4,29 @@ import (
 	"bytes"
 	"encoding/gob"
 	"io"
+	"math/rand"
 	"sync"
 )
 
 type Peers struct {
 	sync.RWMutex
-	ourself *LocalPeer
-	table   map[PeerName]*Peer
-	onGC    []func(*Peer)
+	ourself   *LocalPeer
+	byName    map[PeerName]*Peer
+	byShortID map[PeerShortID]ShortIDPeers
+	onGC      []func(*Peer)
+
+	// Called when the mapping from short ids to peers changes
+	onInvalidateShortIDs []func()
+}
+
+type ShortIDPeers struct {
+	// If we know about a single peer with the short id, this is
+	// that peer.  If there is a collision, this is the peer with
+	// the lowest Name.
+	peer *Peer
+
+	// In case of a collision, this holds the other peers.
+	others []*Peer
 }
 
 type UnknownPeerError struct {
@@ -31,32 +46,223 @@ type ConnectionSummary struct {
 	Established   bool
 }
 
+// Pending notifications due to changes to Peers that need to be sent
+// out once the Peers is unlocked.
+type PeersPendingNotifications struct {
+	// Peers that have been GCed
+	removed []*Peer
+
+	// The mapping from shorts ids to peers changed
+	invalidateShortIDs bool
+
+	// The local short ID needs reassigning due to a collision
+	reassignLocalShortID bool
+}
+
 func NewPeers(ourself *LocalPeer) *Peers {
-	return &Peers{ourself: ourself, table: make(map[PeerName]*Peer)}
+	peers := &Peers{
+		ourself:   ourself,
+		byName:    make(map[PeerName]*Peer),
+		byShortID: make(map[PeerShortID]ShortIDPeers),
+	}
+	peers.FetchWithDefault(ourself.Peer)
+	return peers
 }
 
 func (peers *Peers) OnGC(callback func(*Peer)) {
 	peers.Lock()
 	defer peers.Unlock()
+
+	// Although the array underlying peers.onGC might be accessed
+	// without holding the lock in unlockAndNotify, we don't
+	// support removing callbacks, so a simple append here is
+	// safe.
 	peers.onGC = append(peers.onGC, callback)
 }
 
-func (peers *Peers) invokeOnGCCallbacks(removed []*Peer) {
-	for _, callback := range peers.onGC {
-		for _, peer := range removed {
-			callback(peer)
+func (peers *Peers) OnInvalidateShortIDs(callback func()) {
+	peers.Lock()
+	defer peers.Unlock()
+
+	// Safe, as in OnGC
+	peers.onInvalidateShortIDs = append(peers.onInvalidateShortIDs, callback)
+}
+
+func (peers *Peers) unlockAndNotify(pending *PeersPendingNotifications) {
+	broadcastLocalPeer := pending.reassignLocalShortID && peers.reassignLocalShortID(pending)
+	onGC := peers.onGC
+	onInvalidateShortIDs := peers.onInvalidateShortIDs
+	peers.Unlock()
+
+	if pending.removed != nil {
+		for _, callback := range onGC {
+			for _, peer := range pending.removed {
+				callback(peer)
+			}
 		}
 	}
+
+	if pending.invalidateShortIDs {
+		for _, callback := range onInvalidateShortIDs {
+			callback()
+		}
+	}
+
+	if broadcastLocalPeer {
+		peers.ourself.broadcastPeerUpdate()
+	}
+}
+
+func (peers *Peers) addByShortID(peer *Peer, pending *PeersPendingNotifications) {
+	entry, ok := peers.byShortID[peer.ShortID]
+	if !ok {
+		entry = ShortIDPeers{peer: peer}
+	} else if entry.peer == nil {
+		// This short ID is free, but was used in the past.
+		// Because we are reusing it, it's an invalidation
+		// event.
+		entry.peer = peer
+		pending.invalidateShortIDs = true
+	} else if peer.Name < entry.peer.Name {
+		// Short ID collision, this peer becomes the principal
+		// peer for the short ID, bumping the previous one
+		// into others.
+
+		if entry.peer == peers.ourself.Peer {
+			// The bumped peer is peers.ourself, so we
+			// need to look for a new short id
+			pending.reassignLocalShortID = true
+		}
+
+		entry.others = append(entry.others, entry.peer)
+		entry.peer = peer
+		pending.invalidateShortIDs = true
+	} else {
+		// Short ID collision, this peer is secondary
+		entry.others = append(entry.others, peer)
+	}
+
+	peers.byShortID[peer.ShortID] = entry
+}
+
+func (peers *Peers) deleteByShortID(peer *Peer, pending *PeersPendingNotifications) {
+	entry := peers.byShortID[peer.ShortID]
+	var otherIndex int
+
+	if peer != entry.peer {
+		// peer is secondary, find its index in others
+		otherIndex = -1
+
+		for i, other := range entry.others {
+			if peer == other {
+				otherIndex = i
+				break
+			}
+		}
+
+		if otherIndex < 0 {
+			return
+		}
+	} else if len(entry.others) != 0 {
+		// need to find the peer with the lowest name to
+		// become the new principal one
+		otherIndex = 0
+		minName := entry.others[0].Name
+
+		for i := 1; i < len(entry.others); i++ {
+			otherName := entry.others[i].Name
+			if otherName < minName {
+				minName = otherName
+				otherIndex = i
+			}
+		}
+
+		entry.peer = entry.others[otherIndex]
+		pending.invalidateShortIDs = true
+	} else {
+		// This is the last peer with the short id.  We clear
+		// the entry, don't delete it, in order to detect when
+		// it gets re-used.
+		peers.byShortID[peer.ShortID] = ShortIDPeers{}
+		return
+	}
+
+	entry.others[otherIndex] = entry.others[len(entry.others)-1]
+	entry.others = entry.others[:len(entry.others)-1]
+	peers.byShortID[peer.ShortID] = entry
+}
+
+func (peers *Peers) reassignLocalShortID(pending *PeersPendingNotifications) bool {
+	newShortID, ok := peers.chooseShortID()
+	if ok {
+		peers.setLocalShortID(newShortID, pending)
+		return true
+	}
+
+	// Otherwise we'll try again later on in garbageColleect
+	return false
+}
+
+func (peers *Peers) setLocalShortID(newShortID PeerShortID, pending *PeersPendingNotifications) {
+	peers.deleteByShortID(peers.ourself.Peer, pending)
+	peers.ourself.setShortID(newShortID)
+	peers.addByShortID(peers.ourself.Peer, pending)
+}
+
+// Choose an available short id at random
+func (peers *Peers) chooseShortID() (PeerShortID, bool) {
+	rng := rand.New(rand.NewSource(int64(randUint64())))
+
+	// First, just try picking some short ids at random, and
+	// seeing if they are available:
+	for i := 0; i < 10; i++ {
+		shortID := PeerShortID(rng.Intn(1 << PeerShortIDBits))
+		if peers.byShortID[shortID].peer == nil {
+			return shortID, true
+		}
+	}
+
+	// Looks like most short ids are used.  So count the number of
+	// unused ones, and pick one at random.
+	available := int(1 << PeerShortIDBits)
+	for _, entry := range peers.byShortID {
+		if entry.peer != nil {
+			available--
+		}
+	}
+
+	if available == 0 {
+		// All short ids are used.
+		return 0, false
+	}
+
+	n := rng.Intn(available)
+	var i PeerShortID
+	for {
+		if peers.byShortID[i].peer == nil {
+			if n == 0 {
+				return PeerShortID(i), true
+			}
+
+			n--
+		}
+
+		i++
+	}
 }
 
 func (peers *Peers) FetchWithDefault(peer *Peer) *Peer {
 	peers.Lock()
-	defer peers.Unlock()
-	if existingPeer, found := peers.table[peer.Name]; found {
+	var pending PeersPendingNotifications
+	defer peers.unlockAndNotify(&pending)
+
+	if existingPeer, found := peers.byName[peer.Name]; found {
 		existingPeer.localRefCount++
 		return existingPeer
 	}
-	peers.table[peer.Name] = peer
+
+	peers.byName[peer.Name] = peer
+	peers.addByShortID(peer, &pending)
 	peer.localRefCount++
 	return peer
 }
@@ -64,19 +270,25 @@ func (peers *Peers) FetchWithDefault(peer *Peer) *Peer {
 func (peers *Peers) Fetch(name PeerName) *Peer {
 	peers.RLock()
 	defer peers.RUnlock()
-	return peers.table[name]
+	return peers.byName[name]
 }
 
 func (peers *Peers) FetchAndAddRef(name PeerName) *Peer {
 	peers.Lock()
 	defer peers.Unlock()
-	peer := peers.table[name]
+	peer := peers.byName[name]
 	if peer != nil {
 		peer.localRefCount++
 	}
 	return peer
 }
 
+func (peers *Peers) FetchByShortID(shortID PeerShortID) *Peer {
+	peers.RLock()
+	defer peers.RUnlock()
+	return peers.byShortID[shortID].peer
+}
+
 func (peers *Peers) Dereference(peer *Peer) {
 	peers.Lock()
 	defer peers.Unlock()
@@ -86,7 +298,7 @@ func (peers *Peers) Dereference(peer *Peer) {
 func (peers *Peers) ForEach(fun func(*Peer)) {
 	peers.RLock()
 	defer peers.RUnlock()
-	for _, peer := range peers.table {
+	for _, peer := range peers.byName {
 		fun(peer)
 	}
 }
@@ -99,42 +311,55 @@ func (peers *Peers) ForEach(fun func(*Peer)) {
 // "improved" update containing just these new/updated elements.
 func (peers *Peers) ApplyUpdate(update []byte) (PeerNameSet, PeerNameSet, error) {
 	peers.Lock()
+	var pending PeersPendingNotifications
+	defer peers.unlockAndNotify(&pending)
 
 	newPeers, decodedUpdate, decodedConns, err := peers.decodeUpdate(update)
 	if err != nil {
-		peers.Unlock()
 		return nil, nil, err
 	}
 
+	ourVersion := peers.ourself.Version
+
 	// By this point, we know the update doesn't refer to any peers we
 	// have no knowledge of. We can now apply the update. Start by
-	// adding in any new peers into the cache.
+	// adding in any new peers.
 	for name, newPeer := range newPeers {
-		peers.table[name] = newPeer
+		peers.byName[name] = newPeer
+		peers.addByShortID(newPeer, &pending)
 	}
 
 	// Now apply the updates
-	newUpdate := peers.applyUpdate(decodedUpdate, decodedConns)
-	removed := peers.garbageCollect()
-	for _, peerRemoved := range removed {
+	newUpdate := peers.applyUpdate(decodedUpdate, decodedConns, &pending)
+	peers.garbageCollect(&pending)
+	for _, peerRemoved := range pending.removed {
 		delete(newUpdate, peerRemoved.Name)
 	}
 
-	// Don't need to hold peers lock any longer
-	peers.Unlock()
-	peers.invokeOnGCCallbacks(removed)
-
 	updateNames := make(PeerNameSet)
 	for _, peer := range decodedUpdate {
 		updateNames[peer.Name] = void
 	}
-	return updateNames, setFromPeersMap(newUpdate), nil
+
+	if peers.ourself.Version != ourVersion {
+		// Our short id changed, i.e. due to a short ID
+		// change.  So we need to include ourself in the
+		// update
+		newUpdate[peers.ourself.Name] = void
+	}
+
+	return updateNames, newUpdate, nil
 }
 
 func (peers *Peers) Names() PeerNameSet {
 	peers.RLock()
 	defer peers.RUnlock()
-	return setFromPeersMap(peers.table)
+
+	names := make(PeerNameSet)
+	for name := range peers.byName {
+		names[name] = void
+	}
+	return names
 }
 
 func (peers *Peers) EncodePeers(names PeerNameSet) []byte {
@@ -143,7 +368,7 @@ func (peers *Peers) EncodePeers(names PeerNameSet) []byte {
 	peers.RLock()
 	defer peers.RUnlock()
 	for name := range names {
-		if peer, found := peers.table[name]; found {
+		if peer, found := peers.byName[name]; found {
 			if peer == peers.ourself.Peer {
 				peers.ourself.Encode(enc)
 			} else {
@@ -154,34 +379,33 @@ func (peers *Peers) EncodePeers(names PeerNameSet) []byte {
 	return buf.Bytes()
 }
 
-func (peers *Peers) GarbageCollect() []*Peer {
+func (peers *Peers) GarbageCollect() {
 	peers.Lock()
-	removed := peers.garbageCollect()
-	peers.Unlock()
-	peers.invokeOnGCCallbacks(removed)
-	return removed
+	var pending PeersPendingNotifications
+	defer peers.unlockAndNotify(&pending)
+
+	peers.garbageCollect(&pending)
 }
 
-func (peers *Peers) garbageCollect() []*Peer {
-	removed := []*Peer{}
+func (peers *Peers) garbageCollect(pending *PeersPendingNotifications) {
 	peers.ourself.RLock()
 	_, reached := peers.ourself.Routes(nil, false)
 	peers.ourself.RUnlock()
-	for name, peer := range peers.table {
+
+	for name, peer := range peers.byName {
 		if _, found := reached[peer.Name]; !found && peer.localRefCount == 0 {
-			delete(peers.table, name)
-			removed = append(removed, peer)
+			delete(peers.byName, name)
+			peers.deleteByShortID(peer, pending)
+			pending.removed = append(pending.removed, peer)
 		}
 	}
-	return removed
-}
 
-func setFromPeersMap(peers map[PeerName]*Peer) PeerNameSet {
-	names := make(PeerNameSet)
-	for name := range peers {
-		names[name] = void
+	if peers.byShortID[peers.ourself.ShortID].peer != peers.ourself.Peer {
+		// The local peer doesn't own its short id.  Garbage
+		// collection might have freed some up, so try to
+		// reassign.
+		pending.reassignLocalShortID = true
 	}
-	return names
 }
 
 func (peers *Peers) decodeUpdate(update []byte) (newPeers map[PeerName]*Peer, decodedUpdate []*Peer, decodedConns [][]ConnectionSummary, err error) {
@@ -202,7 +426,7 @@ func (peers *Peers) decodeUpdate(update []byte) (newPeers map[PeerName]*Peer, de
 		newPeer := NewPeerFromSummary(peerSummary)
 		decodedUpdate = append(decodedUpdate, newPeer)
 		decodedConns = append(decodedConns, connSummaries)
-		existingPeer, found := peers.table[newPeer.Name]
+		existingPeer, found := peers.byName[newPeer.Name]
 		if !found {
 			newPeers[newPeer.Name] = newPeer
 		} else if existingPeer.UID != newPeer.UID {
@@ -217,7 +441,7 @@ func (peers *Peers) decodeUpdate(update []byte) (newPeers map[PeerName]*Peer, de
 			if _, found := newPeers[remoteName]; found {
 				continue
 			}
-			if _, found := peers.table[remoteName]; found {
+			if _, found := peers.byName[remoteName]; found {
 				continue
 			}
 			// Update refers to a peer which we have no knowledge
@@ -229,13 +453,13 @@ func (peers *Peers) decodeUpdate(update []byte) (newPeers map[PeerName]*Peer, de
 	return
 }
 
-func (peers *Peers) applyUpdate(decodedUpdate []*Peer, decodedConns [][]ConnectionSummary) map[PeerName]*Peer {
-	newUpdate := make(map[PeerName]*Peer)
+func (peers *Peers) applyUpdate(decodedUpdate []*Peer, decodedConns [][]ConnectionSummary, pending *PeersPendingNotifications) PeerNameSet {
+	newUpdate := make(PeerNameSet)
 	for idx, newPeer := range decodedUpdate {
 		connSummaries := decodedConns[idx]
 		name := newPeer.Name
-		// guaranteed to find peer in the peers.table
-		peer := peers.table[name]
+		// guaranteed to find peer in the peers.byName
+		peer := peers.byName[name]
 		if peer != newPeer &&
 			(peer == peers.ourself.Peer || peer.Version >= newPeer.Version) {
 			// Nobody but us updates us. And if we know more about a
@@ -254,9 +478,17 @@ func (peers *Peers) applyUpdate(decodedUpdate []*Peer, decodedConns [][]Connecti
 		// router.Peers.ApplyUpdate. But ApplyUpdate takes the Lock on
 		// the router.Peers, so there can be no race here.
 		peer.Version = newPeer.Version
-		peer.connections = makeConnsMap(peer, connSummaries, peers.table)
-		newUpdate[name] = peer
+		peer.connections = makeConnsMap(peer, connSummaries, peers.byName)
+
+		if newPeer.ShortID != peer.ShortID {
+			peers.deleteByShortID(peer, pending)
+			peer.ShortID = newPeer.ShortID
+			peers.addByShortID(peer, pending)
+		}
+
+		newUpdate[name] = void
 	}
+
 	return newUpdate
 }
 
@@ -292,11 +524,11 @@ func decodePeer(dec *gob.Decoder) (peerSummary PeerSummary, connSummaries []Conn
 	return
 }
 
-func makeConnsMap(peer *Peer, connSummaries []ConnectionSummary, table map[PeerName]*Peer) map[PeerName]Connection {
+func makeConnsMap(peer *Peer, connSummaries []ConnectionSummary, byName map[PeerName]*Peer) map[PeerName]Connection {
 	conns := make(map[PeerName]Connection)
 	for _, connSummary := range connSummaries {
 		name := PeerNameFromBin(connSummary.NameByte)
-		remotePeer := table[name]
+		remotePeer := byName[name]
 		conn := NewRemoteConnection(peer, remotePeer, connSummary.RemoteTCPAddr, connSummary.Outbound, connSummary.Established)
 		conns[name] = conn
 	}

router/peers_test.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"math/rand"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 )
@@ -20,7 +21,6 @@ import (
 func newNode(name PeerName) (*Peer, *Peers) {
 	peer := NewLocalPeer(name, "", nil)
 	peers := NewPeers(peer)
-	peers.FetchWithDefault(peer.Peer)
 	return peer.Peer, peers
 }
 
@@ -72,6 +72,13 @@ func TestPeersEncoding(t *testing.T) {
 	}
 }
 
+func garbageCollect(peers *Peers) []*Peer {
+	var removed []*Peer
+	peers.OnGC(func(peer *Peer) { removed = append(removed, peer) })
+	peers.GarbageCollect()
+	return removed
+}
+
 func TestPeersGarbageCollection(t *testing.T) {
 	const (
 		peer1NameString = "01:00:00:01:00:00"
@@ -98,17 +105,272 @@ func TestPeersGarbageCollection(t *testing.T) {
 	ps2.AddTestRemoteConnection(p3, p1)
 
 	// Every peer is referenced, so nothing should be dropped
-	require.Empty(t, ps1.GarbageCollect(), "peers removed")
-	require.Empty(t, ps2.GarbageCollect(), "peers removed")
-	require.Empty(t, ps3.GarbageCollect(), "peers removed")
+	require.Empty(t, garbageCollect(ps1), "peers removed")
+	require.Empty(t, garbageCollect(ps2), "peers removed")
+	require.Empty(t, garbageCollect(ps3), "peers removed")
 
 	// Drop the connection from 2 to 3, and 3 isn't garbage-collected
 	// because 1 has a connection to 3
 	ps2.DeleteTestConnection(p3)
-	require.Empty(t, ps2.GarbageCollect(), "peers removed")
+	require.Empty(t, garbageCollect(ps2), "peers removed")
 
 	// Drop the connection from 1 to 3, and 3 will get removed by
 	// garbage-collection
 	ps1.DeleteTestConnection(p3)
-	checkPeerArray(t, ps1.GarbageCollect(), p3)
+	checkPeerArray(t, garbageCollect(ps1), p3)
+}
+
+func TestShortIDCollisions(t *testing.T) {
+	rng := rand.New(rand.NewSource(time.Now().UnixNano()))
+	_, peers := newNode(PeerName(1 << PeerShortIDBits))
+
+	// Make enough peers that short id collisions are
+	// overwhelmingly likely
+	ps := make([]*Peer, 1<<PeerShortIDBits)
+	for i := 0; i < 1<<PeerShortIDBits; i++ {
+		ps[i] = NewPeer(PeerName(i), "", PeerUID(i), 0,
+			PeerShortID(rng.Intn(1<<PeerShortIDBits)))
+	}
+
+	shuffle := func() {
+		for i := range ps {
+			j := rng.Intn(i + 1)
+			ps[i], ps[j] = ps[j], ps[i]
+		}
+	}
+
+	// Fill peers
+	shuffle()
+	var pending PeersPendingNotifications
+	for _, p := range ps {
+		peers.addByShortID(p, &pending)
+	}
+
+	// Check invariants
+	counts := make([]int, 1<<PeerShortIDBits)
+	saw := func(p *Peer) {
+		if p != peers.ourself.Peer {
+			counts[p.UID]++
+		}
+	}
+
+	for shortID, entry := range peers.byShortID {
+		if entry.peer == nil {
+			// no principal peer for this short id, so
+			// others must be empty
+			require.Empty(t, entry.others)
+			continue
+		}
+
+		require.Equal(t, shortID, entry.peer.ShortID)
+		saw(entry.peer)
+
+		for _, p := range entry.others {
+			saw(p)
+			require.Equal(t, shortID, p.ShortID)
+
+			// the principal peer should have the lowest name
+			require.True(t, p.Name > entry.peer.Name)
+		}
+	}
+
+	// Check that every peer was seen
+	for _, n := range counts {
+		require.Equal(t, 1, n)
+	}
+
+	// Delete all the peers
+	shuffle()
+	for _, p := range ps {
+		peers.deleteByShortID(p, &pending)
+	}
+
+	for _, entry := range peers.byShortID {
+		if entry.peer != peers.ourself.Peer {
+			require.Nil(t, entry.peer)
+		}
+
+		require.Empty(t, entry.others)
+	}
+}
+
+// Test the easy case of short id reassignment, when few short ids are taken
+func TestShortIDReassignmentEasy(t *testing.T) {
+	rng := rand.New(rand.NewSource(time.Now().UnixNano()))
+	_, peers := newNode(PeerName(0))
+
+	for i := 1; i <= 10; i++ {
+		peers.FetchWithDefault(NewPeer(PeerName(i), "", PeerUID(i), 0,
+			PeerShortID(rng.Intn(1<<PeerShortIDBits))))
+	}
+
+	checkShortIDReassignment(t, peers)
+}
+
+// Test the hard case of short id reassignment, when most short ids are taken
+func TestShortIDReassignmentHard(t *testing.T) {
+	rng := rand.New(rand.NewSource(time.Now().UnixNano()))
+	_, peers := newNode(PeerName(1 << PeerShortIDBits))
+
+	// Take all short ids
+	ps := make([]*Peer, 1<<PeerShortIDBits)
+	var pending PeersPendingNotifications
+	for i := 0; i < 1<<PeerShortIDBits; i++ {
+		ps[i] = NewPeer(PeerName(i), "", PeerUID(i), 0,
+			PeerShortID(i))
+		peers.addByShortID(ps[i], &pending)
+	}
+
+	// As all short ids are taken, an attempted reassigment won't
+	// do anything
+	oldShortID := peers.ourself.ShortID
+	require.False(t, peers.reassignLocalShortID(&pending))
+	require.Equal(t, oldShortID, peers.ourself.ShortID)
+
+	// Free up a few ids
+	for i := 0; i < 10; i++ {
+		x := rng.Intn(len(ps))
+		if ps[x] != nil {
+			peers.deleteByShortID(ps[x], &pending)
+			ps[x] = nil
+		}
+	}
+
+	checkShortIDReassignment(t, peers)
+}
+
+func checkShortIDReassignment(t *testing.T, peers *Peers) {
+	oldShortID := peers.ourself.ShortID
+	peers.reassignLocalShortID(&PeersPendingNotifications{})
+	require.NotEqual(t, oldShortID, peers.ourself.ShortID)
+	require.Equal(t, peers.ourself.Peer, peers.byShortID[peers.ourself.ShortID].peer)
+}
+
+func TestShortIDInvalidation(t *testing.T) {
+	_, peers := newNode(PeerName(1 << PeerShortIDBits))
+
+	// need to use a short id that is not the local peer's
+	shortID := peers.ourself.ShortID + 1
+
+	var pending PeersPendingNotifications
+
+	requireInvalidateShortIDs := func(expect bool) {
+		require.Equal(t, expect, pending.invalidateShortIDs)
+		pending.invalidateShortIDs = false
+	}
+
+	// The use of a fresh short id does not cause invalidation
+	a := NewPeer(PeerName(1), "", PeerUID(1), 0, shortID)
+	peers.addByShortID(a, &pending)
+	requireInvalidateShortIDs(false)
+
+	// An addition which does not change the mapping
+	b := NewPeer(PeerName(2), "", PeerUID(2), 0, shortID)
+	peers.addByShortID(b, &pending)
+	requireInvalidateShortIDs(false)
+
+	// An addition which does change the mapping
+	c := NewPeer(PeerName(0), "", PeerUID(0), 0, shortID)
+	peers.addByShortID(c, &pending)
+	requireInvalidateShortIDs(true)
+
+	// A deletion which does not change the mapping
+	peers.deleteByShortID(b, &pending)
+	requireInvalidateShortIDs(false)
+
+	// A deletion which does change the mapping
+	peers.deleteByShortID(c, &pending)
+	requireInvalidateShortIDs(true)
+
+	// Deleting the last peer with a short id does not cause invalidation
+	peers.deleteByShortID(a, &pending)
+	requireInvalidateShortIDs(false)
+
+	// .. but subsequent reuse of that short id does cause invalidation
+	peers.addByShortID(a, &pending)
+	requireInvalidateShortIDs(true)
+}
+
+func TestShortIDPropagation(t *testing.T) {
+	_, peers1 := newNode(PeerName(1))
+	_, peers2 := newNode(PeerName(2))
+
+	peers1.AddTestConnection(peers2.ourself.Peer)
+	peers1.ApplyUpdate(peers2.EncodePeers(peers2.Names()))
+	peers12 := peers1.Fetch(PeerName(2))
+	old := peers12.PeerSummary
+
+	require.True(t,
+		peers2.reassignLocalShortID(&PeersPendingNotifications{}))
+	peers1.ApplyUpdate(peers2.EncodePeers(peers2.Names()))
+	require.NotEqual(t, old.Version, peers12.Version)
+	require.NotEqual(t, old.ShortID, peers12.ShortID)
+}
+
+func TestShortIDCollision(t *testing.T) {
+	// Create 3 peers
+	_, peers1 := newNode(PeerName(1))
+	_, peers2 := newNode(PeerName(2))
+	_, peers3 := newNode(PeerName(3))
+
+	var pending PeersPendingNotifications
+	peers1.setLocalShortID(1, &pending)
+	peers2.setLocalShortID(2, &pending)
+	peers3.setLocalShortID(3, &pending)
+
+	peers2.AddTestConnection(peers1.ourself.Peer)
+	peers3.AddTestConnection(peers2.ourself.Peer)
+
+	// Propogate from 1 to 2 to 3
+	peers2.ApplyUpdate(peers1.EncodePeers(peers1.Names()))
+	peers3.ApplyUpdate(peers2.EncodePeers(peers2.Names()))
+
+	// Force the short id of peer 1 to collide with peer 2.  Peer
+	// 1 has the lowest name, so it gets to keep the short id
+	peers1.setLocalShortID(2, &pending)
+
+	oldShortID := peers2.ourself.ShortID
+	_, updated, _ := peers2.ApplyUpdate(peers1.EncodePeers(peers1.Names()))
+
+	// peer 2 should have noticed the collision and resolved it
+	require.NotEqual(t, oldShortID, peers2.ourself.ShortID)
+
+	// The Peers do not have a Router, so broadcastPeerUpdate does
+	// nothing in the context of this test.  So we fake what it
+	// would do.
+	updated[PeerName(2)] = void
+
+	// the update from peer 2 should include its short id change
+	peers3.ApplyUpdate(peers2.EncodePeers(updated))
+	require.Equal(t, peers2.ourself.ShortID,
+		peers3.Fetch(PeerName(2)).ShortID)
+}
+
+// Test the case where all short ids are taken, but then some peers go
+// away, so the local peer reassigns
+func TestDeferredShortIDReassignment(t *testing.T) {
+	rng := rand.New(rand.NewSource(time.Now().UnixNano()))
+	_, us := newNode(PeerName(1 << PeerShortIDBits))
+
+	// Connect us to other peers occupying all short ids
+	others := make([]*Peers, 1<<PeerShortIDBits)
+	var pending PeersPendingNotifications
+	for i := range others {
+		_, others[i] = newNode(PeerName(i))
+		others[i].setLocalShortID(PeerShortID(i), &pending)
+		us.AddTestConnection(others[i].ourself.Peer)
+	}
+
+	// Check that, as expected, the local peer does not own its
+	// short id
+	require.NotEqual(t, us.ourself.Peer,
+		us.byShortID[us.ourself.ShortID].peer)
+
+	// Disconnect one peer, and we should now be able to claim its
+	// short id
+	other := others[rng.Intn(1<<PeerShortIDBits)]
+	us.DeleteTestConnection(other.ourself.Peer)
+	us.GarbageCollect()
+
+	require.Equal(t, us.ourself.Peer, us.byShortID[us.ourself.ShortID].peer)
 }

router/protocol.go

@@ -328,7 +328,7 @@ func (res *ProtocolIntroResults) setupCrypto(params ProtocolIntroParams, remoteP
 type ProtocolTag byte
 
 const (
-	ProtocolHeartbeat ProtocolTag = iota
+	ProtocolHeartbeat = iota
 	ProtocolConnectionEstablished
 	ProtocolFragmentationReceived
 	ProtocolPMTUVerified

router/router.go

@@ -62,18 +62,18 @@ func NewRouter(config Config, name PeerName, nickName string) *Router {
 		router.Overlay = NullOverlay{}
 	}
 
-	onMacExpiry := func(mac net.HardwareAddr, peer *Peer) {
-		log.Println("Expired MAC", mac, "at", peer)
-	}
-	onPeerGC := func(peer *Peer) {
+	router.Ourself = NewLocalPeer(name, nickName, router)
+	router.Macs = NewMacCache(macMaxAge,
+		func(mac net.HardwareAddr, peer *Peer) {
+			log.Println("Expired MAC", mac, "at", peer)
+		})
+	router.Peers = NewPeers(router.Ourself)
+	router.Peers.OnGC(func(peer *Peer) {
 		router.Macs.Delete(peer)
 		log.Println("Removed unreachable peer", peer)
-	}
-	router.Ourself = NewLocalPeer(name, nickName, router)
-	router.Macs = NewMacCache(macMaxAge, onMacExpiry)
-	router.Peers = NewPeers(router.Ourself)
-	router.Peers.OnGC(onPeerGC)
-	router.Peers.FetchWithDefault(router.Ourself.Peer)
+	})
+	router.Peers.OnInvalidateShortIDs(router.Overlay.InvalidateShortIDs)
+
 	router.Routes = NewRoutes(router.Ourself, router.Peers, router.Overlay.InvalidateRoutes)
 	router.ConnectionMaker = NewConnectionMaker(router.Ourself, router.Peers, router.Port, router.PeerDiscovery)
 	router.TopologyGossip = router.NewGossip("topology", router)
@@ -261,12 +261,16 @@ type TopologyGossipData struct {
 	update PeerNameSet
 }
 
-func NewTopologyGossipData(peers *Peers, update ...*Peer) *TopologyGossipData {
+func (router *Router) BroadcastTopologyUpdate(update []*Peer) {
 	names := make(PeerNameSet)
 	for _, p := range update {
 		names[p.Name] = void
 	}
-	return &TopologyGossipData{peers: peers, update: names}
+
+	router.TopologyGossip.GossipBroadcast(&TopologyGossipData{
+		peers:  router.Peers,
+		update: names,
+	})
 }
 
 func (d *TopologyGossipData) Merge(other GossipData) {

router/routes.go

@@ -65,21 +65,13 @@ func (routes *Routes) UnicastAll(name PeerName) (PeerName, bool) {
 func (routes *Routes) Broadcast(name PeerName) []PeerName {
 	routes.RLock()
 	defer routes.RUnlock()
-	hops, found := routes.broadcast[name]
-	if !found {
-		return []PeerName{}
-	}
-	return hops
+	return routes.broadcast[name]
 }
 
 func (routes *Routes) BroadcastAll(name PeerName) []PeerName {
 	routes.RLock()
 	defer routes.RUnlock()
-	hops, found := routes.broadcastAll[name]
-	if !found {
-		return []PeerName{}
-	}
-	return hops
+	return routes.broadcastAll[name]
 }
 
 // Choose min(log2(n_peers), n_neighbouring_peers) neighbours, with a
@@ -208,7 +200,7 @@ func (routes *Routes) calculateUnicast(establishedAndSymmetric bool) unicastRout
 // where <= is the subset relationship on keys of the returned map.
 func (routes *Routes) calculateBroadcast(establishedAndSymmetric bool) broadcastRoutes {
 	broadcast := make(broadcastRoutes)
-	for _, peer := range routes.peers.table {
+	for _, peer := range routes.peers.byName {
 		hops := []PeerName{}
 		if found, reached := peer.Routes(routes.ourself.Peer, establishedAndSymmetric); found {
 			routes.ourself.ForEachConnectedPeer(establishedAndSymmetric, reached,

router/sleeve.go

@@ -45,18 +45,14 @@ import (
 // sleeveForwarder.mtu                     <-------------------------->
 
 const (
-	EthernetOverhead    = 14
-	UDPOverhead         = 28 // 20 bytes for IPv4, 8 bytes for UDP
-	DefaultMTU          = 65535
-	FragTestSize        = 60001
-	PMTUDiscoverySize   = 60000
-	FastHeartbeat       = 500 * time.Millisecond
-	SlowHeartbeat       = 10 * time.Second
-	FragTestInterval    = 5 * time.Minute
-	MTUVerifyAttempts   = 8
-	MTUVerifyTimeout    = 10 * time.Millisecond // doubled with each attempt
-	MaxMissedHeartbeats = 6
-	HeartbeatTimeout    = MaxMissedHeartbeats * SlowHeartbeat
+	EthernetOverhead  = 14
+	UDPOverhead       = 28 // 20 bytes for IPv4, 8 bytes for UDP
+	DefaultMTU        = 65535
+	FragTestSize      = 60001
+	PMTUDiscoverySize = 60000
+	FragTestInterval  = 5 * time.Minute
+	MTUVerifyAttempts = 8
+	MTUVerifyTimeout  = 10 * time.Millisecond // doubled with each attempt
 )
 
 type SleeveOverlay struct {
@@ -129,6 +125,14 @@ func (*SleeveOverlay) InvalidateRoutes() {
 	// no cached information, so nothing to do
 }
 
+func (*SleeveOverlay) InvalidateShortIDs() {
+	// no cached information, so nothing to do
+}
+
+func (*SleeveOverlay) AddFeaturesTo(map[string]string) {
+	// No features to be provided, to facilitate compatibility
+}
+
 func (sleeve *SleeveOverlay) lookupForwarder(peer PeerName) *sleeveForwarder {
 	sleeve.lock.Lock()
 	defer sleeve.lock.Unlock()
@@ -277,7 +281,7 @@ type sleeveForwarder struct {
 	sleeve         *SleeveOverlay
 	remotePeer     *Peer
 	remotePeerBin  []byte
-	sendControlMsg func(ProtocolTag, []byte) error
+	sendControlMsg func(byte, []byte) error
 	connUID        uint64
 
 	// Channels to communicate with the aggregator goroutine
@@ -288,9 +292,12 @@ type sleeveForwarder struct {
 	confirmedChan    chan<- struct{}
 	finishedChan     <-chan struct{}
 
+	// listener channels
+	establishedChan chan struct{}
+	errorChan       chan error
+
 	// Explicitly locked state
 	lock       sync.RWMutex
-	listener   OverlayForwarderListener
 	remoteAddr *net.UDPAddr
 
 	// These fields are accessed and updated independently, so no
@@ -335,7 +342,7 @@ type specialFrame struct {
 
 // A control message
 type controlMessage struct {
-	tag ProtocolTag
+	tag byte
 	msg []byte
 }
 
@@ -371,6 +378,8 @@ func (sleeve *SleeveOverlay) MakeForwarder(params ForwarderParams) (OverlayForwa
 		controlMsgChan:   controlMsgChan,
 		confirmedChan:    confirmedChan,
 		finishedChan:     finishedChan,
+		establishedChan:  make(chan struct{}),
+		errorChan:        make(chan error, 1),
 		remoteAddr:       params.RemoteAddr,
 		mtu:              DefaultMTU,
 		crypto:           crypto,
@@ -386,7 +395,7 @@ func (sleeve *SleeveOverlay) MakeForwarder(params ForwarderParams) (OverlayForwa
 }
 
 func (fwd *sleeveForwarder) logPrefixFor(sender *net.UDPAddr) string {
-	return fmt.Sprintf("->[%s|%s]: ", sender, fwd.remotePeer)
+	return fmt.Sprintf("sleeve ->[%s|%s]: ", sender, fwd.remotePeer)
 }
 
 func (fwd *sleeveForwarder) logPrefix() string {
@@ -396,23 +405,23 @@ func (fwd *sleeveForwarder) logPrefix() string {
 	return fwd.logPrefixFor(remoteAddr)
 }
 
-func (fwd *sleeveForwarder) SetListener(listener OverlayForwarderListener) {
-	log.Debug(fwd.logPrefix(), "SetListener ", listener)
+func (fwd *sleeveForwarder) Confirm() {
+	log.Debug(fwd.logPrefix(), "Confirm")
 
-	fwd.lock.Lock()
-	fwd.listener = listener
-	fwd.lock.Unlock()
-
-	// Setting the listener confirms that the forwarder is really
-	// wanted
-	if listener != nil {
-		select {
-		case fwd.confirmedChan <- struct{}{}:
-		case <-fwd.finishedChan:
-		}
+	select {
+	case fwd.confirmedChan <- struct{}{}:
+	case <-fwd.finishedChan:
 	}
 }
 
+func (fwd *sleeveForwarder) EstablishedChannel() <-chan struct{} {
+	return fwd.establishedChan
+}
+
+func (fwd *sleeveForwarder) ErrorChannel() <-chan error {
+	return fwd.errorChan
+}
+
 type curriedForward struct {
 	fwd *sleeveForwarder
 	key ForwardPacketKey
@@ -577,7 +586,7 @@ func frameTooBig(frame []byte, mtu int) bool {
 	return len(frame) > mtu+EthernetOverhead
 }
 
-func (fwd *sleeveForwarder) ControlMessage(tag ProtocolTag, msg []byte) {
+func (fwd *sleeveForwarder) ControlMessage(tag byte, msg []byte) {
 	select {
 	case fwd.controlMsgChan <- controlMessage{tag, msg}:
 	case <-fwd.finishedChan:
@@ -586,7 +595,6 @@ func (fwd *sleeveForwarder) ControlMessage(tag ProtocolTag, msg []byte) {
 
 func (fwd *sleeveForwarder) Stop() {
 	fwd.sleeve.removeForwarder(fwd.remotePeer.Name, fwd)
-	fwd.SetListener(nil)
 
 	// Tell the forwarder goroutine to finish.  We don't need to
 	// wait for it.
@@ -660,9 +668,9 @@ loop:
 
 	fwd.lock.RLock()
 	defer fwd.lock.RUnlock()
-	if fwd.listener != nil {
-		fwd.listener.Error(err)
-	}
+
+	// this is the only place we send an error to errorChan
+	fwd.errorChan <- err
 }
 
 func (fwd *sleeveForwarder) aggregateAndSend(frame aggregatorFrame,
@@ -853,14 +861,14 @@ func (fwd *sleeveForwarder) setRemoteAddr(addr *net.UDPAddr) {
 func (fwd *sleeveForwarder) handleHeartbeatAck() error {
 	log.Debug(fwd.logPrefix(), "handleHeartbeatAck")
 
-	// The connection is now regarded as established
-	fwd.notifyEstablished()
-
 	if fwd.heartbeatInterval != SlowHeartbeat {
 		fwd.heartbeatInterval = SlowHeartbeat
 		if fwd.heartbeatTimer != nil {
 			fwd.heartbeatTimer.Reset(fwd.heartbeatInterval)
 		}
+
+		// The connection is now regarded as established
+		close(fwd.establishedChan)
 	}
 
 	fwd.fragTestTicker = time.NewTicker(FragTestInterval)
@@ -875,14 +883,6 @@ func (fwd *sleeveForwarder) handleHeartbeatAck() error {
 		make([]byte, PMTUDiscoverySize))
 }
 
-func (fwd *sleeveForwarder) notifyEstablished() {
-	fwd.lock.RLock()
-	defer fwd.lock.RUnlock()
-	if fwd.listener != nil {
-		fwd.listener.Established()
-	}
-}
-
 func (fwd *sleeveForwarder) sendFragTest() error {
 	log.Debug(fwd.logPrefix(), "sendFragTest")
 	fwd.stackFrag = false

router/utils.go

@@ -3,6 +3,7 @@ package router
 import (
 	"bytes"
 	"crypto/rand"
+	"encoding/binary"
 	"encoding/gob"
 	"fmt"
 	"net"
@@ -62,15 +63,19 @@ func Concat(elems ...[]byte) []byte {
 	return res
 }
 
-func randUint64() (r uint64) {
-	buf := make([]byte, 8)
+func randBytes(n int) []byte {
+	buf := make([]byte, n)
 	_, err := rand.Read(buf)
 	checkFatal(err)
-	for _, v := range buf {
-		r <<= 8
-		r |= uint64(v)
-	}
-	return
+	return buf
+}
+
+func randUint64() (r uint64) {
+	return binary.LittleEndian.Uint64(randBytes(8))
+}
+
+func randUint16() (r uint16) {
+	return binary.LittleEndian.Uint16(randBytes(2))
 }
 
 func GobEncode(items ...interface{}) []byte {

test/190_no_fastdp_2_test.sh

@@ -0,0 +1,27 @@
+#! /bin/bash
+
+. ./config.sh
+
+C1=10.2.1.46
+C2=10.2.1.47
+
+start_suite "WEAVE_NO_FASTDP operation"
+
+WEAVE_NO_FASTDP=1 weave_on $HOST1 launch
+WEAVE_NO_FASTDP=1 weave_on $HOST2 launch $HOST1
+
+start_container $HOST1 $C1/24 --name=c1
+start_container $HOST2 $C2/24 --name=c2
+
+# Without fastdp, ethwe should have an MTU of 65535
+assert_raises "exec_on $HOST1 c1 sh -c 'ip link show ethwe | grep \ mtu\ 65535\ >/dev/null'"
+
+# Pump some data over a TCP socket between the containers.  This
+# should cause PMTU discovery on the overlay network, and hence
+# exercise the sleeve code to generate an ICMP frag-needed.
+assert_raises "exec_on $HOST1 c1 sh -c 'nc -l -p 5555 >/tmp/foo'" &
+sleep 5
+assert_raises "exec_on $HOST2 c2 sh -c 'dd if=/dev/zero bs=10k count=1 | nc $C1 5555'"
+assert_raises "exec_on $HOST1 c1 sh -c 'test \$(stat -c %s /tmp/foo) -eq 10240'"
+
+end_suite

weave

@@ -107,6 +107,8 @@ exec_remote() {
         -e WEAVE_PASSWORD \
         -e WEAVE_PORT \
         -e WEAVE_CONTAINER_NAME \
+        -e WEAVE_MTU \
+        -e WEAVE_NO_FASTDP \
         -e DOCKER_BRIDGE \
         -e DOCKER_CLIENT_HOST="$DOCKER_CLIENT_HOST" \
         -e DOCKER_CLIENT_TLS_VERIFY="$DOCKER_CLIENT_TLS_VERIFY" \
@@ -263,7 +265,8 @@ DOCKER_BRIDGE=${DOCKER_BRIDGE:-docker0}
 CONTAINER_NAME=${WEAVE_CONTAINER_NAME:-weave}
 BRIDGE=weave
 CONTAINER_IFNAME=ethwe
-MTU=65535
+# ROUTER_HOSTNETNS_IFNAME is only used for fastdp with encryption
+ROUTER_HOSTNETNS_IFNAME=weave2
 PORT=${WEAVE_PORT:-6783}
 HTTP_PORT=6784
 PROXY_PORT=12375
@@ -366,33 +369,57 @@ random_mac() {
 ######################################################################
 
 create_bridge() {
+    BRIDGE_TYPE=
+
     if [ ! -d /sys/class/net/$BRIDGE ] ; then
-        ip link add name $BRIDGE type bridge
-        # Set a random MAC address on the bridge.  Current Linux
-        # kernels already do this when creating a bridge, but there
-        # are rumours it was not always so.
-        ip link set dev $BRIDGE address $(random_mac)
-        # Attempting to set the bridge MTU to a high value directly
-        # fails. Bridges take the lowest MTU of their interfaces. So
-        # instead we create a temporary interface with the desired
-        # MTU, attach that to the bridge, and then remove it again.
-        ip link add name v${CONTAINER_IFNAME}du mtu $MTU type dummy
-        ip link set dev v${CONTAINER_IFNAME}du master $BRIDGE
-        ip link del dev v${CONTAINER_IFNAME}du
-        # Drop traffic from Docker bridge to Weave; it can break subnet isolation
+        if [ -n "$WEAVE_NO_FASTDP" ] ; then
+            BRIDGE_TYPE=bridge
+        elif docker run --rm --privileged --net=host $IMAGE $COVERAGE_ARGS --create-datapath --datapath=$BRIDGE ; then
+            BRIDGE_TYPE=fastdp
+        elif [ $? = 17 ] ; then
+            # Exit status of 17 means the kernel doesn't have ODP
+            BRIDGE_TYPE=bridge
+        else
+            return 1
+        fi
+
+        init_bridge_$BRIDGE_TYPE
+
+        # Drop traffic from Docker bridge to Weave; it can break
+        # subnet isolation
         if [ "$DOCKER_BRIDGE" != "$BRIDGE" ] ; then
             # Note using -I to insert ahead of Docker's bridge rules
             run_iptables -t filter -I FORWARD -i $DOCKER_BRIDGE -o $BRIDGE -j DROP
         fi
+
         # Work around the situation where there are no rules allowing traffic
         # across our bridge. E.g. ufw
         add_iptables_rule filter FORWARD -i $BRIDGE -o $BRIDGE -j ACCEPT
+
         # create a chain for masquerading
         run_iptables -t nat -N WEAVE >/dev/null 2>&1 || true
         add_iptables_rule nat POSTROUTING -j WEAVE
+    else
+        # Detect whether fast datapath is in use on
+        # $BRIDGE. Unfortunately there's no simple way to positively
+        # check whether $BRIDGE is a ODP netdev, so we have to check
+        # whether it is a bridge instead.
+        if [ -d /sys/class/net/$BRIDGE/bridge ] ; then
+            BRIDGE_TYPE=bridge
+        else
+            BRIDGE_TYPE=fastdp
+        fi
+
+        # WEAVE_MTU may have been specified when the bridge was
+        # created (perhaps implicitly with WEAVE_NO_FASTDP).  So take
+        # the MTU from the bridge unless it is explicitly specified
+        # for this invocation.
+        MTU=${WEAVE_MTU:-$(cat /sys/class/net/$BRIDGE/mtu)}
     fi
 
-    [ "$1" != "--without-ethtool" ] && ethtool -K $BRIDGE tx off >/dev/null
+    if [ "$1" != "--without-ethtool" ] ; then
+        ethtool_tx_off_$BRIDGE_TYPE $BRIDGE
+    fi
 
     ip link set dev $BRIDGE up
 
@@ -401,11 +428,58 @@ create_bridge() {
     configure_arp_cache $BRIDGE
 }
 
+init_bridge_fastdp () {
+    # GCE has the lowest underlay network MTU we're likely to encounter on
+    # a local network, at 1460 bytes.  To get the overlay MTU from that we
+    # subtract 20 bytes for the outer IPv4 header, 8 bytes for the outer
+    # UDP header, 8 bytes for the vxlan header, and 14 bytes for the inner
+    # ethernet header.
+    MTU=${WEAVE_MTU:-1410}
+
+    # create_bridge already created the datapath netdev
+    ip link set dev $BRIDGE mtu $MTU
+}
+
+init_bridge_bridge () {
+    MTU=${WEAVE_MTU:-65535}
+
+    ip link add name $BRIDGE type bridge
+
+    # Set a random MAC address on the bridge.  Current Linux
+    # kernels already do this when creating a bridge, but there
+    # are rumours it was not always so.
+    ip link set dev $BRIDGE address $(random_mac)
+
+    # Attempting to set the bridge MTU to a high value directly
+    # fails. Bridges take the lowest MTU of their interfaces. So
+    # instead we create a temporary interface with the desired
+    # MTU, attach that to the bridge, and then remove it again.
+    ip link add name v${CONTAINER_IFNAME}du mtu $MTU type dummy
+    ip link set dev v${CONTAINER_IFNAME}du master $BRIDGE
+    ip link del dev v${CONTAINER_IFNAME}du
+}
+
+ethtool_tx_off_fastdp () {
+    true
+}
+
+ethtool_tx_off_bridge () {
+    ethtool -K $1 tx off >/dev/null
+}
+
 destroy_bridge() {
-    [ -d /sys/class/net/$BRIDGE ] && ip link del dev $BRIDGE
+    if [ -d /sys/class/net/$BRIDGE ] ; then
+        if [ -d /sys/class/net/$BRIDGE/bridge ] ; then
+            ip link del dev $BRIDGE
+        else
+            docker run --rm --privileged --net=host $IMAGE $COVERAGE_ARGS --delete-datapath --datapath=$BRIDGE
+        fi
+    fi
+
     if [ "$DOCKER_BRIDGE" != "$BRIDGE" ] ; then
         run_iptables -t filter -D FORWARD -i $DOCKER_BRIDGE -o $BRIDGE -j DROP 2>/dev/null || true
     fi
+
     run_iptables -t filter -D FORWARD -i $BRIDGE -o $BRIDGE -j ACCEPT 2>/dev/null || true
     run_iptables -t nat -F WEAVE >/dev/null 2>&1 || true
     run_iptables -t nat -D POSTROUTING -j WEAVE >/dev/null 2>&1 || true
@@ -474,17 +548,11 @@ netnsenter() {
     nsenter --net=$PROCFS/$CONTAINER_PID/ns/net "$@"
 }
 
+# connect_container_to_bridge <container inteface name>
 connect_container_to_bridge() {
-    if [ -h "$PROCFS/$CONTAINER_PID/ns/net" -a -h "/proc/$$/ns/net" -a "$(readlink $PROCFS/$CONTAINER_PID/ns/net)" = "$(readlink /proc/$$/ns/net)" ] ; then
-        echo "Container is running in the host network namespace, and therefore cannot be" >&2
-        echo "connected to weave. Perhaps the container was started with --net=host." >&2
-        return 1
-    fi
     ip link add name $LOCAL_IFNAME mtu $MTU type veth peer name $GUEST_IFNAME mtu $MTU || return 1
 
-    if ! ethtool -K $GUEST_IFNAME tx off >/dev/null ||
-        ! ip link set $LOCAL_IFNAME master $BRIDGE ||
-        ! ip link set $LOCAL_IFNAME up ||
+    if ! ethtool_tx_off_$BRIDGE_TYPE $GUEST_IFNAME ||
         ! ip link set $GUEST_IFNAME netns $PROCFS/$CONTAINER_PID/ns/net ; then
         # failed before we assigned the veth to the container's
         # namespace
@@ -492,12 +560,30 @@ connect_container_to_bridge() {
         return 1
     fi
 
-    if ! netnsenter ip link set $GUEST_IFNAME name $CONTAINER_IFNAME ||
-       ! configure_arp_cache $CONTAINER_IFNAME "netnsenter" ; then
+    # Versions of the 'ip' command differ in whether they create a
+    # veth pair in the 'up' or 'down' state.  Furthermore, moving one
+    # end into a netns normally sets them to 'down'.  But, moving an
+    # end into the default netns is a null operation, so they stay up,
+    # which leads to errors later on.  So to get a consistent result,
+    # we have to be explicit in setting them 'up' or 'down' as
+    # necessary.
+    if ! netnsenter ip link set $GUEST_IFNAME down ||
+       ! netnsenter ip link set $GUEST_IFNAME name $1 up ||
+       ! ip link set $LOCAL_IFNAME up ||
+       ! add_iface_$BRIDGE_TYPE $LOCAL_IFNAME ||
+       ! configure_arp_cache $1 "netnsenter" ; then
         return 1
     fi
 }
 
+add_iface_fastdp () {
+    docker run --rm --privileged --net=host $IMAGE $COVERAGE_ARGS --datapath=$BRIDGE --add-datapath-iface=$1
+}
+
+add_iface_bridge () {
+    ip link set $1 master $BRIDGE
+}
+
 ask_version() {
     if ! DOCKERIMAGE=$(docker inspect --format='{{.Image}}' $1 2>/dev/null) ; then
             if ! DOCKERIMAGE=$(docker inspect --format='{{.Id}}' $2 2>/dev/null) ; then
@@ -507,23 +593,65 @@ ask_version() {
     [ -n "$DOCKERIMAGE" ] && docker run --rm -e WEAVE_CIDR=none $3 $DOCKERIMAGE $COVERAGE_ARGS --version
 }
 
+router_opts_fastdp () {
+    if [ -z "$WEAVE_PASSWORD" ] ; then
+        echo "--datapath $BRIDGE"
+    else
+        # When using encryption, we still do bridging on the ODP
+        # datapath, because you can 'weave launch' without encryption
+        # and then later restart the router with encryption, or vice
+        # versa.  Encryption disables the use of the fastdp Overlay,
+        # but the router could still use the fastdp Bridge to receive
+        # packets.  However, pcap has better performance when sniffing
+        # every packet.  So we pass --iface to use the pcap Bridge.
+        #
+        # Why don't we simply pass "--iface $BRIDGE".  We could,
+        # except for the fact that NetworkManager likes to down the
+        # odp $BRIDGE netdev (at least under ubuntu), and you can only
+        # use pcap on an interface that is up.  We avoid that by use
+        # pcap via a veth pair (NetworkManager leaves them alone).
+        # Having a netdev in the host netns called "ethwe" might
+        # surprise people, so it is called $ROUTER_HOSTNETNS_IFNAME
+        # instead.
+        echo "--datapath $BRIDGE --iface $ROUTER_HOSTNETNS_IFNAME"
+    fi
+}
+
+router_opts_bridge () {
+    echo "--iface $CONTAINER_IFNAME"
+}
+
 ######################################################################
 # functions invoked through with_container_netns
 ######################################################################
 
-launch() {
+setup_router_iface_fastdp() {
+    if [ -n "$WEAVE_PASSWORD" ] ; then
+        # See router_opts_fastdp
+        connect_container_to_bridge $ROUTER_HOSTNETNS_IFNAME &&
+            ip link set $ROUTER_HOSTNETNS_IFNAME up
+    fi
+}
+
+setup_router_iface_bridge() {
     if ! netnsenter ip link show eth0 >/dev/null ; then
         echo "Perhaps you are running the docker daemon with container networking disabled (-b=none)." >&2
         return 1
     fi
-    connect_container_to_bridge &&
+    connect_container_to_bridge $CONTAINER_IFNAME &&
         netnsenter ethtool -K eth0 tx off >/dev/null &&
         netnsenter ip link set $CONTAINER_IFNAME up
 }
 
 attach() {
+    if [ -h "$PROCFS/$CONTAINER_PID/ns/net" -a -h "/proc/$$/ns/net" -a "$(readlink $PROCFS/$CONTAINER_PID/ns/net)" = "$(readlink /proc/$$/ns/net)" ] ; then
+        echo "Container is running in the host network namespace, and therefore cannot be" >&2
+        echo "connected to weave. Perhaps the container was started with --net=host." >&2
+        return 1
+    fi
+
     if ! netnsenter ip link show $CONTAINER_IFNAME >/dev/null 2>&1 ; then
-        connect_container_to_bridge || return 1
+        connect_container_to_bridge $CONTAINER_IFNAME || return 1
     fi
 
     NEW_ADDRS=
@@ -645,8 +773,8 @@ container_ip() {
     fi
     case "$status" in
         "true ")
-            echo "$1 container has no IP address; is Docker networking enabled?" >&2
-            return 1
+            #echo "$1 container has no IP address; is Docker networking enabled?" >&2
+            CONTAINER_IP=127.0.0.1
             ;;
         true*)
             CONTAINER_IP="${status#true }"
@@ -1127,10 +1255,17 @@ launch_router() {
         shift 1
     fi
     CONTAINER_PORT=$PORT
-    ARGS=""
+    ARGS=
     IPRANGE=
     IPRANGE_SPECIFIED=
+
+    # DNS_PORT_MAPPING is only used if the router runs in a container
+    # netns.  DNS_ROUTER_OPTS is only used if the router runs in the
+    # host netns.
     DNS_PORT_MAPPING="-p $DOCKER_BRIDGE_IP:53:53/udp -p $DOCKER_BRIDGE_IP:53:53/tcp"
+    DNS_ROUTER_OPTS="--dns-listen-address $DOCKER_BRIDGE_IP:53"
+    NO_DNS_OPT=
+
     while [ $# -gt 0 ] ; do
         case "$1" in
             -password|--password)
@@ -1163,6 +1298,8 @@ launch_router() {
                 ;;
             --no-dns)
                 DNS_PORT_MAPPING=
+                DNS_ROUTER_OPTS=
+                NO_DNS_OPT="--no-dns"
                 ARGS="$ARGS $1"
                 ;;
             *)
@@ -1185,16 +1322,28 @@ launch_router() {
                 echo "Unless this is deliberate, you must pick another range and set it on all hosts." >&2
         fi
     fi
+
+    if [ "$BRIDGE_TYPE" = fastdp ] ; then
+        NETHOST_OPT="--net=host"
+    fi
+
     # Set WEAVE_DOCKER_ARGS in the environment in order to supply
     # additional parameters, such as resource limits, to docker
     # when launching the weave container.
     ROUTER_CONTAINER=$(docker run --privileged -d --name=$CONTAINER_NAME \
-        -p $PORT:$CONTAINER_PORT/tcp -p $PORT:$CONTAINER_PORT/udp $DNS_PORT_MAPPING \
         -v /var/run/docker.sock:/var/run/docker.sock \
+        -p $PORT:$CONTAINER_PORT/tcp -p $PORT:$CONTAINER_PORT/udp \
+        ${NETHOST_OPT:-$DNS_PORT_MAPPING} \
         -e WEAVE_PASSWORD \
         -e WEAVE_CIDR=none \
-        $WEAVE_DOCKER_ARGS $IMAGE $COVERAGE_ARGS --iface $CONTAINER_IFNAME --port $CONTAINER_PORT --name "$PEERNAME" --nickname "$(hostname)" --ipalloc-range "$IPRANGE" --dns-effective-listen-address "$DOCKER_BRIDGE_IP" --docker-api "unix:///var/run/docker.sock" "$@")
-    with_container_netns_or_die $ROUTER_CONTAINER launch
+        $WEAVE_DOCKER_ARGS $IMAGE $COVERAGE_ARGS \
+        --port $CONTAINER_PORT --name "$PEERNAME" --nickname "$(hostname)" \
+        $(router_opts_$BRIDGE_TYPE) \
+        --ipalloc-range "$IPRANGE" \
+        --dns-effective-listen-address $DOCKER_BRIDGE_IP \
+        ${NETHOST_OPT:+$DNS_ROUTER_OPTS} $NO_DNS_OPT \
+        --docker-api "unix:///var/run/docker.sock" "$@")
+    with_container_netns_or_die $ROUTER_CONTAINER setup_router_iface_$BRIDGE_TYPE
     err_msg=$(death_msg $CONTAINER_NAME)
     container_ip $CONTAINER_NAME "$err_msg" "$err_msg" || return 1
     wait_for_status $CONTAINER_NAME http_call $CONTAINER_IP:$HTTP_PORT
@@ -1202,7 +1351,7 @@ launch_router() {
         # Tell the newly-started weave IP allocator about existing weave IPs
         with_container_addresses ipam_reclaim weave:expose $(docker ps -q --no-trunc)
     fi
-    if [ -n "$DNS_PORT_MAPPING" ] ; then
+    if [ -z "$NO_DNS_OPT" ] ; then
         # Tell the newly-started weaveDNS about existing weave IPs
         for CONTAINER in $(docker ps -q --no-trunc) ; do
             if CONTAINER_IPS=$(with_container_netns $CONTAINER container_weave_addrs 2>&1 | sed -n -e 's/inet \([^/]*\)\/\(.*\)/\1/p') ; then
@@ -1232,6 +1381,8 @@ launch_proxy() {
         -e PROCFS=/hostproc \
         -e WEAVE_CIDR=none \
         -e DOCKER_BRIDGE \
+        -e WEAVE_DEBUG \
+        -e COVERAGE \
         --entrypoint=/home/weave/weaveproxy \
         $WEAVEPROXY_DOCKER_ARGS $EXEC_IMAGE $COVERAGE_ARGS $PROXY_ARGS)
     wait_for_status $PROXY_CONTAINER_NAME http_call_unix $PROXY_CONTAINER_NAME status.sock