Fast datapath

router/fastdp.go

@@ -0,0 +1,593 @@
+package router
+
+import (
+	"encoding/binary"
+	"fmt"
+	"log"
+	"net"
+	"sync"
+
+	"github.com/weaveworks/go-odp/odp"
+)
+
+type thunk func() FlowOp
+
+// A bridgePortID is either an ODP vport or the router.  We express it
+// like this to use it as a map key.
+type bridgePortID struct {
+	vport  odp.VportID
+	router bool
+}
+
+var routerPortID = bridgePortID{router: true}
+
+type FastDatapath struct {
+	vxlanPort int
+
+	// The lock guards rthe FastDatapath state, and also
+	// synchronises use of the dpif
+	lock              sync.Mutex
+	dpif              *odp.Dpif
+	dp                odp.DatapathHandle
+	consumingMisses   bool
+	missHandlers      map[odp.VportID]func(odp.FlowKeys) thunk
+	vxlanVportID      odp.VportID
+	localPeer         *Peer
+	peers             *Peers
+	interHostConsumer InterHostConsumer
+
+	// Bridge state: A bridge port is represented as a function
+	// that says how to send out a packet on that port.
+	bridgePorts map[bridgePortID]func(PacketKey) thunk
+	bridgeMACs  map[MAC]func(PacketKey) thunk
+
+	// Only accessed from Miss, so not locked
+	dec *EthernetDecoder
+}
+
+func NewFastDatapath(dpname string, vxlanPort int) (*FastDatapath, error) {
+	dpif, err := odp.NewDpif()
+	if err != nil {
+		return nil, err
+	}
+
+	success := false
+	defer func() {
+		if !success {
+			dpif.Close()
+		}
+	}()
+
+	dp, err := dpif.LookupDatapath(dpname)
+	if err != nil {
+		return nil, err
+	}
+
+	fastdp := &FastDatapath{
+		vxlanPort:    vxlanPort,
+		dpif:         dpif,
+		dp:           dp,
+		missHandlers: make(map[odp.VportID]func(odp.FlowKeys) thunk),
+		bridgePorts:  make(map[bridgePortID]func(PacketKey) thunk),
+		bridgeMACs:   make(map[MAC]func(PacketKey) thunk),
+		dec:          NewEthernetDecoder(),
+	}
+
+	if err := fastdp.deleteVxlanVports(); err != nil {
+		return nil, err
+	}
+
+	if err := fastdp.clearFlows(); err != nil {
+		return nil, err
+	}
+
+	fastdp.vxlanVportID, err = fastdp.dp.CreateVport(
+		odp.NewVxlanVportSpec("vxlan", uint16(vxlanPort)))
+	if err != nil {
+		return nil, err
+	}
+
+	success = true
+	return fastdp, nil
+}
+
+func (fastdp *FastDatapath) consumeMisses() error {
+	if !fastdp.consumingMisses {
+		if err := fastdp.dp.ConsumeMisses(fastdp); err != nil {
+			return err
+		}
+
+		fastdp.consumingMisses = true
+	}
+
+	return nil
+}
+
+func (fastdp *FastDatapath) Close() error {
+	err := fastdp.dpif.Close()
+	fastdp.dpif = nil
+	return err
+}
+
+func (fastdp *FastDatapath) clearFlows() error {
+	flows, err := fastdp.dp.EnumerateFlows()
+	if err != nil {
+		return err
+	}
+
+	for _, flow := range flows {
+		err = fastdp.dp.DeleteFlow(flow.FlowSpec)
+		if err != nil && !odp.IsNoSuchFlowError(err) {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (fastdp *FastDatapath) deleteVxlanVports() error {
+	vports, err := fastdp.dp.EnumerateVports()
+	if err != nil {
+		return err
+	}
+
+	for _, vport := range vports {
+		if vport.Spec.TypeName() != "vxlan" {
+			continue
+		}
+
+		err = fastdp.dp.DeleteVport(vport.ID)
+		if err != nil && !odp.IsNoSuchVportError(err) {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (fastdp *FastDatapath) Error(err error, stopped bool) {
+	// XXX fatal if stopped
+	log.Println("Error while listening on datapath:", err)
+}
+
+func (fastdp *FastDatapath) Miss(packet []byte, flowKeys odp.FlowKeys) error {
+	thunk := fastdp.handleMiss(flowKeys)
+	if thunk != nil {
+		fastdp.send(thunk(), packet)
+	}
+
+	return nil
+}
+
+func (fastdp *FastDatapath) handleMiss(fks odp.FlowKeys) thunk {
+	fastdp.lock.Lock()
+	defer fastdp.lock.Unlock()
+
+	ingress := fks[odp.OVS_KEY_ATTR_IN_PORT].(odp.InPortFlowKey).VportID()
+
+	log.Println("Got miss", fks, "on port", ingress)
+
+	handler := fastdp.missHandlers[ingress]
+	if handler == nil {
+		handler = fastdp.makeMissHandler(ingress)
+		if handler == nil {
+			return simpleThunk(vetoFlowCreation)
+		}
+
+		fastdp.missHandlers[ingress] = handler
+	}
+
+	return handler(fks)
+}
+
+func (fastdp *FastDatapath) makeMissHandler(
+	ingress odp.VportID) func(odp.FlowKeys) thunk {
+	// Auto-add netdev and internal vports to the bridge (i.e. not
+	// vxlan vports)
+	vport, err := fastdp.dp.LookupVport(ingress)
+	if err != nil {
+		log.Println(err)
+		return nil
+	}
+
+	typ := vport.Spec.TypeName()
+	if typ != "netdev" && typ != "internal" {
+		return nil
+	}
+
+	// Construct a bridge port for the netdev, which simply
+	// outputs packets
+	bridgePort := func(key PacketKey) thunk {
+		return simpleThunk(fastdp.odpActions(
+			odp.NewOutputAction(ingress)))
+	}
+	fastdp.bridgePorts[bridgePortID{vport: ingress}] = bridgePort
+
+	// Clear flows, in order to recalculate flows for broadcasts
+	// on the bridge
+	checkWarn(fastdp.clearFlows())
+
+	// Packets from the netdev are processed by the bridge
+	return func(fks odp.FlowKeys) thunk {
+		return multiThunk(fastdp.bridge(bridgePortID{vport: ingress},
+			flowKeysToPacketKey(fks)))
+	}
+}
+
+func flowKeysToPacketKey(fks odp.FlowKeys) PacketKey {
+	eth := fks[odp.OVS_KEY_ATTR_ETHERNET].(odp.EthernetFlowKey).Key()
+	return PacketKey{SrcMAC: eth.EthSrc, DstMAC: eth.EthDst}
+}
+
+// Send a packet, creating a corresponding ODP flow rule if possible
+func (fastdp *FastDatapath) send(fops FlowOp, frame []byte) {
+	// Gather the actions from actionFlowOps, execute any others
+	var dec *EthernetDecoder
+	flow := odp.NewFlowSpec()
+	createFlow := true
+
+	for _, xfop := range FlattenFlowOp(fops) {
+		switch fop := xfop.(type) {
+		case interface {
+			updateFlowSpec(*odp.FlowSpec)
+		}:
+			fop.updateFlowSpec(&flow)
+		case vetoFlowCreationFlowOp:
+			createFlow = false
+		default:
+			// A foreign flow op, so send the packet the
+			// normal way, decoding the packet lazily.
+			if dec == nil {
+				dec = fastdp.dec
+				dec.DecodeLayers(frame)
+				createFlow = false
+			}
+
+			if len(dec.decoded) != 0 {
+				fop.Send(frame, dec, false)
+			}
+		}
+	}
+
+	fastdp.lock.Lock()
+	defer fastdp.lock.Unlock()
+
+	if len(flow.Actions) != 0 {
+		checkWarn(fastdp.dp.Execute(frame, nil, flow.Actions))
+	}
+
+	if createFlow {
+		log.Println("Creating flow", flow)
+		checkWarn(fastdp.dp.CreateFlow(flow))
+	}
+}
+
+type odpActionsFlowOp struct {
+	fastdp  *FastDatapath
+	actions []odp.Action
+}
+
+func (fastdp *FastDatapath) odpActions(actions ...odp.Action) FlowOp {
+	return odpActionsFlowOp{
+		fastdp:  fastdp,
+		actions: actions,
+	}
+}
+
+func (fop odpActionsFlowOp) updateFlowSpec(flow *odp.FlowSpec) {
+	flow.AddActions(fop.actions)
+}
+
+func (fop odpActionsFlowOp) Send(frame []byte, dec *EthernetDecoder, bc bool) {
+	fastdp := fop.fastdp
+	fastdp.lock.Lock()
+	defer fastdp.lock.Unlock()
+	checkWarn(fastdp.dp.Execute(frame, nil, fop.actions))
+}
+
+type nopFlowOp struct{}
+
+func (nopFlowOp) Send([]byte, *EthernetDecoder, bool) {
+	// A nopFlowOp just provides a hint about flow creation, it
+	// doesn't send anything
+}
+
+type vetoFlowCreationFlowOp struct {
+	nopFlowOp
+}
+
+var vetoFlowCreation = vetoFlowCreationFlowOp{}
+
+type odpFlowKeyFlowOp struct {
+	key odp.FlowKey
+	nopFlowOp
+}
+
+func odpFlowKey(key odp.FlowKey) FlowOp {
+	return odpFlowKeyFlowOp{key: key}
+}
+
+func (fop odpFlowKeyFlowOp) updateFlowSpec(flow *odp.FlowSpec) {
+	flow.AddKey(fop.key)
+}
+
+func odpEthernetFlowKey(key PacketKey) FlowOp {
+	fk := odp.NewEthernetFlowKey()
+	fk.SetEthSrc(key.SrcMAC)
+	fk.SetEthDst(key.DstMAC)
+	return odpFlowKeyFlowOp{key: fk}
+}
+
+func simpleThunk(fop FlowOp) thunk {
+	return func() FlowOp { return fop }
+}
+
+func callThunks(thunks []thunk) FlowOp {
+	if len(thunks) == 1 {
+		return thunks[0]()
+	}
+
+	mfop := NewMultiFlowOp(false)
+	for _, thunk := range thunks {
+		mfop.Add(thunk())
+	}
+
+	return mfop
+}
+
+func multiThunk(thunks []thunk) thunk {
+	return func() FlowOp { return callThunks(thunks) }
+}
+
+// IntraHost
+
+func (fastdp *FastDatapath) ConsumeIntraHostPackets(
+	consumer IntraHostConsumer) error {
+	fastdp.lock.Lock()
+	defer fastdp.lock.Unlock()
+
+	if fastdp.bridgePorts[routerPortID] != nil {
+		panic("FastDatapath already has an IntraHostConsumer")
+	}
+
+	fastdp.bridgePorts[routerPortID] = func(key PacketKey) thunk {
+		return func() FlowOp {
+			return consumer.CapturedPacket(key)
+		}
+	}
+
+	return fastdp.consumeMisses()
+}
+
+func (fastdp *FastDatapath) InjectPacket(key PacketKey) FlowOp {
+	return callThunks(func() []thunk {
+		fastdp.lock.Lock()
+		defer fastdp.lock.Unlock()
+		return fastdp.bridge(routerPortID, key)
+	}())
+}
+
+// A trivial bridge implementation
+func (fastdp *FastDatapath) bridge(ingress bridgePortID,
+	key PacketKey) []thunk {
+	if fastdp.bridgeMACs[key.SrcMAC] == nil {
+		// Learn the source MAC
+		fastdp.bridgeMACs[key.SrcMAC] = fastdp.bridgePorts[ingress]
+	}
+
+	// If we know about the destination MAC, deliver it to the
+	// associated port.
+	dst := fastdp.bridgeMACs[key.DstMAC]
+	if dst != nil {
+		return []thunk{simpleThunk(odpEthernetFlowKey(key)), dst(key)}
+	}
+
+	// Otherwise, it might be a real broadcast, or it might just
+	// be for a MAC we don't know about yet.  Either way, we'll
+	// broadcast it.
+	var fop FlowOp
+	thunks := make([]thunk, 1, len(fastdp.bridgePorts)+2)
+
+	if (key.DstMAC[0] & 1) == 0 {
+		// Not a real broadcast, so doon't create a flow rule.
+		// If we did, we'd need to clear the flows every time
+		// we learned a new MAC address, or have a more
+		// complicated selective invalidation scheme.
+		fop = vetoFlowCreation
+	} else {
+		// A real broadcast
+		fop = odpEthernetFlowKey(key)
+
+		if !ingress.router {
+			// The rule depends on the ingress vport.
+			thunks = thunks[:2]
+			thunks[1] = simpleThunk(odpFlowKey(odp.NewInPortFlowKey(
+				ingress.vport)))
+		}
+	}
+
+	thunks[0] = simpleThunk(fop)
+
+	// Send to all ports except the one it came in on.
+	for id, dst := range fastdp.bridgePorts {
+		if id != ingress {
+			thunks = append(thunks, dst(key))
+		}
+	}
+
+	return thunks
+}
+
+// InterHost
+
+func (fastdp *FastDatapath) ConsumeInterHostPackets(
+	localPeer *Peer, peers *Peers, consumer InterHostConsumer) error {
+	fastdp.lock.Lock()
+	defer fastdp.lock.Unlock()
+
+	if fastdp.missHandlers[fastdp.vxlanVportID] != nil {
+		panic("FastDatapath already has an InterHostConsumer")
+	}
+
+	fastdp.localPeer = localPeer
+	fastdp.peers = peers
+
+	fastdp.missHandlers[fastdp.vxlanVportID] = func(fks odp.FlowKeys) thunk {
+		tunnel := fks[odp.OVS_KEY_ATTR_TUNNEL].(odp.TunnelFlowKey).Key()
+		srcPeer, dstPeer := fastdp.extractPeers(tunnel.TunnelId)
+		if srcPeer == nil || dstPeer == nil {
+			return simpleThunk(vetoFlowCreation)
+		}
+
+		key := ForwardPacketKey{
+			SrcPeer:   srcPeer,
+			DstPeer:   dstPeer,
+			PacketKey: flowKeysToPacketKey(fks),
+		}
+
+		// The resulting flow rule should be restricted to
+		// packets with the same tunnelID
+		var tunnelFlowKey odp.TunnelFlowKey
+		tunnelFlowKey.SetTunnelId(tunnel.TunnelId)
+		tunnelFlowKey.SetIpv4Src(tunnel.Ipv4Src)
+		tunnelFlowKey.SetIpv4Dst(tunnel.Ipv4Dst)
+
+		return func() FlowOp {
+			fop := NewMultiFlowOp(false)
+			fop.Add(odpFlowKey(tunnelFlowKey))
+			fop.Add(consumer(key))
+			return fop
+		}
+	}
+
+	return fastdp.consumeMisses()
+}
+
+func (fastdp *FastDatapath) InvalidateRoutes() {
+	fmt.Println("InvalidateRoutes")
+	fastdp.lock.Lock()
+	defer fastdp.lock.Unlock()
+	checkWarn(fastdp.clearFlows())
+}
+
+func (fastdp *FastDatapath) InvalidateShortIDs() {
+	fmt.Println("InvalidateShortIDs")
+	fastdp.lock.Lock()
+	defer fastdp.lock.Unlock()
+	checkWarn(fastdp.clearFlows())
+}
+
+func (fastdp *FastDatapath) extractPeers(tunnelID [8]byte) (*Peer, *Peer) {
+	if fastdp.peers == nil {
+		return nil, nil
+	}
+
+	vni := binary.BigEndian.Uint64(tunnelID[:])
+	srcPeer := fastdp.peers.FetchByShortID(PeerShortID(vni & 0xfff))
+	dstPeer := fastdp.peers.FetchByShortID(PeerShortID((vni >> 12) & 0xfff))
+	return srcPeer, dstPeer
+}
+
+type FastDatapathForwarder struct {
+	fastdp         *FastDatapath
+	remotePeer     *Peer
+	localIP        [4]byte
+	sendControlMsg func([]byte) error
+
+	lock        sync.Mutex
+	listener    InterHostForwarderListener
+	remoteIP    [4]byte
+	established bool
+}
+
+func (fastdp *FastDatapath) MakeForwarder(remotePeer *Peer, localIP net.IP,
+	remote *net.UDPAddr, connUID uint64, crypto InterHostCrypto,
+	sendControlMsg func([]byte) error) (InterHostForwarder, error) {
+	if len(localIP) != 4 {
+		return nil, fmt.Errorf("local IP address %s is not IPv4", localIP)
+	}
+
+	fwd := &FastDatapathForwarder{
+		fastdp:         fastdp,
+		remotePeer:     remotePeer,
+		sendControlMsg: sendControlMsg,
+	}
+	copy(fwd.localIP[:], localIP)
+	return fwd, nil
+}
+
+func (fwd *FastDatapathForwarder) SetListener(
+	listener InterHostForwarderListener) {
+	fwd.lock.Lock()
+	defer fwd.lock.Unlock()
+
+	if listener == nil {
+		panic("nil listener")
+	}
+
+	fwd.listener = listener
+	if fwd.established {
+		listener.Established()
+	}
+
+	fwd.sendControlMsg(fwd.localIP[:])
+}
+
+func (fwd *FastDatapathForwarder) ControlMessage(msg []byte) {
+	fwd.lock.Lock()
+	defer fwd.lock.Unlock()
+
+	if len(msg) != 4 {
+		if fwd.listener != nil {
+			fwd.listener.Error(fmt.Errorf("FastDatapath control message wrong length %d", len(msg)))
+			fwd.listener = nil
+		}
+
+		return
+	}
+
+	if !fwd.established {
+		copy(fwd.remoteIP[:], msg)
+		fwd.established = true
+		if fwd.listener != nil {
+			fwd.listener.Established()
+		}
+	}
+}
+
+func (fwd *FastDatapathForwarder) Forward(key ForwardPacketKey) FlowOp {
+	fwd.lock.Lock()
+	defer fwd.lock.Unlock()
+	if !fwd.established {
+		// Ideally we could just return nil.  But then we
+		// would have to invalidate the resulting flows when
+		// we learn the remote IP.  So for now, just prevent
+		// flows.
+		return vetoFlowCreation
+	}
+
+	var sta odp.SetTunnelAction
+	sta.SetTunnelId(tunnelIDFor(key))
+	sta.SetIpv4Src(fwd.localIP)
+	sta.SetIpv4Dst(fwd.remoteIP)
+	sta.SetTos(0)
+	sta.SetTtl(64)
+	sta.SetDf(true)
+	sta.SetCsum(false)
+
+	return fwd.fastdp.odpActions(sta,
+		odp.NewOutputAction(fwd.fastdp.vxlanVportID))
+}
+
+func tunnelIDFor(key ForwardPacketKey) (tunnelID [8]byte) {
+	src := uint64(key.SrcPeer.ShortID)
+	dst := uint64(key.DstPeer.ShortID)
+	binary.BigEndian.PutUint64(tunnelID[:], src|dst<<12)
+	return
+}
+
+func (fwd *FastDatapathForwarder) Close() {
+	// Ideally we would delete all the relevant flows here.  But
+	// until we do that, it's probably not worth clearing all
+	// flows.
+}

router/flow.go

@@ -45,6 +45,28 @@ func (mfop *MultiFlowOp) Add(ops ...FlowOp) {
 func (mfop *MultiFlowOp) Send(frame []byte, dec *EthernetDecoder,
 	broadcast bool) {
 	for _, op := range mfop.ops {
-		op.Send(frame, dec, mfop.broadcast)
+		if op != nil {
+			op.Send(frame, dec, mfop.broadcast)
+		}
 	}
 }
+
+func FlattenFlowOp(fop FlowOp) []FlowOp {
+	return collectFlowOps(nil, fop)
+}
+
+func collectFlowOps(into []FlowOp, fop FlowOp) []FlowOp {
+	if fop == nil {
+		return into
+	}
+
+	if mfop, ok := fop.(*MultiFlowOp); ok {
+		for _, op := range mfop.ops {
+			into = collectFlowOps(into, op)
+		}
+
+		return into
+	}
+
+	return append(into, fop)
+}

router/interhost.go

@@ -7,7 +7,7 @@ import (
 // Interface to inter-host (i.e. overlay network) packet handling
 type InterHost interface {
 	// Start consuming forwarded packets.
-	ConsumePackets(*Peer, *Peers, InterHostConsumer) error
+	ConsumeInterHostPackets(*Peer, *Peers, InterHostConsumer) error
 
 	// The routes have changed, so any cached information should
 	// be discarded.

router/intrahost.go

@@ -6,7 +6,7 @@ type IntraHost interface {
 	InjectPacket(PacketKey) FlowOp
 
 	// Start consuming packets
-	ConsumePackets(IntraHostConsumer) error
+	ConsumeIntraHostPackets(IntraHostConsumer) error
 }
 
 type IntraHostConsumer interface {
@@ -22,7 +22,7 @@ func (NullIntraHost) InjectPacket(PacketKey) FlowOp {
 	return nil
 }
 
-func (NullIntraHost) ConsumePackets(IntraHostConsumer) error {
+func (NullIntraHost) ConsumeIntraHostPackets(IntraHostConsumer) error {
 	return nil
 }
 

router/odp.go

@@ -0,0 +1,60 @@
+package router
+
+import (
+	"github.com/weaveworks/go-odp/odp"
+)
+
+// ODP admin functionality
+
+func CreateDatapath(dpname string) error {
+	dpif, err := odp.NewDpif()
+	if err != nil {
+		return err
+	}
+
+	defer dpif.Close()
+
+	_, err = dpif.CreateDatapath(dpname)
+	if err != nil && !odp.IsDatapathNameAlreadyExistsError(err) {
+		return err
+	}
+
+	return nil
+}
+
+func DeleteDatapath(dpname string) error {
+	dpif, err := odp.NewDpif()
+	if err != nil {
+		return err
+	}
+
+	defer dpif.Close()
+
+	dp, err := dpif.LookupDatapath(dpname)
+	if err != nil {
+		if odp.IsNoSuchDatapathError(err) {
+			return nil
+		} else {
+			return err
+		}
+	}
+
+	return dp.Delete()
+}
+
+func AddDatapathInterface(dpname string, ifname string) error {
+	dpif, err := odp.NewDpif()
+	if err != nil {
+		return err
+	}
+
+	defer dpif.Close()
+
+	dp, err := dpif.LookupDatapath(dpname)
+	if err != nil {
+		return err
+	}
+
+	_, err = dp.CreateVport(odp.NewNetdevVportSpec(ifname))
+	return err
+}

router/pcap.go

@@ -31,7 +31,7 @@ func NewPcap(iface *net.Interface, bufSz int) (IntraHost, error) {
 	return &Pcap{iface: iface, bufSz: bufSz, writeHandle: wh}, nil
 }
 
-func (p *Pcap) ConsumePackets(consumer IntraHostConsumer) error {
+func (p *Pcap) ConsumeIntraHostPackets(consumer IntraHostConsumer) error {
 	rh, err := newPcapHandle(p.iface.Name, true, 65535, p.bufSz)
 	if err != nil {
 		return err

router/router.go

@@ -67,8 +67,8 @@ func NewRouter(config RouterConfig, name PeerName, nickName string) *Router {
 // Start listening for packets from containers, TCP connections and
 // packets forwarded over UDP.
 func (router *Router) Start() {
-	checkFatal(router.InterHost.ConsumePackets(router.Ourself.Peer, router.Peers, router.handleForwardedPacket))
-	checkFatal(router.IntraHost.ConsumePackets(router))
+	checkFatal(router.InterHost.ConsumeInterHostPackets(router.Ourself.Peer, router.Peers, router.handleForwardedPacket))
+	checkFatal(router.IntraHost.ConsumeIntraHostPackets(router))
 	router.listenTCP(router.Port)
 }
 

router/sleeve.go

@@ -39,8 +39,8 @@ func NewSleeveInterHost(localPort int) InterHost {
 	return &SleeveInterHost{localPort: localPort}
 }
 
-func (sleeve *SleeveInterHost) ConsumePackets(localPeer *Peer, peers *Peers,
-	consumer InterHostConsumer) error {
+func (sleeve *SleeveInterHost) ConsumeInterHostPackets(localPeer *Peer,
+	peers *Peers, consumer InterHostConsumer) error {
 	localAddr, err := net.ResolveUDPAddr("udp4",
 		fmt.Sprint(":", sleeve.localPort))
 	if err != nil {

weave

@@ -256,7 +256,8 @@ random_mac() {
 
 create_bridge() {
     [ ! -d /sys/class/net/$BRIDGE ] && {
-        ip link add name $BRIDGE type bridge
+        docker run --rm --privileged --net=host $IMAGE --createdp --dpname=$BRIDGE
+        #ip link add name $BRIDGE type bridge
         # Set a random MAC address on the bridge.  Current Linux
         # kernels already do this when creating a bridge, but there
         # are rumours it was not always so.
@@ -265,9 +266,9 @@ create_bridge() {
         # fails. Bridges take the lowest MTU of their interfaces. So
         # instead we create a temporary interface with the desired
         # MTU, attach that to the bridge, and then remove it again.
-        ip link add name v${CONTAINER_IFNAME}du mtu $MTU type dummy
-        ip link set dev v${CONTAINER_IFNAME}du master $BRIDGE
-        ip link del dev v${CONTAINER_IFNAME}du
+        #ip link add name v${CONTAINER_IFNAME}du mtu $MTU type dummy
+        #ip link set dev v${CONTAINER_IFNAME}du master $BRIDGE
+        #ip link del dev v${CONTAINER_IFNAME}du
         # Drop traffic from Docker bridge to Weave; it can break subnet isolation
         if [ "$DOCKER_BRIDGE" != "$BRIDGE" ] ; then
             # Note using -I to insert ahead of Docker's bridge rules
@@ -293,7 +294,8 @@ create_bridge() {
 
 destroy_bridge() {
     if [ -d /sys/class/net/$BRIDGE ] ; then
-        ip link del dev $BRIDGE
+        docker run --rm --privileged --net=host $IMAGE --deletedp --dpname=$BRIDGE
+        #ip link del dev $BRIDGE
     fi
     if [ "$DOCKER_BRIDGE" != "$BRIDGE" ] ; then
         run_iptables -t filter -D FORWARD -i $DOCKER_BRIDGE -o $BRIDGE -j DROP 2>/dev/null || true
@@ -385,7 +387,7 @@ connect_container_to_bridge() {
     if ! netnsenter ip link set $GUEST_IFNAME down ||
         ! netnsenter ip link set $GUEST_IFNAME name $CONTAINER_IFNAME up ||
         ! ip link set $LOCAL_IFNAME up ||
-        ! ip link set $LOCAL_IFNAME master $BRIDGE ||
+        ! docker run --rm --privileged --net=host $IMAGE --adddpif=$LOCAL_IFNAME --dpname=$BRIDGE ||
         ! configure_arp_cache $CONTAINER_IFNAME "netnsenter" ; then
         return 1
     fi
@@ -815,7 +817,7 @@ case "$COMMAND" in
         # when launching the weave container.
         CONTAINER=$(docker run --privileged -d --name=$CONTAINER_NAME \
             -p $PORT:$CONTAINER_PORT/tcp -p $PORT:$CONTAINER_PORT/udp -e WEAVE_PASSWORD -v /var/run/docker.sock:/var/run/docker.sock --net=host \
-            $WEAVE_DOCKER_ARGS $IMAGE -iface $CONTAINER_IFNAME -port $CONTAINER_PORT -name "$PEERNAME" -nickname "$(hostname)" $IPRANGE "$@")
+            $WEAVE_DOCKER_ARGS $IMAGE -dpname $BRIDGE -port $CONTAINER_PORT -name "$PEERNAME" -nickname "$(hostname)" $IPRANGE "$@")
         with_container_netns $CONTAINER launch >/dev/null
 
         wait_for_status $CONTAINER_NAME $HTTP_PORT

weaver/main.go

@@ -34,26 +34,36 @@ func main() {
 	runtime.GOMAXPROCS(procs)
 
 	var (
-		config      weave.RouterConfig
-		justVersion bool
-		ifaceName   string
-		routerName  string
-		nickName    string
-		password    string
-		wait        int
-		debug       bool
-		pktdebug    bool
-		prof        string
-		peers       []string
-		bufSzMB     int
-		httpAddr    string
-		iprangeCIDR string
-		peerCount   int
-		apiPath     string
-		iface       *net.Interface
+		// flags that cause immediate exit
+		justVersion          bool
+		createDatapath       bool
+		deleteDatapath       bool
+		addDatapathInterface string
+
+		config       weave.RouterConfig
+		ifaceName    string
+		routerName   string
+		nickName     string
+		password     string
+		wait         int
+		debug        bool
+		pktdebug     bool
+		prof         string
+		peers        []string
+		bufSzMB      int
+		httpAddr     string
+		iprangeCIDR  string
+		peerCount    int
+		apiPath      string
+		iface        *net.Interface
+		datapathName string
 	)
 
 	flag.BoolVar(&justVersion, "version", false, "print version and exit")
+	flag.BoolVar(&createDatapath, "createdp", false, "create ODP datapath and exit")
+	flag.BoolVar(&deleteDatapath, "deletedp", false, "delete ODP datapath and exit")
+	flag.StringVar(&addDatapathInterface, "adddpif", "", "add a network interface to the ODP datapath and exit")
+
 	flag.IntVar(&config.Port, "port", weave.Port, "router port")
 	flag.StringVar(&ifaceName, "iface", "", "name of interface to capture/inject from (disabled if blank)")
 	flag.StringVar(&routerName, "name", "", "name of router (defaults to MAC of interface)")
@@ -69,13 +79,25 @@ func main() {
 	flag.StringVar(&iprangeCIDR, "iprange", "", "IP address range to allocate within, in CIDR notation")
 	flag.IntVar(&peerCount, "initpeercount", 0, "number of peers in network (for IP address allocation)")
 	flag.StringVar(&apiPath, "api", "unix:///var/run/docker.sock", "Path to Docker API socket")
+	flag.StringVar(&datapathName, "dpname", "", "ODP datapath name")
 	flag.Parse()
 	peers = flag.Args()
 
 	InitDefaultLogging(debug)
-	if justVersion {
+
+	switch {
+	case justVersion:
 		fmt.Printf("weave router %s\n", version)
 		os.Exit(0)
+	case createDatapath:
+		checkFatal(weave.CreateDatapath(datapathName))
+		os.Exit(0)
+	case deleteDatapath:
+		checkFatal(weave.DeleteDatapath(datapathName))
+		os.Exit(0)
+	case addDatapathInterface != "":
+		checkFatal(weave.AddDatapathInterface(datapathName, addDatapathInterface))
+		os.Exit(0)
 	}
 
 	log.Println("Command line options:", options())
@@ -83,17 +105,32 @@ func main() {
 
 	var err error
 
+	if password == "" {
+		password = os.Getenv("WEAVE_PASSWORD")
+	}
+
 	if ifaceName != "" {
 		iface, err := weavenet.EnsureInterface(ifaceName, wait)
-		if err != nil {
-			log.Fatal(err)
-		}
-
+		checkFatal(err)
 		// bufsz flag is in MB
 		config.IntraHost, err = weave.NewPcap(iface, bufSzMB*1024*1024)
-		if err != nil {
-			log.Fatal(err)
+		checkFatal(err)
+		config.InterHost = weave.NewSleeveInterHost(config.Port)
+	}
+
+	if datapathName != "" {
+		if ifaceName != "" {
+			log.Fatal("Both -iface and -dpname supplied")
 		}
+
+		if password != "" {
+			log.Fatal("-password supplied; encryption is not suppoorted with fast datapath")
+		}
+
+		fastdp, err := weave.NewFastDatapath(datapathName, config.Port)
+		checkFatal(err)
+		config.IntraHost = fastdp
+		config.InterHost = fastdp
 	}
 
 	if routerName == "" {
@@ -104,19 +141,11 @@ func main() {
 	}
 
 	name, err := weave.PeerNameFromUserInput(routerName)
-	if err != nil {
-		log.Fatal(err)
-	}
+	checkFatal(err)
 
 	if nickName == "" {
 		nickName, err = os.Hostname()
-		if err != nil {
-			log.Fatal(err)
-		}
-	}
-
-	if password == "" {
-		password = os.Getenv("WEAVE_PASSWORD")
+		checkFatal(err)
 	}
 
 	if password == "" {
@@ -133,7 +162,6 @@ func main() {
 		defer profile.Start(&p).Stop()
 	}
 
-	config.InterHost = weave.NewSleeveInterHost(config.Port)
 	config.LogFrame = logFrameFunc(pktdebug)
 
 	router := weave.NewRouter(config, name, nickName)
@@ -195,17 +223,13 @@ func logFrameFunc(debug bool) weave.LogFrameFunc {
 
 func initiateConnections(router *weave.Router, peers []string) {
 	for _, peer := range peers {
-		if err := router.ConnectionMaker.InitiateConnection(peer); err != nil {
-			log.Fatal(err)
-		}
+		checkFatal(router.ConnectionMaker.InitiateConnection(peer))
 	}
 }
 
 func createAllocator(router *weave.Router, apiPath string, iprangeCIDR string, quorum uint) *ipam.Allocator {
 	allocator, err := ipam.NewAllocator(router.Ourself.Peer.Name, router.Ourself.Peer.UID, router.Ourself.Peer.NickName, iprangeCIDR, quorum)
-	if err != nil {
-		log.Fatal(err)
-	}
+	checkFatal(err)
 	allocator.SetInterfaces(router.NewGossip("IPallocation", allocator))
 	allocator.Start()
 	err = updater.Start(apiPath, allocator)
@@ -288,3 +312,9 @@ func handleHTTP(router *weave.Router, httpAddr string, allocator *ipam.Allocator
 		log.Fatal("Unable to create http server", err)
 	}
 }
+
+func checkFatal(e error) {
+	if e != nil {
+		log.Fatal(e)
+	}
+}