Merge pull request #1522 from dpw/fdp-compat

Fast datapath compatibility with 1.1

prog/weaver/main.go

@@ -197,7 +197,9 @@ func main() {
 	if fastDPOverlay != nil {
 		overlays.Add("fastdp", fastDPOverlay)
 	}
-	overlays.Add("sleeve", weave.NewSleeveOverlay(config.Port))
+	sleeve := weave.NewSleeveOverlay(config.Port)
+	overlays.Add("sleeve", sleeve)
+	overlays.SetCompatOverlay(sleeve)
 	config.Overlay = overlays
 
 	if routerName == "" {

router/connection.go

@@ -277,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", "ShortID", "UID", "ConnID"}); err != nil {
+	if err := features.MustHave([]string{"PeerNameFlavour", "Name", "NickName", "UID", "ConnID"}); err != nil {
 		return nil, err
 	}
 
@@ -293,10 +293,15 @@ 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
+	var shortID uint64
+	var hasShortID bool
+	if shortIDStr, present := features["ShortID"]; present {
+		hasShortID = true
+		shortID, err = strconv.ParseUint(shortIDStr, 10,
+			PeerShortIDBits)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	uid, err := ParsePeerUID(features.Get("UID"))
@@ -310,7 +315,9 @@ func (conn *LocalConnection) parseFeatures(features features) (*Peer, error) {
 	}
 
 	conn.uid ^= remoteConnID
-	return NewPeer(name, nickName, uid, 0, PeerShortID(shortID)), nil
+	peer := NewPeer(name, nickName, uid, 0, PeerShortID(shortID))
+	peer.HasShortID = hasShortID
+	return peer, nil
 }
 
 func (conn *LocalConnection) registerRemote(remote *Peer, acceptNewPeer bool) error {

router/fastdp.go

@@ -443,8 +443,10 @@ func (fastdp *FastDatapath) getVxlanVportID(udpPort int) (odp.VportID, error) {
 		pk := flowKeysToPacketKey(fks)
 		var zeroMAC MAC
 		if pk.SrcMAC == zeroMAC && pk.DstMAC == zeroMAC {
-			return vxlanSpecialPacketFlowOp{fastdp, srcPeer,
-				&net.UDPAddr{
+			return vxlanSpecialPacketFlowOp{
+				fastdp:  fastdp,
+				srcPeer: srcPeer,
+				sender: &net.UDPAddr{
 					IP:   net.IP(tunKey.Ipv4Src[:]),
 					Port: udpPort,
 				},
@@ -477,6 +479,7 @@ func (fastdp *FastDatapath) extractPeers(tunnelID [8]byte) (*Peer, *Peer) {
 }
 
 type vxlanSpecialPacketFlowOp struct {
+	NonDiscardingFlowOp
 	fastdp  *FastDatapath
 	srcPeer *Peer
 	sender  *net.UDPAddr
@@ -751,6 +754,10 @@ func (fwd *fastDatapathForwarder) handleHeartbeatAck() {
 }
 
 func (fwd *fastDatapathForwarder) Forward(key ForwardPacketKey) FlowOp {
+	if !key.SrcPeer.HasShortID || !key.DstPeer.HasShortID {
+		return nil
+	}
+
 	fwd.lock.RLock()
 	defer fwd.lock.RUnlock()
 
@@ -765,7 +772,7 @@ func (fwd *fastDatapathForwarder) Forward(key ForwardPacketKey) FlowOp {
 	remoteIP, err := ipv4Bytes(fwd.remoteAddr.IP)
 	if err != nil {
 		log.Error(err)
-		return nil
+		return DiscardingFlowOp{}
 	}
 
 	var sta odp.SetTunnelAction
@@ -1134,6 +1141,7 @@ func (fastdp *FastDatapath) takeDecoder(lock *fastDatapathLock) *EthernetDecoder
 }
 
 type odpActionsFlowOp struct {
+	NonDiscardingFlowOp
 	fastdp  *FastDatapath
 	actions []odp.Action
 }
@@ -1157,22 +1165,15 @@ func (fop odpActionsFlowOp) Process(frame []byte, dec *EthernetDecoder,
 	checkWarn(fastdp.dp.Execute(frame, nil, fop.actions))
 }
 
-type nopFlowOp struct{}
-
-func (nopFlowOp) Process([]byte, *EthernetDecoder, bool) {
-	// A nopFlowOp just provides a hint about flow creation, it
-	// doesn't send anything
-}
-
 // A vetoFlowCreationFlowOp flags that no flow should be created
 type vetoFlowCreationFlowOp struct {
-	nopFlowOp
+	DiscardingFlowOp
 }
 
 // A odpFlowKeyFlowOp adds a FlowKey to the resulting flow
 type odpFlowKeyFlowOp struct {
+	DiscardingFlowOp
 	key odp.FlowKey
-	nopFlowOp
 }
 
 func odpFlowKey(key odp.FlowKey) FlowOp {

router/flow.go

@@ -27,6 +27,24 @@ type FlowOp interface {
 	// The broadcast parameter is a hint whether the packet is
 	// being broadcast.
 	Process(frame []byte, dec *EthernetDecoder, broadcast bool)
+
+	// Does the FlowOp discard the packet?
+	Discards() bool
+}
+
+type DiscardingFlowOp struct{}
+
+func (DiscardingFlowOp) Process([]byte, *EthernetDecoder, bool) {
+}
+
+func (DiscardingFlowOp) Discards() bool {
+	return true
+}
+
+type NonDiscardingFlowOp struct{}
+
+func (NonDiscardingFlowOp) Discards() bool {
+	return false
 }
 
 type MultiFlowOp struct {
@@ -36,16 +54,14 @@ type MultiFlowOp struct {
 
 func NewMultiFlowOp(broadcast bool, ops ...FlowOp) *MultiFlowOp {
 	mfop := &MultiFlowOp{broadcast: broadcast}
-	mfop.Add(ops...)
+	for _, op := range ops {
+		mfop.Add(op)
+	}
 	return mfop
 }
 
-func (mfop *MultiFlowOp) Add(ops ...FlowOp) {
-	for _, op := range ops {
-		if op != nil {
-			mfop.ops = append(mfop.ops, op)
-		}
-	}
+func (mfop *MultiFlowOp) Add(op FlowOp) {
+	mfop.ops = append(mfop.ops, op)
 }
 
 func (mfop *MultiFlowOp) Process(frame []byte, dec *EthernetDecoder,
@@ -55,6 +71,16 @@ func (mfop *MultiFlowOp) Process(frame []byte, dec *EthernetDecoder,
 	}
 }
 
+func (mfop *MultiFlowOp) Discards() bool {
+	for _, op := range mfop.ops {
+		if !op.Discards() {
+			return false
+		}
+	}
+
+	return true
+}
+
 func FlattenFlowOp(fop FlowOp) []FlowOp {
 	return collectFlowOps(nil, fop)
 }

router/overlay.go

@@ -68,7 +68,8 @@ type OverlayCrypto struct {
 type OverlayForwarder interface {
 	// Forward a packet across the connection.  May be called as
 	// soon as the forwarder is created, in particular before
-	// Confirm().
+	// Confirm().  The return value nil means the key could not be
+	// handled by this forwarder.
 	Forward(ForwardPacketKey) FlowOp
 
 	// Confirm that the connection is really wanted, and so the
@@ -127,7 +128,7 @@ func (NullOverlay) AddFeaturesTo(map[string]string) {
 }
 
 func (NullOverlay) Forward(ForwardPacketKey) FlowOp {
-	return nil
+	return DiscardingFlowOp{}
 }
 
 func (NullOverlay) Stop() {

router/overlay_switch.go

@@ -13,8 +13,9 @@ import (
 // uses the best one that seems to be working.
 
 type OverlaySwitch struct {
-	overlays     map[string]Overlay
-	overlayNames []string
+	overlays      map[string]Overlay
+	overlayNames  []string
+	compatOverlay Overlay
 }
 
 func NewOverlaySwitch() *OverlaySwitch {
@@ -31,6 +32,10 @@ func (osw *OverlaySwitch) Add(name string, overlay Overlay) {
 	osw.overlayNames = append(osw.overlayNames, name)
 }
 
+func (osw *OverlaySwitch) SetCompatOverlay(overlay Overlay) {
+	osw.compatOverlay = overlay
+}
+
 func (osw *OverlaySwitch) AddFeaturesTo(features map[string]string) {
 	features["Overlays"] = strings.Join(osw.overlayNames, " ")
 }
@@ -119,8 +124,8 @@ type overlaySwitchForwarder struct {
 
 	lock sync.Mutex
 
-	// the forwarder to send on
-	best OverlayForwarder
+	// the index of the forwarder to send on
+	best int
 
 	// the subsidiary forwarders
 	forwarders []subForwarder
@@ -159,6 +164,10 @@ type subForwarderEvent struct {
 
 func (osw *OverlaySwitch) MakeForwarder(
 	params ForwarderParams) (OverlayForwarder, error) {
+	if _, present := params.Features["Overlays"]; !present && osw.compatOverlay != nil {
+		return osw.compatOverlay.MakeForwarder(params)
+	}
+
 	overlays, err := osw.commonOverlays(params)
 	if err != nil {
 		return nil, err
@@ -174,6 +183,7 @@ func (osw *OverlaySwitch) MakeForwarder(
 	fwd := &overlaySwitchForwarder{
 		remotePeer: params.RemotePeer,
 
+		best:       -1,
 		forwarders: make([]subForwarder, len(overlays)),
 		stopChan:   stopChan,
 
@@ -276,9 +286,6 @@ func (fwd *overlaySwitchForwarder) established(index int) {
 
 	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)
@@ -314,7 +321,8 @@ func (fwd *overlaySwitchForwarder) stopFrom(index int) {
 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
+	bestEstablished := -1
+	bestWorking := -1
 
 	for i := range fwd.forwarders {
 		subFwd := &fwd.forwarders[i]
@@ -322,18 +330,18 @@ func (fwd *overlaySwitchForwarder) chooseBest() {
 			continue
 		}
 
-		if bestWorking == nil {
-			bestWorking = subFwd
+		if bestWorking < 0 {
+			bestWorking = i
 		}
 
-		if bestEstablished == nil && subFwd.established {
-			bestEstablished = subFwd
+		if bestEstablished < 0 && subFwd.established {
+			bestEstablished = i
 		}
 	}
 
 	best := bestEstablished
-	if best == nil {
-		if bestWorking == nil {
+	if best < 0 {
+		if bestWorking < 0 {
 			select {
 			case fwd.errorChan <- fmt.Errorf("no working forwarders to %s", fwd.remotePeer):
 			default:
@@ -345,9 +353,10 @@ func (fwd *overlaySwitchForwarder) chooseBest() {
 		best = bestWorking
 	}
 
-	if fwd.best != best.fwd {
-		fwd.best = best.fwd
-		log.Info(fwd.logPrefix(), "using ", best.overlayName)
+	if fwd.best != best {
+		fwd.best = best
+		log.Info(fwd.logPrefix(),
+			"using ", fwd.forwarders[best].overlayName)
 	}
 }
 
@@ -369,14 +378,23 @@ func (fwd *overlaySwitchForwarder) Confirm() {
 
 func (fwd *overlaySwitchForwarder) Forward(pk ForwardPacketKey) FlowOp {
 	fwd.lock.Lock()
-	best := fwd.best
-	fwd.lock.Unlock()
 
-	if best == nil {
-		return nil
+	if fwd.best >= 0 {
+		for i := fwd.best; i < len(fwd.forwarders); i++ {
+			best := fwd.forwarders[i].fwd
+			if best != nil {
+				fwd.lock.Unlock()
+				if op := best.Forward(pk); op != nil {
+					return op
+				}
+				fwd.lock.Lock()
+			}
+		}
 	}
 
-	return best.Forward(pk)
+	fwd.lock.Unlock()
+
+	return DiscardingFlowOp{}
 }
 
 func (fwd *overlaySwitchForwarder) EstablishedChannel() <-chan struct{} {
@@ -403,13 +421,17 @@ func (fwd *overlaySwitchForwarder) ControlMessage(tag byte, msg []byte) {
 }
 
 func (fwd *overlaySwitchForwarder) DisplayName() string {
+	var best OverlayForwarder
+
 	fwd.lock.Lock()
-	best := fwd.best
+	if fwd.best > 0 {
+		best = fwd.forwarders[fwd.best].fwd
+	}
 	fwd.lock.Unlock()
 
-	if best == nil {
-		return "none"
+	if best != nil {
+		return best.DisplayName()
 	}
 
-	return best.DisplayName()
+	return "none"
 }

router/pcap.go

@@ -9,6 +9,8 @@ import (
 )
 
 type Pcap struct {
+	NonDiscardingFlowOp
+
 	iface *net.Interface
 	bufSz int
 
@@ -131,7 +133,7 @@ func (p *Pcap) sniff(readHandle *pcap.Handle, consumer BridgeConsumer) {
 			continue
 		}
 
-		if fop := consumer(dec.PacketKey()); fop != nil {
+		if fop := consumer(dec.PacketKey()); !fop.Discards() {
 			// We are handing over the frame to
 			// forwarders, so we need to make a copy of it
 			// in order to prevent the next capture from

router/peer.go

@@ -30,11 +30,12 @@ func randomPeerShortID() PeerShortID {
 }
 
 type PeerSummary struct {
-	NameByte []byte
-	NickName string
-	UID      PeerUID
-	Version  uint64
-	ShortID  PeerShortID
+	NameByte   []byte
+	NickName   string
+	UID        PeerUID
+	Version    uint64
+	ShortID    PeerShortID
+	HasShortID bool
 }
 
 type Peer struct {
@@ -56,11 +57,12 @@ func NewPeerFromSummary(summary PeerSummary) *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,
+		NameByte:   name.Bin(),
+		NickName:   nickName,
+		UID:        uid,
+		Version:    version,
+		ShortID:    shortID,
+		HasShortID: true,
 	})
 }
 

router/peers.go

@@ -114,6 +114,10 @@ func (peers *Peers) unlockAndNotify(pending *PeersPendingNotifications) {
 }
 
 func (peers *Peers) addByShortID(peer *Peer, pending *PeersPendingNotifications) {
+	if !peer.HasShortID {
+		return
+	}
+
 	entry, ok := peers.byShortID[peer.ShortID]
 	if !ok {
 		entry = ShortIDPeers{peer: peer}
@@ -146,6 +150,10 @@ func (peers *Peers) addByShortID(peer *Peer, pending *PeersPendingNotifications)
 }
 
 func (peers *Peers) deleteByShortID(peer *Peer, pending *PeersPendingNotifications) {
+	if !peer.HasShortID {
+		return
+	}
+
 	entry := peers.byShortID[peer.ShortID]
 	var otherIndex int
 

router/router.go

@@ -114,12 +114,12 @@ func (router *Router) handleCapturedPacket(key PacketKey) FlowOp {
 		// we are seeing a frame we injected ourself.  That
 		// shouldn't happen, but discard it just in case.
 		log.Errorln("Captured frame from MAC (", srcMac, ") associated with another peer", conflictPeer)
-		return nil
+		return DiscardingFlowOp{}
 	}
 
 	// Discard STP broadcasts
 	if key.DstMAC == [...]byte{0x01, 0x80, 0xC2, 0x00, 0x00, 0x00} {
-		return nil
+		return DiscardingFlowOp{}
 	}
 
 	dstMac := net.HardwareAddr(key.DstMAC[:])
@@ -130,7 +130,7 @@ func (router *Router) handleCapturedPacket(key PacketKey) FlowOp {
 		// it's likely to be broadcasting the packet to all
 		// ports.  So if it happens, just drop the packet to
 		// avoid warnings if we try to forward it.
-		return nil
+		return DiscardingFlowOp{}
 	case nil:
 		// If we don't know which peer corresponds to the dest
 		// MAC, broadcast it.
@@ -223,14 +223,14 @@ func (router *Router) relay(key ForwardPacketKey) FlowOp {
 		// Not necessarily an error as there could be a race with the
 		// dst disappearing whilst the frame is in flight
 		log.Println("Received packet for unknown destination:", key.DstPeer)
-		return nil
+		return DiscardingFlowOp{}
 	}
 
 	conn, found := router.Ourself.ConnectionTo(relayPeerName)
 	if !found {
 		// Again, could just be a race, not necessarily an error
 		log.Println("Unable to find connection to relay peer", relayPeerName)
-		return nil
+		return DiscardingFlowOp{}
 	}
 
 	return conn.(*LocalConnection).Forward(key)
@@ -239,7 +239,7 @@ func (router *Router) relay(key ForwardPacketKey) FlowOp {
 func (router *Router) relayBroadcast(srcPeer *Peer, key PacketKey) FlowOp {
 	nextHops := router.Routes.Broadcast(srcPeer.Name)
 	if len(nextHops) == 0 {
-		return nil
+		return DiscardingFlowOp{}
 	}
 
 	op := NewMultiFlowOp(true)

router/sleeve.go

@@ -427,12 +427,13 @@ func (fwd *sleeveForwarder) ErrorChannel() <-chan error {
 }
 
 type curriedForward struct {
+	NonDiscardingFlowOp
 	fwd *sleeveForwarder
 	key ForwardPacketKey
 }
 
 func (fwd *sleeveForwarder) Forward(key ForwardPacketKey) FlowOp {
-	return curriedForward{fwd, key}
+	return curriedForward{fwd: fwd, key: key}
 }
 
 func (f curriedForward) Process(frame []byte, dec *EthernetDecoder,