OverlaySwitch

router/connection.go

@@ -199,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
@@ -248,7 +249,7 @@ 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,
@@ -256,6 +257,8 @@ func (conn *LocalConnection) makeFeatures() map[string]string {
 		"UID":             fmt.Sprint(conn.local.UID),
 		"ConnID":          fmt.Sprint(conn.uid),
 	}
+	conn.Router.Overlay.AddFeaturesTo(features)
+	return features
 }
 
 type features map[string]string
@@ -392,8 +395,8 @@ func (conn *LocalConnection) forwarderCrypto() *OverlayCrypto {
 	}
 }
 
-func (conn *LocalConnection) sendOverlayControlMessage(tag ProtocolTag, msg []byte) error {
-	return conn.sendProtocolMsg(ProtocolMsg{tag, msg})
+func (conn *LocalConnection) sendOverlayControlMessage(tag byte, msg []byte) error {
+	return conn.sendProtocolMsg(ProtocolMsg{ProtocolTag(tag), msg})
 }
 
 // Helpers
@@ -430,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/overlay.go

@@ -18,6 +18,9 @@ type Overlay interface {
 
 	// 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 {
@@ -41,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
@@ -81,7 +87,7 @@ type OverlayForwarder interface {
 	// Handle a message from the peer.  'tag' exists for
 	// compatibility, and should always be
 	// ProtocolOverlayControlMessage for non-sleeve overlays.
-	ControlMessage(tag ProtocolTag, msg []byte)
+	ControlMessage(tag byte, msg []byte)
 }
 
 type NullOverlay struct{}
@@ -111,6 +117,9 @@ func (NullOverlay) ErrorChannel() <-chan error {
 	return nil
 }
 
+func (NullOverlay) AddFeaturesTo(map[string]string) {
+}
+
 func (NullOverlay) Forward(ForwardPacketKey) FlowOp {
 	return nil
 }
@@ -118,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/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/sleeve.go

@@ -133,6 +133,10 @@ 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()
@@ -281,7 +285,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
@@ -342,7 +346,7 @@ type specialFrame struct {
 
 // A control message
 type controlMessage struct {
-	tag ProtocolTag
+	tag byte
 	msg []byte
 }
 
@@ -586,7 +590,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: