Small fixes on server and client deb package

Makefile

@@ -76,11 +76,9 @@ dataplane/proto/sai/%.grpc.pb.o:  dataplane/proto/sai/%.grpc.pb.cc
 dataplane/standalone/sai/%.o: dataplane/standalone/sai/%.cc $(PROTO_SRC)
 	g++ -fPIC -c $< -o $@ -I . -I external/com_github_opencomputeproject_sai -I external/com_github_opencomputeproject_sai/inc  -I external/com_github_opencomputeproject_sai/experimental
 
-packetio:
-	go build -o dataplane/standalone/packetio/packetio.a -buildmode=c-archive ./dataplane/standalone/packetio 
 
-libsai.so: $(PROTO_OBJ) $(GRPC_OBJ) $(SAI_OBJ) packetio
-	g++ -fPIC -o libsai.so -shared dataplane/standalone/entrypoint.cc dataplane/proto/sai/*.o dataplane/standalone/sai/*.o dataplane/standalone/packetio/packetio.a -lglog -lprotobuf -lgrpc++ -I . -I external/com_github_opencomputeproject_sai -I external/com_github_opencomputeproject_sai/inc -I external/com_github_opencomputeproject_sai/experimental
+libsai.so: $(PROTO_OBJ) $(GRPC_OBJ) $(SAI_OBJ)
+	g++ -fPIC -o libsai.so -shared dataplane/standalone/entrypoint.cc dataplane/proto/sai/*.o dataplane/standalone/sai/*.o -lglog -lprotobuf -lgrpc++ -I . -I external/com_github_opencomputeproject_sai -I external/com_github_opencomputeproject_sai/inc -I external/com_github_opencomputeproject_sai/experimental
 
 define DEB_CONTROL =
 Package: lucius-libsai

cloudbuild/presubmit.yaml

@@ -19,7 +19,7 @@ steps:
       - USERNAME=user
       - SSH_ARGS=--internal-ip --ssh-key-expire-after=1d
       - INSTANCE_NAME=kne-presubmit-$BUILD_ID
-      - INSTANCE_ARGS=--network cloudbuild-workers --image-project gep-kne --image-family kne --machine-type e2-standard-16 --scopes=cloud-platform
+      - INSTANCE_ARGS=--network cloudbuild-workers --image-project gep-kne --image-family kne --machine-type e2-standard-32 --scopes=cloud-platform
       - ZONE=us-central1-a
       - REMOTE_WORKSPACE=/tmp/workspace
       - COMMAND=sh -c "BUILD=$BUILD_ID /tmp/workspace/cloudbuild/presubmit.sh"

dataplane/dplaneopts/dplaneopts.go

@@ -91,13 +91,20 @@ func WithPortMap(m map[string]string) Option {
 
 // WithEthDevAsLane enables treating ethX and hardware lane X.
 // If a port is created with multiple lanes only the first is used.
-// Default: none
+// Default: false
 func WithEthDevAsLane(enable bool) Option {
 	return func(o *Options) {
 		o.EthDevAsLane = enable
 	}
 }
 
+// WithEthDevAsLane enables sending all packets from/to the CP port over gRPC
+func WithRemoteCPUPort(enable bool) Option {
+	return func(o *Options) {
+		o.RemoteCPUPort = enable
+	}
+}
+
 // Port contains configuration data for a single port.
 type Port struct {
 	Lanes string `json:"lanes"`

dataplane/forwarding/fwd.go

@@ -925,41 +925,63 @@ func (e *Server) CPUPacketStream(srv fwdpb.Forwarding_CPUPacketStreamServer) err
 	if err != nil {
 		return err
 	}
-	ctx, err := e.FindContext(init.GetContextId())
+	fwdCtx, err := e.FindContext(init.GetContextId())
 	if err != nil {
 		return fmt.Errorf("failed to get context, err %v", err)
 	}
 
+	fwdCtx.RLock()
 	cpuPortID := ""
-	for _, id := range ctx.Objects.IDs() {
-		obj, err := ctx.Objects.FindID(&fwdpb.ObjectId{Id: string(id)})
+	for _, id := range fwdCtx.Objects.IDs() {
+		obj, err := fwdCtx.Objects.FindID(&fwdpb.ObjectId{Id: string(id)})
 		if err != nil {
+			fwdCtx.RUnlock()
 			return err
 		}
 		if _, ok := obj.(*ports.CPUPort); ok {
 			cpuPortID = string(obj.ID())
 		}
 	}
+	fwdCtx.RUnlock()
 	if cpuPortID == "" {
 		return fmt.Errorf("couldn't find cpu port")
 	}
 
-	ctx.SetCPUPortSink(func(po *fwdpb.PacketOut) error {
-		return srv.Send(po)
-	})
+	packetCh := make(chan *fwdpb.PacketIn)
+	ctx, cancel := context.WithCancel(srv.Context())
 
-	for {
-		pkt, err := srv.Recv()
-		if err != nil {
-			continue
+	// Since Recv() is blocking and we want this func to return immediately on cancel.
+	// Run the Recv in a seperate goroutine.
+	go func() {
+		for {
+			pkt, err := srv.Recv()
+			if err != nil {
+				continue
+			}
+			packetCh <- pkt
 		}
-		acts := []*fwdpb.ActionDesc{fwdconfig.Action(fwdconfig.UpdateAction(fwdpb.UpdateType_UPDATE_TYPE_SET, fwdpb.PacketFieldNum_PACKET_FIELD_NUM_HOST_PORT_ID).
-			WithUint64Value(pkt.GetPacket().GetHostPort())).Build()}
+	}()
 
-		err = e.injectPacket(init.GetContextId(), &fwdpb.PortId{ObjectId: &fwdpb.ObjectId{Id: cpuPortID}}, fwdpb.PacketHeaderId_PACKET_HEADER_ID_ETHERNET,
-			pkt.GetPacket().GetFrame(), acts, true, fwdpb.PortAction_PORT_ACTION_INPUT)
-		if err != nil {
-			continue
+	fn := func(po *fwdpb.PacketOut) error {
+		return srv.Send(po)
+	}
+
+	fwdCtx.Lock()
+	fwdCtx.SetCPUPortSink(fn, cancel)
+	fwdCtx.Unlock()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case pkt := <-packetCh:
+			acts := []*fwdpb.ActionDesc{fwdconfig.Action(fwdconfig.UpdateAction(fwdpb.UpdateType_UPDATE_TYPE_SET, fwdpb.PacketFieldNum_PACKET_FIELD_NUM_HOST_PORT_ID).
+				WithUint64Value(pkt.GetPacket().GetHostPort())).Build()}
+			err = e.injectPacket(init.GetContextId(), &fwdpb.PortId{ObjectId: &fwdpb.ObjectId{Id: cpuPortID}}, fwdpb.PacketHeaderId_PACKET_HEADER_ID_ETHERNET,
+				pkt.GetPacket().GetFrame(), acts, true, fwdpb.PortAction_PORT_ACTION_INPUT)
+			if err != nil {
+				continue
+			}
 		}
 	}
 }
@@ -969,28 +991,40 @@ func (e *Server) HostPortControl(srv fwdpb.Forwarding_HostPortControlServer) err
 	if err != nil {
 		return err
 	}
-	ctx, err := e.FindContext(init.GetContextId())
+	fwdCtx, err := e.FindContext(init.GetContextId())
 	if err != nil {
 		return fmt.Errorf("failed to get context, err %v", err)
 	}
 	reqCh := make(chan *fwdpb.HostPortControlMessage)
 	respCh := make(chan *fwdpb.HostPortControlRequest)
+	ctx, cancel := context.WithCancel(srv.Context())
 	defer close(respCh)
 
-	ctx.SetPortControl(func(msg *fwdpb.HostPortControlMessage) error {
+	fn := func(msg *fwdpb.HostPortControlMessage) error {
 		reqCh <- msg
 		resp := <-respCh
 		return status.FromProto(resp.GetStatus()).Err()
-	})
+	}
+	fwdCtx.Lock()
+	fwdCtx.SetPortControl(fn, cancel)
+	fwdCtx.Unlock()
+	log.Info("initialized host port control channel")
+
 	for {
-		req := <-reqCh
-		if err := srv.Send(req); err != nil {
-			return err
-		}
-		resp, err := srv.Recv()
-		if err != nil {
-			return err
+		select {
+		case <-ctx.Done():
+			return nil
+		case req := <-reqCh:
+			if err := srv.Send(req); err != nil {
+				return err
+			}
+			log.Info("sent message to client: %+v", req)
+			resp, err := srv.Recv()
+			if err != nil {
+				return err
+			}
+			respCh <- resp
+			log.Info("received message from client: %+v", req)
 		}
-		respCh <- resp
 	}
 }

dataplane/forwarding/infra/fwdcontext/context.go

@@ -78,7 +78,9 @@ type Context struct {
 	// FakePortManager is the implementation of the port creator for the Fake port type.
 	FakePortManager FakePortManager
 	portCtl         PortControl
+	portCtlDone     func()
 	cpuPortSink     CPUPortSink
+	cpuPortSinkDone func()
 }
 
 type PortControl func(*fwdpb.HostPortControlMessage) error
@@ -189,8 +191,9 @@ func (ctx *Context) PacketSink() PacketCallback {
 }
 
 // SetPacketSink sets the port control service for the context
-func (ctx *Context) SetPortControl(fn PortControl) error {
+func (ctx *Context) SetPortControl(fn PortControl, doneFn func()) error {
 	ctx.portCtl = fn
+	ctx.portCtlDone = doneFn
 	return nil
 }
 
@@ -200,8 +203,9 @@ func (ctx *Context) PortControl() PortControl {
 }
 
 // SetCPUPortSink sets the port control service for the context
-func (ctx *Context) SetCPUPortSink(fn CPUPortSink) error {
+func (ctx *Context) SetCPUPortSink(fn CPUPortSink, doneFn func()) error {
 	ctx.cpuPortSink = fn
+	ctx.cpuPortSinkDone = doneFn
 	return nil
 }
 
@@ -218,6 +222,15 @@ func (ctx *Context) Cleanup(ch chan bool, isPort func(*fwdpb.ObjectId) bool) {
 	ctx.SetPacketSink(nil)
 	ctx.SetNotification(nil)
 
+	if ctx.cpuPortSinkDone != nil {
+		ctx.cpuPortSinkDone()
+	}
+	if ctx.portCtlDone != nil {
+		ctx.portCtlDone()
+	}
+	ctx.SetCPUPortSink(nil, nil)
+	ctx.SetPortControl(nil, nil)
+
 	ids := ctx.Objects.IDs()
 
 	// First remove the ports.

dataplane/internal/kernel/tap.go

@@ -26,7 +26,8 @@ func NewTap(name string) (*TapInterface, error) {
 		LinkAttrs: netlink.LinkAttrs{
 			Name: name,
 		},
-		Mode: netlink.TUNTAP_MODE_TAP,
+		Mode:   netlink.TUNTAP_MODE_TAP,
+		Queues: 1,
 	}
 	if err := netlink.LinkAdd(tap); err != nil {
 		return nil, err

dataplane/saiserver/hostif.go

@@ -28,6 +28,8 @@ import (
 	"github.com/openconfig/lemming/dataplane/forwarding/fwdconfig"
 	"github.com/openconfig/lemming/dataplane/saiserver/attrmgr"
 
+	log "github.com/golang/glog"
+
 	saipb "github.com/openconfig/lemming/dataplane/proto/sai"
 	fwdpb "github.com/openconfig/lemming/proto/forwarding"
 )
@@ -262,7 +264,7 @@ func (hostif *hostif) createRemoteHostif(ctx context.Context, req *saipb.CreateH
 		}
 		entry := fwdconfig.TableEntryAddRequest(hostif.dataplane.ID(), hostifToPortTable).
 			AppendEntry(fwdconfig.EntryDesc(fwdconfig.ExactEntry(fwdconfig.PacketFieldBytes(fwdpb.PacketFieldNum_PACKET_FIELD_NUM_HOST_PORT_ID).WithUint64(id))),
-				fwdconfig.Action(fwdconfig.TransmitAction(fmt.Sprint(req.ObjId)))).Build()
+				fwdconfig.Action(fwdconfig.TransmitAction(fmt.Sprint(req.GetObjId())))).Build()
 
 		if req.GetObjId() == resp.GetAttr().GetCpuPort() {
 			entry.Entries[0].Actions = getForwardingPipeline()
@@ -311,16 +313,22 @@ func (hostif *hostif) createRemoteHostif(ctx context.Context, req *saipb.CreateH
 			Port:          portReq.GetPort(),
 			DataplanePort: &fwdpb.PortId{ObjectId: &fwdpb.ObjectId{Id: fmt.Sprint(req.GetObjId())}},
 		}
+		log.Infof("sending port ctl message: %+v", ctlReq)
 		if err := ctl(ctlReq); err != nil {
 			return nil, err
 		}
+	} else {
+		log.Warning("didn't send port control message: channel is nil")
 	}
 	return &saipb.CreateHostifResponse{Oid: id}, nil
 }
 
 // SetHostifAttribute sets the attributes in the request.
 func (hostif *hostif) SetHostifAttribute(ctx context.Context, req *saipb.SetHostifAttributeRequest) (*saipb.SetHostifAttributeResponse, error) {
 	if req.OperStatus != nil {
+		if hostif.opts.RemoteCPUPort {
+			return nil, nil
+		}
 		stateReq := &fwdpb.PortStateRequest{
 			ContextId: &fwdpb.ContextId{Id: hostif.dataplane.ID()},
 			PortId:    &fwdpb.PortId{ObjectId: &fwdpb.ObjectId{Id: fmt.Sprint(req.GetOid())}},

dataplane/standalone/BUILD

@@ -9,7 +9,6 @@ cc_library(
     visibility = ["//visibility:public"],
     deps = [
         "//dataplane/proto/sai:sai_cc_grpc",
-        "//dataplane/standalone/packetio",
         "//dataplane/standalone/sai",
         "@com_github_google_glog//:glog",
         "@com_github_grpc_grpc//:grpc++",

dataplane/standalone/entrypoint.cc

@@ -71,7 +71,6 @@
 #include "dataplane/proto/sai/virtual_router.grpc.pb.h"
 #include "dataplane/proto/sai/vlan.grpc.pb.h"
 #include "dataplane/proto/sai/wred.grpc.pb.h"
-#include "dataplane/standalone/packetio/packetio.h"
 #include "dataplane/standalone/sai/acl.h"
 #include "dataplane/standalone/sai/bfd.h"
 #include "dataplane/standalone/sai/bmtor.h"
@@ -257,7 +256,6 @@ sai_status_t sai_api_initialize(
     LOG(ERROR) << status.error_message();
     return SAI_STATUS_FAILURE;
   }
-  StartSink();
 
   return SAI_STATUS_SUCCESS;
 }

dataplane/standalone/lucius/lucius.go

@@ -39,6 +39,7 @@ var (
 	configFile    = flag.String("config_file", "", "Path to config file")
 	portMapString = flag.String("port_map", "", "Map of modeled port names to Linux interface to  as comma seperated list (eg Ethernet8:eth1,Ethernet10,eth2)")
 	ethDevAsLane  = flag.Bool("eth_dev_as_lane", false, "If true, when creating ports, use ethX and hardware lane X")
+	remoteCPUPort = flag.Bool("remote_cpu_port", false, "If true, send all packets from/to the CPU port over gRPC")
 )
 
 func main() {
@@ -89,6 +90,7 @@ func start(port int) {
 		dplaneopts.WithPortConfigFile(*configFile),
 		dplaneopts.WithPortMap(portMap),
 		dplaneopts.WithEthDevAsLane(*ethDevAsLane),
+		dplaneopts.WithRemoteCPUPort(*remoteCPUPort),
 	)
 
 	if _, err := saiserver.New(context.Background(), mgr, srv, opts); err != nil {

dataplane/standalone/pkthandler/BUILD

@@ -0,0 +1,67 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_binary", "go_library")
+load("@rules_pkg//pkg:deb.bzl", "pkg_deb")
+load("@rules_pkg//pkg:tar.bzl", "pkg_tar")
+load("@rules_pkg//pkg:mappings.bzl", "pkg_attributes", "pkg_files")
+
+go_library(
+    name = "pkthandler_lib",
+    srcs = ["main.go"],
+    importpath = "github.com/openconfig/lemming/dataplane/standalone/pkthandler",
+    visibility = ["//visibility:private"],
+    deps = [
+        "//dataplane/standalone/pkthandler/pktiohandler",
+        "//proto/forwarding",
+        "@com_github_golang_glog//:glog",
+        "@org_golang_google_grpc//:go_default_library",
+        "@org_golang_google_grpc//credentials/insecure",
+    ],
+)
+
+go_binary(
+    name = "pkthandler",
+    embed = [":pkthandler_lib"],
+    visibility = ["//visibility:public"],
+)
+
+pkg_files(
+    name = "pkthandler-bin",
+    srcs = [
+        ":pkthandler",
+    ],
+    attributes = pkg_attributes(
+        mode = "0755",
+    ),
+    prefix = "/usr/bin",
+)
+
+pkg_files(
+    name = "pkthandler-service",
+    srcs = [
+        "lucius-pkthandler.service",
+    ],
+    prefix = "/etc/systemd/system/",
+)
+
+pkg_tar(
+    name = "pkthandler-tar",
+    srcs = [
+        ":pkthandler-bin",
+        ":pkthandler-service",
+    ],
+    tags = ["manual"],
+)
+
+pkg_deb(
+    name = "pkthandler-deb",
+    architecture = "amd64",
+    data = ":pkthandler-tar",
+    description = "Packet Handler for lucius",
+    maintainer = "OpenConfig",
+    package = "lucius-pkthandler",
+    postinst = "postinst",
+    postrm = "postrm",
+    preinst = "preinst",
+    prerm = "prerm",
+    tags = ["manual"],
+    version = "1.0-0",
+)

dataplane/standalone/pkthandler/lucius-pkthandler.service

@@ -0,0 +1,12 @@
+[Unit]
+Description=Lucius Packet Handler
+Wants=network-online.target
+After=network-online.target
+
+[Service]
+Type=exec
+ExecStart=/usr/bin/pkthandler --alsologtostderr
+Restart=always
+
+[Install]
+WantedBy=multi-user.target