Fast datapath

prog/weaver/main.go

@@ -13,6 +13,7 @@ import (
 	"github.com/davecheney/profile"
 	"github.com/docker/docker/pkg/mflag"
 	"github.com/gorilla/mux"
+	"github.com/weaveworks/go-odp/odp"
 
 	. "github.com/weaveworks/weave/common"
 	"github.com/weaveworks/weave/common/docker"
@@ -35,8 +36,13 @@ func main() {
 	runtime.GOMAXPROCS(procs)
 
 	var (
+		// flags that cause immediate exit
+		justVersion          bool
+		createDatapath       bool
+		deleteDatapath       bool
+		addDatapathInterface string
+
 		config                    weave.Config
-		justVersion               bool
 		protocolMinVersion        int
 		ifaceName                 string
 		routerName                string
@@ -60,9 +66,14 @@ func main() {
 		dnsClientTimeout          time.Duration
 		dnsEffectiveListenAddress string
 		iface                     *net.Interface
+		datapathName              string
 	)
 
 	mflag.BoolVar(&justVersion, []string{"#version", "-version"}, false, "print version and exit")
+	mflag.BoolVar(&createDatapath, []string{"-create-datapath"}, false, "create ODP datapath and exit")
+	mflag.BoolVar(&deleteDatapath, []string{"-delete-datapath"}, false, "delete ODP datapath and exit")
+	mflag.StringVar(&addDatapathInterface, []string{"-add-datapath-iface"}, "", "add a network interface to the ODP datapath and exit")
+
 	mflag.IntVar(&config.Port, []string{"#port", "-port"}, weave.Port, "router port")
 	mflag.IntVar(&protocolMinVersion, []string{"-min-protocol-version"}, weave.ProtocolMinVersion, "minimum weave protocol version")
 	mflag.StringVar(&ifaceName, []string{"#iface", "-iface"}, "", "name of interface to capture/inject from (disabled if blank)")
@@ -86,6 +97,7 @@ func main() {
 	mflag.IntVar(&dnsTTL, []string{"-dns-ttl"}, nameserver.DefaultTTL, "TTL for DNS request from our domain")
 	mflag.DurationVar(&dnsClientTimeout, []string{"-dns-fallback-timeout"}, nameserver.DefaultClientTimeout, "timeout for fallback DNS requests")
 	mflag.StringVar(&dnsEffectiveListenAddress, []string{"-dns-effective-listen-address"}, "", "address DNS will actually be listening, after Docker port mapping")
+	mflag.StringVar(&datapathName, []string{"-datapath"}, "", "ODP datapath name")
 
 	// crude way of detecting that we probably have been started in a
 	// container, with `weave launch` --> suppress misleading paths in
@@ -94,14 +106,37 @@ func main() {
 		os.Args[0] = "weave"
 		mflag.CommandLine.Init("weave", mflag.ExitOnError)
 	}
+
 	mflag.Parse()
 
 	peers = mflag.Args()
 
 	SetLogLevel(logLevel)
-	if justVersion {
+
+	switch {
+	case justVersion:
 		fmt.Printf("weave router %s\n", version)
 		os.Exit(0)
+
+	case createDatapath:
+		err := weave.CreateDatapath(datapathName)
+		if odp.IsKernelLacksODPError(err) {
+			// When the kernel lacks ODP support, exit
+			// with a special status to distinguish it for
+			// the weave script.
+			os.Exit(17)
+		}
+
+		checkFatal(err)
+		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())
@@ -112,9 +147,29 @@ func main() {
 	}
 	config.ProtocolMinVersion = byte(protocolMinVersion)
 
-	var err error
+	var fastDPOverlay weave.Overlay
+	if datapathName != "" {
+		// A datapath name implies that "Bridge" and "Overlay"
+		// packet handling use fast datapath, although other
+		// options can override that below.  Even if both
+		// things are overridden, we might need bridging on
+		// the datapath.
+		fastdp, err := weave.NewFastDatapath(weave.FastDatapathConfig{
+			DatapathName: datapathName,
+			Port:         config.Port,
+		})
+
+		checkFatal(err)
+		config.Bridge = fastdp.Bridge()
+		fastDPOverlay = fastdp.Overlay()
+	}
 
 	if ifaceName != "" {
+		// -iface can coexist with -datapath, because
+		// pcap-based packet capture is a bit more efficient
+		// than capture via ODP misses, even when using an
+		// ODP-based bridge.  So when using weave encyption,
+		// it's preferable to use -iface.
 		iface, err := weavenet.EnsureInterface(ifaceName)
 		checkFatal(err)
 
@@ -123,6 +178,27 @@ func main() {
 		checkFatal(err)
 	}
 
+	if password == "" {
+		password = os.Getenv("WEAVE_PASSWORD")
+	}
+
+	if password == "" {
+		Log.Println("Communication between peers is unencrypted.")
+	} else {
+		config.Password = []byte(password)
+		Log.Println("Communication between peers is encrypted.")
+
+		// fastdp doesn't support encryption
+		fastDPOverlay = nil
+	}
+
+	overlays := weave.NewOverlaySwitch()
+	if fastDPOverlay != nil {
+		overlays.Add("fastdp", fastDPOverlay)
+	}
+	overlays.Add("sleeve", weave.NewSleeveOverlay(config.Port))
+	config.Overlay = overlays
+
 	if routerName == "" {
 		if iface == nil {
 			Log.Fatal("Either an interface must be specified with --iface or a name with -name")
@@ -138,16 +214,6 @@ func main() {
 		checkFatal(err)
 	}
 
-	if password == "" {
-		password = os.Getenv("WEAVE_PASSWORD")
-	}
-	if password == "" {
-		Log.Println("Communication between peers is unencrypted.")
-	} else {
-		config.Password = []byte(password)
-		Log.Println("Communication between peers is encrypted.")
-	}
-
 	if prof != "" {
 		p := *profile.CPUProfile
 		p.ProfilePath = prof
@@ -156,7 +222,6 @@ func main() {
 	}
 
 	config.PeerDiscovery = !noDiscovery
-	config.Overlay = weave.NewSleeveOverlay(config.Port)
 
 	if pktdebug {
 		config.PacketLogging = packetLogging{}

router/consts.go

@@ -6,12 +6,16 @@ import (
 )
 
 const (
-	Port             = 6783
-	HTTPPort         = Port + 1
-	MaxUDPPacketSize = 65535
-	ChannelSize      = 16
-	TCPHeartbeat     = 30 * time.Second
-	GossipInterval   = 30 * time.Second
-	MaxDuration      = time.Duration(math.MaxInt64)
-	MaxTCPMsgSize    = 10 * 1024 * 1024
+	Port                = 6783
+	HTTPPort            = Port + 1
+	MaxUDPPacketSize    = 65535
+	ChannelSize         = 16
+	TCPHeartbeat        = 30 * time.Second
+	GossipInterval      = 30 * time.Second
+	MaxDuration         = time.Duration(math.MaxInt64)
+	MaxTCPMsgSize       = 10 * 1024 * 1024
+	FastHeartbeat       = 500 * time.Millisecond
+	SlowHeartbeat       = 10 * time.Second
+	MaxMissedHeartbeats = 6
+	HeartbeatTimeout    = MaxMissedHeartbeats * SlowHeartbeat
 )

router/flow.go

@@ -34,12 +34,18 @@ type MultiFlowOp struct {
 	ops       []FlowOp
 }
 
-func NewMultiFlowOp(broadcast bool) *MultiFlowOp {
-	return &MultiFlowOp{broadcast: broadcast}
+func NewMultiFlowOp(broadcast bool, ops ...FlowOp) *MultiFlowOp {
+	mfop := &MultiFlowOp{broadcast: broadcast}
+	mfop.Add(ops...)
+	return mfop
 }
 
 func (mfop *MultiFlowOp) Add(ops ...FlowOp) {
-	mfop.ops = append(mfop.ops, ops...)
+	for _, op := range ops {
+		if op != nil {
+			mfop.ops = append(mfop.ops, op)
+		}
+	}
 }
 
 func (mfop *MultiFlowOp) Process(frame []byte, dec *EthernetDecoder,
@@ -48,3 +54,23 @@ func (mfop *MultiFlowOp) Process(frame []byte, dec *EthernetDecoder,
 		op.Process(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/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
+		}
+
+		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/sleeve.go

@@ -45,18 +45,14 @@ import (
 // sleeveForwarder.mtu                     <-------------------------->
 
 const (
-	EthernetOverhead    = 14
-	UDPOverhead         = 28 // 20 bytes for IPv4, 8 bytes for UDP
-	DefaultMTU          = 65535
-	FragTestSize        = 60001
-	PMTUDiscoverySize   = 60000
-	FastHeartbeat       = 500 * time.Millisecond
-	SlowHeartbeat       = 10 * time.Second
-	FragTestInterval    = 5 * time.Minute
-	MTUVerifyAttempts   = 8
-	MTUVerifyTimeout    = 10 * time.Millisecond // doubled with each attempt
-	MaxMissedHeartbeats = 6
-	HeartbeatTimeout    = MaxMissedHeartbeats * SlowHeartbeat
+	EthernetOverhead  = 14
+	UDPOverhead       = 28 // 20 bytes for IPv4, 8 bytes for UDP
+	DefaultMTU        = 65535
+	FragTestSize      = 60001
+	PMTUDiscoverySize = 60000
+	FragTestInterval  = 5 * time.Minute
+	MTUVerifyAttempts = 8
+	MTUVerifyTimeout  = 10 * time.Millisecond // doubled with each attempt
 )
 
 type SleeveOverlay struct {