Flows

router/connection.go

@@ -379,8 +379,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 {

router/ethernet_decoder.go

@@ -25,6 +25,12 @@ func (dec *EthernetDecoder) DecodeLayers(data []byte) error {
 	return 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{
@@ -61,15 +67,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)

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.
+	Send(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) Send(frame []byte, dec *EthernetDecoder,
+	broadcast bool) {
+	for _, op := range mfop.ops {
+		op.Send(frame, dec, mfop.broadcast)
+	}
+}

router/interhost.go

@@ -19,8 +19,7 @@ type InterHost interface {
 
 // When a consumer is called, the decoder will already have been used
 // to decode the frame.
-type InterHostConsumer func(src *Peer, dst *Peer, frame []byte,
-	dec *EthernetDecoder)
+type InterHostConsumer func(ForwardPacketKey) FlowOp
 
 // Crypto settings for a forwarder.
 type InterHostCrypto struct {
@@ -38,11 +37,8 @@ type InterHostForwarder interface {
 	// (e.g. on another thread).
 	SetListener(InterHostForwarderListener)
 
-	// 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
 
 	Close()
 
@@ -69,8 +65,8 @@ func (NullInterHost) MakeForwarder(*Peer, net.IP, *net.UDPAddr, uint64,
 func (NullInterHost) SetListener(InterHostForwarderListener) {
 }
 
-func (NullInterHost) Forward(*Peer, *Peer, []byte, *EthernetDecoder,
-	bool) {
+func (NullInterHost) Forward(ForwardPacketKey) FlowOp {
+	return nil
 }
 
 func (NullInterHost) Close() {

router/intrahost.go

@@ -3,7 +3,7 @@ package router
 // Interface to intra-host (i.e. container) packet handling
 type IntraHost interface {
 	// Inject a packet to be delivered locally
-	InjectPacket([]byte) error
+	InjectPacket(PacketKey) FlowOp
 
 	// Start consuming packets
 	ConsumePackets(IntraHostConsumer) error
@@ -13,12 +13,12 @@ type IntraHostConsumer interface {
 	// A locally captured packet.  The caller must supply an
 	// EthernetDecoder specific to this thread, but it should not
 	// have been used to decode the packet data.
-	CapturedPacket([]byte, *EthernetDecoder)
+	CapturedPacket(PacketKey) FlowOp
 }
 
 type NullIntraHost struct{}
 
-func (NullIntraHost) InjectPacket([]byte) error {
+func (NullIntraHost) InjectPacket(PacketKey) FlowOp {
 	return nil
 }
 

router/local_peer.go

@@ -28,40 +28,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) Info() string {

router/mac_cache.go

@@ -58,15 +58,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 {

router/pcap.go

@@ -94,10 +94,14 @@ func (p *Pcap) String() string {
 	return fmt.Sprint(p.iface, "(via pcap)")
 }
 
-func (p *Pcap) InjectPacket(pkt []byte) error {
+func (p *Pcap) InjectPacket(PacketKey) FlowOp {
+	return p
+}
+
+func (p *Pcap) Send(frame []byte, dec *EthernetDecoder, broadcast bool) {
 	p.writeHandleMutex.Lock()
 	defer p.writeHandleMutex.Unlock()
-	return p.writeHandle.WritePacketData(pkt)
+	checkWarn(p.writeHandle.WritePacketData(frame))
 }
 
 func (p *Pcap) sniff(readHandle *pcap.Handle, consumer IntraHostConsumer) {
@@ -111,6 +115,23 @@ func (p *Pcap) sniff(readHandle *pcap.Handle, consumer IntraHostConsumer) {
 		}
 
 		checkFatal(err)
-		consumer.CapturedPacket(pkt, dec)
+
+		dec.DecodeLayers(pkt)
+		decodedLen := len(dec.decoded)
+		if decodedLen == 0 {
+			return
+		}
+
+		if fop := consumer.CapturedPacket(dec.PacketKey()); fop != nil {
+			// 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
+			pktLen := len(pkt)
+			pktCopy := make([]byte, pktLen, pktLen)
+			copy(pktCopy, pkt)
+
+			fop.Send(pktCopy, dec, false)
+		}
 	}
 }

router/router.go

@@ -93,49 +93,39 @@ func (router *Router) Status() string {
 	return buf.String()
 }
 
-func (router *Router) CapturedPacket(frameData []byte, dec *EthernetDecoder) {
-	router.LogFrame("Sniffed", frameData, nil)
-	dec.DecodeLayers(frameData)
-	decodedLen := len(dec.decoded)
-	if decodedLen == 0 {
-		return
-	}
-	srcMac := dec.Eth.SrcMAC
-	srcPeer, found := router.Macs.Lookup(srcMac)
+func (router *Router) CapturedPacket(key PacketKey) FlowOp {
+	//router.LogFrame("Sniffed", frameData, nil)
+	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:
+		return nil
+	case nil:
+		// If we don't know which peer corresponds to the dest
+		// MAC, broadcast it.
+		return router.Ourself.Broadcast(key)
+	default:
+		//router.LogFrame("Forwarding", frameData, dec)
+		return router.Ourself.Forward(dstPeer, key)
+	}
 }
 
 func (router *Router) listenTCP(localPort int) {
@@ -166,29 +156,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.LogFrame("Relaying", frame, dec)
+		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.IntraHost.InjectPacket(frame))
+	//router.LogFrame("Injecting", frame, dec)
+	injectFop := router.IntraHost.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.LogFrame("Relaying broadcast", frame, dec)
+	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
 	}
 }
 

router/sleeve.go

@@ -185,11 +185,23 @@ func (sleeve *SleeveInterHost) handleFrame(sender *net.UDPAddr,
 		return
 	}
 
+	sleeve.sendToConsumer(srcPeer, dstPeer, frame, dec)
+}
+
+func (sleeve *SleeveInterHost) 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.Send(frame, dec, false)
+	}
 }
 
 type udpSender interface {
@@ -345,8 +357,18 @@ func (fwd *sleeveForwarder) SetListener(listener InterHostForwarderListener) {
 	}
 }
 
-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) Send(frame []byte, dec *EthernetDecoder,
+	broadcast bool) {
+	fwd := f.fwd
 	fwd.lock.RLock()
 	haveContact := (fwd.remoteAddr != nil)
 	effectivePMTU := fwd.effectivePMTU
@@ -358,8 +380,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
@@ -399,7 +421,7 @@ 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, fn, dec)
+		fwd.sleeve.sendToConsumer(f.key.DstPeer, f.key.SrcPeer, fn, dec)
 		return
 	}