From c2ad4c56332b22ca33f7504ee2d4f2d3aedbb2ad Mon Sep 17 00:00:00 2001 From: David Wragg Date: Mon, 3 Aug 2015 16:13:54 +0100 Subject: [PATCH] Flows For fast datapath, the router core will only be invoked upon datapath misses. And it needs to yield routing decisions in a form that can be used to configure the in-kernel datapath. So here we make the router core yield FlowOps, which represent operations which actually do something with a packet. The router core obtains these in turn from the Bridge or the Overlay, depending on what it wants to do with a packet. For fast datapath, it's critical that the router makes decisions based only on the required packet header information (as the same header information will be used when configuring the ODP flow rules). So we pass only that information into the router core, in the form of PacketKeys. The overall effect is to curry the router core: Rather than taking a packet and doing something with it, it takes packet header information and returns a function that does something with a packet. --- prog/weaver/main.go | 38 ++++++------ router/bridge.go | 10 ++-- router/connection.go | 4 +- router/ethernet_decoder.go | 14 ++--- router/flow.go | 50 ++++++++++++++++ router/local_peer.go | 42 ++++++++----- router/mac_cache.go | 6 +- router/overlay.go | 14 ++--- router/pcap.go | 24 +++++++- router/router.go | 117 +++++++++++++++++++++---------------- router/sleeve.go | 35 +++++++++-- 11 files changed, 235 insertions(+), 119 deletions(-) create mode 100644 router/flow.go diff --git a/prog/weaver/main.go b/prog/weaver/main.go index e7ec13ff..b8bd3326 100644 --- a/prog/weaver/main.go +++ b/prog/weaver/main.go @@ -1,7 +1,6 @@ package main import ( - "crypto/sha256" "fmt" "net" "net/http" @@ -164,9 +163,14 @@ func main() { defer profile.Start(&p).Stop() } - config.Overlay = weave.NewSleeveOverlay(config.Port) - config.LogFrame = logFrameFunc(pktdebug) config.PeerDiscovery = !noDiscovery + config.Overlay = weave.NewSleeveOverlay(config.Port) + + if pktdebug { + config.PacketLogging = packetLogging{} + } else { + config.PacketLogging = nopPacketLogging{} + } router := weave.NewRouter(config, name, nickName) Log.Println("Our name is", router.Ourself) @@ -255,24 +259,22 @@ func canonicalName(f *mflag.Flag) string { return "" } -func logFrameFunc(debug bool) weave.LogFrameFunc { - if !debug { - return func(prefix string, frame []byte, dec *weave.EthernetDecoder) {} - } - return func(prefix string, frame []byte, dec *weave.EthernetDecoder) { - h := fmt.Sprintf("%x", sha256.Sum256(frame)) - parts := []interface{}{prefix, len(frame), "bytes (", h, ")"} +type packetLogging struct{} - if dec != nil { - parts = append(parts, dec.Eth.SrcMAC, "->", dec.Eth.DstMAC) +func (packetLogging) LogPacket(msg string, key weave.PacketKey) { + Log.Println(msg, key.SrcMAC, "->", key.DstMAC) +} - if dec.DF() { - parts = append(parts, "(DF)") - } - } +func (packetLogging) LogForwardPacket(msg string, key weave.ForwardPacketKey) { + Log.Println(msg, key.SrcPeer, key.SrcMAC, "->", key.DstPeer, key.DstMAC) +} - Log.Println(parts...) - } +type nopPacketLogging struct{} + +func (nopPacketLogging) LogPacket(string, weave.PacketKey) { +} + +func (nopPacketLogging) LogForwardPacket(string, weave.ForwardPacketKey) { } func parseAndCheckCIDR(cidrStr string) address.CIDR { diff --git a/router/bridge.go b/router/bridge.go index 9e069177..603d8687 100644 --- a/router/bridge.go +++ b/router/bridge.go @@ -3,7 +3,7 @@ package router // Interface to packet handling on the local virtual bridge type Bridge interface { // Inject a packet to be delivered locally - InjectPacket([]byte) error + InjectPacket(PacketKey) FlowOp // Start consuming packets from the bridge StartConsumingPackets(BridgeConsumer) error @@ -12,14 +12,12 @@ type Bridge interface { Stats() map[string]int } -// A function that accepts locally captured packets. The ethernet -// decoder is specific to this thread, and will already have been used -// to to decode the packet data. -type BridgeConsumer func([]byte, *EthernetDecoder) +// A function that determines how to handle locally captured packets. +type BridgeConsumer func(PacketKey) FlowOp type NullBridge struct{} -func (NullBridge) InjectPacket([]byte) error { +func (NullBridge) InjectPacket(PacketKey) FlowOp { return nil } diff --git a/router/connection.go b/router/connection.go index 8a844cb7..e3967ec6 100644 --- a/router/connection.go +++ b/router/connection.go @@ -446,8 +446,8 @@ func (conn *LocalConnection) extendReadDeadline() { conn.TCPConn.SetReadDeadline(time.Now().Add(TCPHeartbeat * 2)) } -func (conn *LocalConnection) Forward(srcPeer, dstPeer *Peer, frame []byte, dec *EthernetDecoder, broadcast bool) { - conn.forwarder.Forward(srcPeer, dstPeer, frame, dec, broadcast) +func (conn *LocalConnection) Forward(key ForwardPacketKey) FlowOp { + return conn.forwarder.Forward(key) } func tickerChan(ticker *time.Ticker) <-chan time.Time { diff --git a/router/ethernet_decoder.go b/router/ethernet_decoder.go index ebc9b252..2eb1a429 100644 --- a/router/ethernet_decoder.go +++ b/router/ethernet_decoder.go @@ -29,6 +29,12 @@ func (dec *EthernetDecoder) DecodeLayers(data []byte) { dec.parser.DecodeLayers(data, &dec.decoded) } +func (dec *EthernetDecoder) PacketKey() (key PacketKey) { + copy(key.SrcMAC[:], dec.Eth.SrcMAC) + copy(key.DstMAC[:], dec.Eth.DstMAC) + return +} + func (dec *EthernetDecoder) makeICMPFragNeeded(mtu int) ([]byte, error) { buf := gopacket.NewSerializeBuffer() opts := gopacket.SerializeOptions{ @@ -65,15 +71,9 @@ func (dec *EthernetDecoder) makeICMPFragNeeded(mtu int) ([]byte, error) { } var ( - // see http://en.wikipedia.org/wiki/Multicast_address#Ethernet - stpMACPrefix = []byte{0x01, 0x80, 0xC2, 0x00, 0x00} - zeroMAC, _ = net.ParseMAC("00:00:00:00:00:00") + zeroMAC, _ = net.ParseMAC("00:00:00:00:00:00") ) -func (dec *EthernetDecoder) DropFrame() bool { - return bytes.Equal(stpMACPrefix, dec.Eth.DstMAC[:len(stpMACPrefix)]) -} - func (dec *EthernetDecoder) IsSpecial() bool { return dec.Eth.Length == 0 && dec.Eth.EthernetType == layers.EthernetTypeLLC && bytes.Equal(zeroMAC, dec.Eth.SrcMAC) && bytes.Equal(zeroMAC, dec.Eth.DstMAC) diff --git a/router/flow.go b/router/flow.go new file mode 100644 index 00000000..2a7ac039 --- /dev/null +++ b/router/flow.go @@ -0,0 +1,50 @@ +package router + +import "net" + +// Just enough flow machinery for the weave router + +type MAC [6]byte + +func (mac MAC) String() string { + return net.HardwareAddr(mac[:]).String() +} + +type PacketKey struct { + SrcMAC MAC + DstMAC MAC +} + +type ForwardPacketKey struct { + SrcPeer *Peer + DstPeer *Peer + PacketKey +} + +type FlowOp interface { + // The caller must supply an EthernetDecoder specific to this + // thread, which has already been used to decode the frame. + // The broadcast parameter is a hint whether the packet is + // being broadcast. + Process(frame []byte, dec *EthernetDecoder, broadcast bool) +} + +type MultiFlowOp struct { + broadcast bool + ops []FlowOp +} + +func NewMultiFlowOp(broadcast bool) *MultiFlowOp { + return &MultiFlowOp{broadcast: broadcast} +} + +func (mfop *MultiFlowOp) Add(ops ...FlowOp) { + mfop.ops = append(mfop.ops, ops...) +} + +func (mfop *MultiFlowOp) Process(frame []byte, dec *EthernetDecoder, + broadcast bool) { + for _, op := range mfop.ops { + op.Process(frame, dec, mfop.broadcast) + } +} diff --git a/router/local_peer.go b/router/local_peer.go index 7f331556..3a57e8a4 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -27,40 +27,54 @@ func NewLocalPeer(name PeerName, nickName string, router *Router) *LocalPeer { return peer } -func (peer *LocalPeer) Forward(dstPeer *Peer, frame []byte, dec *EthernetDecoder) { - peer.Relay(peer.Peer, dstPeer, frame, dec) +func (peer *LocalPeer) Forward(dstPeer *Peer, key PacketKey) FlowOp { + return peer.Relay(ForwardPacketKey{ + PacketKey: key, + SrcPeer: peer.Peer, + DstPeer: dstPeer, + }) } -func (peer *LocalPeer) Broadcast(frame []byte, dec *EthernetDecoder) { - peer.RelayBroadcast(peer.Peer, frame, dec) +func (peer *LocalPeer) Broadcast(key PacketKey) FlowOp { + return peer.RelayBroadcast(peer.Peer, key) } -func (peer *LocalPeer) Relay(srcPeer, dstPeer *Peer, frame []byte, dec *EthernetDecoder) { - relayPeerName, found := peer.router.Routes.Unicast(dstPeer.Name) +func (peer *LocalPeer) Relay(key ForwardPacketKey) FlowOp { + relayPeerName, found := peer.router.Routes.Unicast(key.DstPeer.Name) if !found { // 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:", dstPeer) - return + log.Println("Received packet for unknown destination:", key.DstPeer) + return nil } + conn, found := peer.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 + return nil } - conn.(*LocalConnection).Forward(srcPeer, dstPeer, frame, dec, false) + + return conn.(*LocalConnection).Forward(key) } -func (peer *LocalPeer) RelayBroadcast(srcPeer *Peer, frame []byte, dec *EthernetDecoder) { +func (peer *LocalPeer) RelayBroadcast(srcPeer *Peer, key PacketKey) FlowOp { nextHops := peer.router.Routes.Broadcast(srcPeer.Name) if len(nextHops) == 0 { - return + return nil } + + op := NewMultiFlowOp(true) + for _, conn := range peer.ConnectionsTo(nextHops) { - conn.(*LocalConnection).Forward(srcPeer, conn.Remote(), frame, - dec, true) + op.Add(conn.(*LocalConnection).Forward(ForwardPacketKey{ + PacketKey: key, + SrcPeer: srcPeer, + DstPeer: conn.Remote(), + })) } + + return op } func (peer *LocalPeer) Connections() ConnectionSet { diff --git a/router/mac_cache.go b/router/mac_cache.go index 021030c9..a798c48a 100644 --- a/router/mac_cache.go +++ b/router/mac_cache.go @@ -56,15 +56,15 @@ func (cache *MacCache) Enter(mac net.HardwareAddr, peer *Peer) bool { return false } -func (cache *MacCache) Lookup(mac net.HardwareAddr) (*Peer, bool) { +func (cache *MacCache) Lookup(mac net.HardwareAddr) *Peer { key := macint(mac) cache.RLock() defer cache.RUnlock() entry, found := cache.table[key] if !found { - return nil, false + return nil } - return entry.peer, true + return entry.peer } func (cache *MacCache) Delete(peer *Peer) bool { diff --git a/router/overlay.go b/router/overlay.go index b8ac0699..c8c8d0c7 100644 --- a/router/overlay.go +++ b/router/overlay.go @@ -39,8 +39,7 @@ type ForwarderParams struct { // When a consumer is called, the decoder will already have been used // to decode the frame. -type OverlayConsumer func(src *Peer, dst *Peer, frame []byte, - dec *EthernetDecoder) +type OverlayConsumer func(ForwardPacketKey) FlowOp // Crypto settings for a forwarder. type OverlayCrypto struct { @@ -58,11 +57,8 @@ type OverlayForwarder interface { // (e.g. on another thread). SetListener(OverlayForwarderListener) - // Forward a packet across the connection. The caller must - // supply an EthernetDecoder specific to this thread, which - // has already been used to decode the frame. - Forward(src *Peer, dest *Peer, frame []byte, dec *EthernetDecoder, - broadcast bool) + // Forward a packet across the connection. + Forward(ForwardPacketKey) FlowOp Stop() @@ -88,8 +84,8 @@ func (NullOverlay) MakeForwarder(ForwarderParams) (OverlayForwarder, error) { func (NullOverlay) SetListener(OverlayForwarderListener) { } -func (NullOverlay) Forward(*Peer, *Peer, []byte, *EthernetDecoder, - bool) { +func (NullOverlay) Forward(ForwardPacketKey) FlowOp { + return nil } func (NullOverlay) Stop() { diff --git a/router/pcap.go b/router/pcap.go index 138cfcd1..cddc366e 100644 --- a/router/pcap.go +++ b/router/pcap.go @@ -106,10 +106,14 @@ func (p *Pcap) String() string { return fmt.Sprint(p.iface.Name, " (via pcap)") } -func (p *Pcap) InjectPacket(pkt []byte) error { +func (p *Pcap) InjectPacket(PacketKey) FlowOp { + return p +} + +func (p *Pcap) Process(frame []byte, dec *EthernetDecoder, broadcast bool) { p.mutex.Lock() defer p.mutex.Unlock() - return p.writeHandle.WritePacketData(pkt) + checkWarn(p.writeHandle.WritePacketData(frame)) } func (p *Pcap) sniff(readHandle *pcap.Handle, consumer BridgeConsumer) { @@ -123,7 +127,21 @@ func (p *Pcap) sniff(readHandle *pcap.Handle, consumer BridgeConsumer) { checkFatal(err) dec.DecodeLayers(pkt) - consumer(pkt, dec) + if len(dec.decoded) == 0 { + continue + } + + if fop := consumer(dec.PacketKey()); fop != nil { + // 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 + // overwriting the data + pktLen := len(pkt) + pktCopy := make([]byte, pktLen, pktLen) + copy(pktCopy, pkt) + + fop.Process(pktCopy, dec, false) + } } } diff --git a/router/router.go b/router/router.go index a26745f0..85140055 100644 --- a/router/router.go +++ b/router/router.go @@ -20,8 +20,6 @@ const ( // [3] control rate at which new tokens are added to the bucket -type LogFrameFunc func(string, []byte, *EthernetDecoder) - type Config struct { Port int ProtocolMinVersion byte @@ -29,11 +27,16 @@ type Config struct { ConnLimit int PeerDiscovery bool BufSz int - LogFrame LogFrameFunc + PacketLogging PacketLogging Bridge Bridge Overlay Overlay } +type PacketLogging interface { + LogPacket(string, PacketKey) + LogForwardPacket(string, ForwardPacketKey) +} + type Router struct { Config Ourself *LocalPeer @@ -97,48 +100,45 @@ func (router *Router) UsingPassword() bool { return router.Password != nil } -func (router *Router) handleCapturedPacket(frameData []byte, dec *EthernetDecoder) { - router.LogFrame("Sniffed", frameData, dec) - decodedLen := len(dec.decoded) - if decodedLen == 0 { - return - } - srcMac := dec.Eth.SrcMAC - srcPeer, found := router.Macs.Lookup(srcMac) +func (router *Router) handleCapturedPacket(key PacketKey) FlowOp { + router.PacketLogging.LogPacket("Captured", key) + srcMac := net.HardwareAddr(key.SrcMAC[:]) + srcPeer := router.Macs.Lookup(srcMac) + // We need to filter out frames we injected ourselves. For such // frames, the srcMAC will have been recorded as associated with a // different peer. - if found && srcPeer != router.Ourself.Peer { - return + if srcPeer != nil && srcPeer != router.Ourself.Peer { + return nil } + if router.Macs.Enter(srcMac, router.Ourself.Peer) { log.Println("Discovered local MAC", srcMac) } - if dec.DropFrame() { - return - } - dstMac := dec.Eth.DstMAC - dstPeer, found := router.Macs.Lookup(dstMac) - if found && dstPeer == router.Ourself.Peer { - return - } - router.LogFrame("Forwarding", frameData, dec) - // at this point 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 overwriting the data - frameLen := len(frameData) - frameCopy := make([]byte, frameLen, frameLen) - copy(frameCopy, frameData) - - // If we don't know which peer corresponds to the dest MAC, - // broadcast it. - if !found { - router.Ourself.Broadcast(frameCopy, dec) - return + // Discard STP broadcasts + if key.DstMAC == [...]byte{0x01, 0x80, 0xC2, 0x00, 0x00, 0x00} { + return nil } - router.Ourself.Forward(dstPeer, frameCopy, dec) + dstMac := net.HardwareAddr(key.DstMAC[:]) + switch dstPeer := router.Macs.Lookup(dstMac); dstPeer { + case router.Ourself.Peer: + // The packet is destined for a local MAC. The bridge + // won't normally send us such packets, and if it does + // 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 + case nil: + // If we don't know which peer corresponds to the dest + // MAC, broadcast it. + router.PacketLogging.LogPacket("Broadcasting", key) + return router.Ourself.Broadcast(key) + default: + router.PacketLogging.LogPacket("Forwarding", key) + return router.Ourself.Forward(dstPeer, key) + } } func (router *Router) listenTCP(localPort int) { @@ -170,29 +170,44 @@ func (router *Router) acceptTCP(tcpConn *net.TCPConn) { StartLocalConnection(connRemote, tcpConn, nil, router, true) } -func (router *Router) handleForwardedPacket(srcPeer *Peer, dstPeer *Peer, - frame []byte, dec *EthernetDecoder) { - if dstPeer != router.Ourself.Peer { +func (router *Router) handleForwardedPacket(key ForwardPacketKey) FlowOp { + if key.DstPeer != router.Ourself.Peer { // it's not for us, we're just relaying it - router.LogFrame("Relaying", frame, dec) - router.Ourself.Relay(srcPeer, dstPeer, frame, dec) - return + router.PacketLogging.LogForwardPacket("Relaying", key) + return router.Ourself.Relay(key) } - srcMac := dec.Eth.SrcMAC - dstMac := dec.Eth.DstMAC + // At this point, it's either unicast to us, or a broadcast + // (because the DstPeer on a forwarded broadcast packet is + // always set to the peer being forwarded to) - if router.Macs.Enter(srcMac, srcPeer) { - log.Println("Discovered remote MAC", srcMac, "at", srcPeer) + srcMac := net.HardwareAddr(key.SrcMAC[:]) + dstMac := net.HardwareAddr(key.DstMAC[:]) + if router.Macs.Enter(srcMac, key.SrcPeer) { + log.Println("Discovered remote MAC", srcMac, "at", key.SrcPeer) } - router.LogFrame("Injecting", frame, dec) - checkWarn(router.Bridge.InjectPacket(frame)) + router.PacketLogging.LogForwardPacket("Injecting", key) + injectFop := router.Bridge.InjectPacket(key.PacketKey) + dstPeer := router.Macs.Lookup(dstMac) + if dstPeer == router.Ourself.Peer { + return injectFop + } - dstPeer, found := router.Macs.Lookup(dstMac) - if !found || dstPeer != router.Ourself.Peer { - router.LogFrame("Relaying broadcast", frame, dec) - router.Ourself.RelayBroadcast(srcPeer, frame, dec) + router.PacketLogging.LogForwardPacket("Relaying broadcast", key) + relayFop := router.Ourself.RelayBroadcast(key.SrcPeer, key.PacketKey) + switch { + case injectFop == nil: + return relayFop + + case relayFop == nil: + return injectFop + + default: + mfop := NewMultiFlowOp(false) + mfop.Add(injectFop) + mfop.Add(relayFop) + return mfop } } diff --git a/router/sleeve.go b/router/sleeve.go index 0c002bd8..f844f4c1 100644 --- a/router/sleeve.go +++ b/router/sleeve.go @@ -231,11 +231,23 @@ func (sleeve *SleeveOverlay) handleFrame(sender *net.UDPAddr, return } + sleeve.sendToConsumer(srcPeer, dstPeer, frame, dec) +} + +func (sleeve *SleeveOverlay) sendToConsumer(srcPeer, dstPeer *Peer, + frame []byte, dec *EthernetDecoder) { if sleeve.consumer == nil { return } - sleeve.consumer(srcPeer, dstPeer, frame, dec) + fop := sleeve.consumer(ForwardPacketKey{ + SrcPeer: srcPeer, + DstPeer: dstPeer, + PacketKey: dec.PacketKey(), + }) + if fop != nil { + fop.Process(frame, dec, false) + } } type udpSender interface { @@ -389,8 +401,18 @@ func (fwd *sleeveForwarder) SetListener(listener OverlayForwarderListener) { } } -func (fwd *sleeveForwarder) Forward(src *Peer, dst *Peer, frame []byte, - dec *EthernetDecoder, broadcast bool) { +type curriedForward struct { + fwd *sleeveForwarder + key ForwardPacketKey +} + +func (fwd *sleeveForwarder) Forward(key ForwardPacketKey) FlowOp { + return curriedForward{fwd, key} +} + +func (f curriedForward) Process(frame []byte, dec *EthernetDecoder, + broadcast bool) { + fwd := f.fwd fwd.lock.RLock() haveContact := (fwd.remoteAddr != nil) mtu := fwd.mtu @@ -403,8 +425,8 @@ func (fwd *sleeveForwarder) Forward(src *Peer, dst *Peer, frame []byte, return } - srcName := src.NameByte - dstName := dst.NameByte + srcName := f.key.SrcPeer.NameByte + dstName := f.key.DstPeer.NameByte // We could use non-blocking channel sends here, i.e. drop frames // on the floor when the forwarder is busy. This would allow our @@ -444,7 +466,8 @@ func (fwd *sleeveForwarder) Forward(src *Peer, dst *Peer, frame []byte, // The frag-needed packet does not have DF set, so the // potential recursion here is bounded. - fwd.sleeve.consumer(dst, src, fragNeededPacket, dec) + fwd.sleeve.sendToConsumer(f.key.DstPeer, f.key.SrcPeer, + fragNeededPacket, dec) return }