From f56457271da73bed3971df056e63e4809a85b601 Mon Sep 17 00:00:00 2001 From: David Wragg Date: Wed, 9 Sep 2015 11:55:09 +0100 Subject: [PATCH] OverlaySwitch --- router/connection.go | 11 +- router/overlay.go | 15 +- router/overlay_switch.go | 395 +++++++++++++++++++++++++++++++++++++++ router/protocol.go | 2 +- router/sleeve.go | 10 +- 5 files changed, 422 insertions(+), 11 deletions(-) create mode 100644 router/overlay_switch.go diff --git a/router/connection.go b/router/connection.go index 1090aa89..926d1392 100644 --- a/router/connection.go +++ b/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: diff --git a/router/overlay.go b/router/overlay.go index 9da023be..e8b44ab1 100644 --- a/router/overlay.go +++ b/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) { } diff --git a/router/overlay_switch.go b/router/overlay_switch.go new file mode 100644 index 00000000..b2ae6af3 --- /dev/null +++ b/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:]) + } +} diff --git a/router/protocol.go b/router/protocol.go index f1959373..add17a17 100644 --- a/router/protocol.go +++ b/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 diff --git a/router/sleeve.go b/router/sleeve.go index ed08d44e..7ed39f34 100644 --- a/router/sleeve.go +++ b/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: