Adding preliminary modules for dispatching at NIC

dispatch_system/dispatch_daemon/config.json

-{"id":"192.168.0.106","master_node":"192.168.0.105"}
\ No newline at end of file
+{"id":"10.129.2.201","master_node":"192.168.0.105"}
\ No newline at end of file

dispatch_system/dispatch_daemon/execute.js

@@ -97,19 +97,32 @@ function runContainer(metadata) {
                     else {
                         let process = null;
                         if (constants.network.use_bridge)
-                            process = spawn('docker', ["run", "--rm", `--network=${constants.network.network_bridge}`, "-p", `${port}:${port}`, 
-                                "--name", resource_id, registry_url + imageName,
+                            process = spawn('docker', ["create", "--rm", `--network=${constants.network.network_bridge}`, "-p", `${port}:${port}`, 
+                                "-p", `${port}:${port}/udp`, "--name", resource_id, registry_url + imageName,
                                 resource_id, imageName, port, "container", constants.network.internal.kafka_host]);
                         else
-                            process = spawn('docker', ["run", "--rm", "-p", `${port}:${port}`,
-                                "--name", resource_id, registry_url + imageName,
+                            process = spawn('docker', ["create", "--rm", "-p", `${port}:${port}`,
+                                "-p", `${port}:${port}/udp`, "--name", resource_id, registry_url + imageName,
                                 resource_id, imageName, port, "container", constants.network.internal.kafka_host]);
+                        
                         let result = "";
                         // timeStart = Date.now()
                         process.stdout.on('data', (data) => {
                             logger.info(`stdout: ${data}`);
                             let timeDifference = Math.ceil((Date.now() - timeStart))
                             logger.info("container run time taken: ", timeDifference);
+                            let add_network = spawn('docker', ['network', 'connect', 'pub_net', resource_id])
+                            let _ = spawn('docker', ['start', resource_id])
+
+                            _.on('data', (data) => {
+                                console.log("container started", data);
+                            })
+                            // add_network.stderr.on('data', (data) => {
+                            //     // console.log("network add error", data);
+                            // })
+                            add_network.on('close', (code) => {
+                                logger.info("Ran command");
+                            })
                             result += data;
                             resolve(resource_id);
                         });
@@ -129,21 +142,43 @@ function runContainer(metadata) {
             } else {
                     logger.info("container starting at port", port);
                     let process = null;
+                    /**
+                     * create docker on the default bridge
+                     */
                     if (constants.network.use_bridge)
-                        process = spawn('docker', ["run", "--rm", `--network=${constants.network.network_bridge}`, 
-                                "-p", `${port}:${port}`, "--name", resource_id, 
+                        process = spawn('docker', ["create", "--rm", `--network=${constants.network.network_bridge}`, 
+                            "-p", `${port}:${port}`, "-p", `${port}:${port}/udp`, "--name", resource_id, 
                             registry_url + imageName, resource_id, imageName, port, "container", constants.network.internal.kafka_host]);
                     else
-                        process = spawn('docker', ["run", "--rm",
-                            "-p", `${port}:${port}`, "--name", resource_id,
+                        process = spawn('docker', ["create",
+                            "-p", `${port}:${port}`, "-p", `${port}:${port}/udp`, "--name", resource_id,
                             registry_url + imageName, resource_id, imageName, port, "container", constants.network.internal.kafka_host]);
                     let result = "";
                     // timeStart = Date.now()
                     process.stdout.on('data', (data) => {
-                        logger.info(`stdout: ${data}`);
+                        logger.info(`stdout: ${data.toString()}`);
                         let timeDifference = Math.ceil((Date.now() - timeStart))
                         logger.info("container run time taken: ", timeDifference);
-                        resolve(resource_id);
+                        /**
+                         * attach smartnic interface
+                         */
+                        let add_network = spawn('docker', ['network', 'connect', 'pub_net', resource_id])
+                        let _ = spawn('docker', ['start', resource_id])
+                        
+                        _.stdout.on('data', (data) => {
+                            logger.info(data.toString())
+                            resolve(resource_id);
+                        })
+
+                        _.stderr.on('data', (data) => {
+                            logger.info(data.toString())
+                        })
+
+                        _.on('close', (data) => {
+                            logger.info("exit exit")
+                            logger.info(data.toString())
+                        })
+                        
                     });
 
                     process.stderr.on('data', (data) => {

dispatch_system/dispatch_daemon/lib.js

@@ -21,7 +21,7 @@ function updateConfig() {
     data = data.substr(0, data.indexOf("\n")).trim()
     data = data.split(' ')
     
-    file.id = data[data.length - 3]
+    file.id = data[data.length - 1]
     fs.writeFileSync('./config.json', JSON.stringify(file));
     console.log("Updated Config file");   
 }
@@ -135,4 +135,4 @@ const logger = winston.createLogger({
 
 module.exports = {
     download, makeid, updateConfig, makeTopic, returnPort, logger
-}
\ No newline at end of file
+}

dispatch_system/dispatch_manager/index.js

@@ -295,6 +295,15 @@ function postDeploy(message) {
         
         return;
     }
+
+    /**
+     * IP changes in case MACVLAN is used to connect worker endpoints
+     */
+    if (resourceMap.has(message.resource_id)) {
+        let resource = resourceMap.get(message.resource_id)
+        resource.node_id = message.node_id.trim()
+    }
+
     if (functionToResource.has(id)) {
         let resourceHeap = functionToResource.get(id)
         heap.push(resourceHeap, {
@@ -621,6 +630,6 @@ async function speculative_deployment(req, runtime) {
     }
 }
 setInterval(libSupport.metrics.broadcastMetrics, 5000)
-setInterval(autoscalar, 1000);
+// setInterval(autoscalar, 1000);
 setInterval(dispatch, 1000);
 app.listen(port, () => logger.info(`Server listening on port ${port}!`))
\ No newline at end of file

jspack @ 4753fb1a

+Subproject commit 4753fb1a8deb8e2871d34dda049cb01958137f6b

dispatch_system/dispatch_manager/lib.js

@@ -75,7 +75,7 @@ function generateExecutor(functionPath, functionHash) {
   * @param {JSON} res Object to use to return the response to the user
   */
 async function reverseProxy(req, res) {
-   if (req.headers['x-chain-type'] !== 'explicit')
+   if (req.headers['x-chain-type'] !== 'explicit' && req.body.type === "tcp")
       branchChainPredictor(req)
    let runtime = req.body.runtime
    let id = req.params.id + runtime
@@ -140,7 +140,7 @@ async function reverseProxy(req, res) {
       catch (err) {
          res.json(err.message).status(err.statusCode)
          forwardTo.open_request_count -= 1
-         heap.heapify(functionHeap, compare)
+         heap.heapify(functionHeap, compare) 
          logger.error("error" + err)
       }
    } else if (req.body.type === "udp") {
@@ -148,11 +148,11 @@ async function reverseProxy(req, res) {
       req.body.request_id = request_id
       // res.request_id = request_id
       requestFlightQueue.set(request_id, res)
-      let payload = req.body.payload
+      let payload = req.body
       payload.request_id = request_id
       payload = JSON.stringify(payload)
       udpProxy.send(payload, 0, payload.length, resource.port, resource.node_id, function (err, bytes) {
-         logger.info("forwarded request via UDP")
+         // logger.info(`forwarded request via UDP, IP 192.168.2.5 Port ${resource.port}`)
       })
    }
 }

dispatch_system/dispatch_manager/nic/RTEInterface.py

+#
+# Copyright (C) 2015-2017,  Netronome Systems, Inc.  All rights reserved.
+#
+
+import os, sys, struct, pprint, threading
+from urlparse import urlparse
+from contextlib import contextmanager
+
+from RTERPCInterface import *
+
+class RTEInterfaceConnection(object):
+    def __init__(self):
+        self.transport = None
+
+        self.Design = None
+        self.Counters = None
+        self.Tables = None
+        self.ParserValueSets = None
+        self.Registers = None
+        self.Meters = None
+        self.TrafficClass = None
+        self.Digests = None
+        self.Multicast = None
+        self.DebugCtl = None
+        self.System = None
+
+        self._DoConnect = None
+        self._DoDisconnect = None
+
+    def Connect(self, rpc, host, port, device_id=0, use_zlib=True, serialise_api=False):
+        if rpc == 'thrift':
+            from RTEThriftInterface import (Design, Counters, Tables, ParserValueSets, 
+                Registers, Meters, TrafficClass, Digests, Multicast, DebugCtl, System)
+            from RTEThriftInterface import RteReturnHandler, DoConnect, DoDisconnect
+
+            global P4CounterType, RegisterType
+            from  RTEThriftInterface import P4CounterType, RegisterType
+            
+        elif rpc == 'grpc':
+            from RTEGRPCInterface import (Design, Counters, Tables, ParserValueSets, 
+                Registers, Meters, TrafficClass, Digests, Multicast, DebugCtl, System)
+            from RTEGRPCInterface import (RteReturnHandler, DoConnect, DoDisconnect)
+        else:
+            raise Exception, 'unsupported rpc: %s'%rpc
+
+        DoConnect(self, host, port, device_id, use_zlib, serialise_api)
+
+        self.Design = Design(self)
+        self.Counters = Counters(self)
+        self.Tables = Tables(self)
+        self.ParserValueSets = ParserValueSets(self)
+        self.Registers = Registers(self)
+        self.Meters = Meters(self)
+        self.TrafficClass = TrafficClass(self)
+        self.Digests = Digests(self)
+        self.Multicast = Multicast(self)
+        self.DebugCtl = DebugCtl(self)
+        self.System = System(self)
+        
+        self._DoConnect = DoConnect
+        self._DoDisconnect = DoDisconnect
+                
+        status = self.Design.LoadStatus()
+        if rpc == 'grpc' and status['is_loaded']:
+            self.Design.RefreshP4Schema()
+
+    def Disconnect(self):
+        if self._DoDisconnect is not None:
+            self._DoDisconnect(self)
+                
+    @contextmanager    
+    def ConnectCtx(self, rpc, host, port, device_id=0, use_zlib=True, serialise_api=False):
+        try:
+            self.Connect(rpc, host, port, device_id, use_zlib, serialise_api)
+            yield self
+        finally:
+            self.Disconnect()
+
+    def Shutdown(self):
+        self.System.Shutdown()
+        self.Disconnect()
+
+RTEInterface = RTEInterfaceConnection()

dispatch_system/dispatch_manager/nic/RTERPCInterface.py

+#
+# Copyright (C) 2017,  Netronome Systems, Inc.  All rights reserved.
+#
+
+import os, sys, struct, pprint, threading
+from urlparse import urlparse
+from contextlib import contextmanager
+
+class RTEError(Exception): pass
+class RTECommError(RTEError): pass
+class RTERPCNotImplemented(RTEError): pass
+class RTEReturnError(RTEError): pass
+
+RTE_RETURN_CODES = [
+    'SUCCESS',
+    'ERROR',          # error reason string provided
+    'ARGINVALID',     # invalid argument
+    'IOERROR',        # platform IO error
+    'MEMORYERROR',    # memory allocation error
+    'FILEIOERROR',    # file IO error
+    'NOTLOADED',      # firmware not loaded
+    'HWINCOMPATIBLE', # platform doesn't support operation
+]
+    
+# XXX a hack for now, but these should be shared definitions for thrift/grpc
+class LogLevel(object):
+    UNKNOWN = 0
+    DISABLE = 1
+    FATAL = 2
+    ERROR = 3
+    WARN = 4
+    INFO = 5
+    DEBUG = 6
+    EXTRA = 7
+    HEAVY = 8
+
+class P4CounterType(object):
+    Global = 'P4CounterType.Global'
+    Direct = 'P4CounterType.Direct'
+    Static = 'P4CounterType.Static'
+
+class P4CounterClass(object):
+    Invalid     = 'P4CounterClass.Invalid'
+    Unspecified = 'P4CounterClass.Unspecified'
+    Packets     = 'P4CounterClass.Packets'
+    Bytes       = 'P4CounterClass.Bytes'
+    Both        = 'P4CounterClass.Both'
+
+class RegisterType(object):
+    Global = 'RegisterType.Global'
+    Direct = 'RegisterType.Direct'
+    Static = 'RegisterType.Static'
+
+class MeterType(object):
+    Invalid = 'MeterType.Invalid'
+    Global  = 'MeterType.Global'
+    Direct  = 'MeterType.Direct'
+    Static  = 'MeterType.Static'
+
+class MeterClass(object):
+    Invalid     = 'MeterClass.Invalid'
+    Unspecified = 'MeterClass.Unspecified'
+    Packets     = 'MeterClass.Packets'
+    Bytes       = 'MeterClass.Bytes'
+
+class MatchType(object):
+    Unspecified = 'MatchType.Unspecified'
+    Valid       = 'MatchType.Valid'
+    Exact       = 'MatchType.Exact'
+    LPM         = 'MatchType.LPM'
+    Ternary     = 'MatchType.Ternary'
+    Range       = 'MatchType.Range'
+
+
+#class RegisterFieldDesc(object):
+#class RegisterType(object):
+    
+def REV_MAP(m):
+    res = {}
+    for k, v in m.items():
+        res['v'] = k
+    return res
+
+class NullCtx():
+    def __enter__(*args): pass
+    def __exit__(*exc_info): pass    
+
+class RTEModule(object):
+    def __init__(self, rte):
+        self.rte = rte
+
+class DesignBase(RTEModule):
+    def Load(self, elf_fw, pif_design, pif_config):
+        raise RTERPCNotImplemented
+
+    def Unload(self):
+        raise RTERPCNotImplemented
+
+    def ConfigReload(self, pif_config):
+        raise RTERPCNotImplemented
+
+    def LoadStatus(self):
+        raise RTERPCNotImplemented
+        
+class CountersBase(RTEModule):
+    def ResolveToCounterId(self, counter):
+        if isinstance(counter, (int, long)):
+            counter_id = counter
+        elif isinstance(counter, basestring):
+            counter_id = self.GetP4CounterByName(counter)['id']
+        elif isinstance(counter, dict):
+            counter_id = counter['id']
+        else:
+            raise RTEError, "Unexpected counter parameter type %s"%type(counter)
+
+    def GetP4Counters(self):
+        return [(counter, self.GetP4Counter(counter)) for counter in self.ListP4Counters()]
+
+    def GetP4CounterByName(self, counter_name):
+        for counter in self.ListP4Counters():
+            if counter['name'] == counter_name:
+                return counter
+        raise RTEError, "Counter '%s' not found"%counter_name
+
+    def ExtractRteValue(self, rv):
+        raise RTERPCNotImplemented
+
+    def ListP4Counters(self):
+        raise RTERPCNotImplemented
+
+    def GetP4Counter(self, counter):
+        raise RTERPCNotImplemented
+
+    def ClearP4Counter(self, counter):
+        raise RTERPCNotImplemented
+
+    def ClearAllP4Counters(self):
+        raise RTERPCNotImplemented
+        
+    def GetSystemCounters(self):
+        raise RTERPCNotImplemented
+
+    def ClearAllSysCounters(self):
+        raise RTERPCNotImplemented
+
+class TablesBase(RTEModule):
+    def GetTableByName(self, table_name):
+        for table in self.List():
+            if table['tbl_name'] == table_name:
+                return table
+        raise RTEError, "Table '%s' not found"%table_name
+
+    def ResolveToTableId(self, tbl_id):
+        if isinstance(tbl_id, int):
+            return tbl_id
+        elif isinstance(tbl_id, basestring):
+            return self.GetTableByName(tbl_id)['tbl_id']
+        else:
+            raise RTEError, 'Unsupported table name type: %s'%type(tbl_id)
+
+    def AddRule(self, tbl_id, rule_name, default_rule, match, actions, priority = None, timeout = None):
+        raise RTERPCNotImplemented
+
+    def EditRule(self, tbl_id, rule_name, default_rule, match, actions, priority = None, timeout = None):
+        raise RTERPCNotImplemented
+
+    def DeleteRule(self, tbl_id, rule_name, default_rule, match, actions):
+        raise RTERPCNotImplemented
+
+    def List(self):
+        raise RTERPCNotImplemented
+    
+    def ListRules(self, tbl_id):
+        raise RTERPCNotImplemented
+
+    def GetVersion(self):
+        raise RTERPCNotImplemented
+
+class RegistersBase(RTEModule):
+    def GetRegisterByName(self, register_name):
+        for reg in self.List():
+            if reg['name'] == register_name:
+                return reg
+        raise RTEError, "Register '%s' not found"%register_name
+
+    def List(self):
+        raise RTERPCNotImplemented
+    
+    def ResolveToRegisterArrayArg(self, register, index, count):
+        raise RTERPCNotImplemented
+            
+    def Get(self, register, index=0, count=1):
+        raise RTERPCNotImplemented
+            
+    def Clear(self, register, index=0, count=1):
+        raise RTERPCNotImplemented
+
+    def Set(self, register, values, index=0, count=1):
+        raise RTERPCNotImplemented
+
+    def SetField(self, register, field_id, value, index=0, count=1):
+        raise RTERPCNotImplemented
+
+class TrafficClassBase(RTEModule):
+    def Get(self, port_id):
+        raise RTERPCNotImplemented
+            
+    def Set(self, port_id, cfgs):
+        raise RTERPCNotImplemented
+
+    def Commit(self, port_id):
+        raise RTERPCNotImplemented
+
+class MetersBase(RTEModule):
+    def List(self):
+        raise RTERPCNotImplemented
+            
+    def GetConfig(self, meter_id):
+        raise RTERPCNotImplemented
+
+    def SetConfig(self, meter_id, configs):
+        raise RTERPCNotImplemented
+
+class DigestsBase(RTEModule):
+    def List(self):
+        raise RTERPCNotImplemented
+    
+    def Register(self, digest_id):
+        raise RTERPCNotImplemented
+
+    def Deregister(self, digest_regid):
+        raise RTERPCNotImplemented
+    
+    def Get(self, digest_handle):
+        raise RTERPCNotImplemented
+
+class MulticastBase(RTEModule):
+    def List(self):
+        raise RTERPCNotImplemented
+    
+    def SetConfig(self, group_id, ports):
+        raise RTERPCNotImplemented
+
+class SystemBase(RTEModule):
+    def Shutdown(self):
+        raise RTERPCNotImplemented
+
+    def Ping(self):
+        raise RTERPCNotImplemented
+    
+    def Echo(self, echo_msg):
+        raise RTERPCNotImplemented
+
+    def GetVersion(self):
+        raise RTERPCNotImplemented
+
+    def GetLogLevel(self):
+        raise RTERPCNotImplemented
+
+    def SetLogLevel(self, level):
+        raise RTERPCNotImplemented
+
+    def GetPortInfo(self):
+        raise RTERPCNotImplemented
+    
+
+class DebugCtlBase(RTEModule):
+    def Execute(self, debug_id, debug_data):
+        raise RTERPCNotImplemented
+    
+    def SetRuleBreakpoint(self, table_name, rule_name, enable):
+        self.Execute('netro_rule_bpt', 'table %s rule %s enabled %s'%(table_name, rule_name, int(enable)))
+        
+    def GetRuleBreakpoint(self, table_name, rule_name):
+        res = self.Execute('netro_rule_bpt', 'table %s rule %s'%(table_name, rule_name))
+        name, val = res.split(None, 1)
+        assert name=='enabled'
+        return bool(int(val))
+
+    def SetMacConfig(self, nbi0_config, nbi1_config):
+        for (conf_id, conf_json) in (('nbi_mac8_json', nbi0_config), 
+                                     ('nbi_mac9_json', nbi1_config)):
+            if conf_json:
+                with open(conf_json, "rb") as f:
+                    self.Execute(conf_id, f.read())
+
+class ParserValueSetsBase(RTEModule):
+    def List(self):
+        raise RTERPCNotImplemented
+
+    def Clear(self, pvs_id):
+        raise RTERPCNotImplemented
+
+    def Add(self, pvs_id, pvs_entries):
+        raise RTERPCNotImplemented
+
+    def Retrieve(self, pvs_id):
+        raise RTERPCNotImplemented
+

dispatch_system/dispatch_manager/nic/assign_ip.sh

+#!/bin/bash -x
+
+sudo ifconfig vf0_0 192.168.2.2/24 up
+sudo ifconfig vf0_1 192.168.2.3/24 up
+
+# create a MAC VLAN for docker attached to vf0_1
+echo "y" | docker system prune
+docker network create -d macvlan --subnet=192.168.2.0/24 --aux-address="vf0_0=192.168.2.2" --aux-address="vf0_1=192.168.2.3" -o parent=vf0_1 pub_net
+
+# move vf0_0 into its own namespace
+# sudo ip netns exec ns_server ip link set vf0_0 netns 1
+# sudo ip netns delete ns_server
+# sudo ip netns add ns_server
+
+# sudo ip link set vf0_0 netns ns_server
+# sudo ip netns exec ns_server ip addr add dev vf0_0 192.168.2.2/24
+# sudo ip netns exec ns_server ip link set dev vf0_0 up
+
+# sudo ip netns exec ns_server arp -s 192.168.2.3 00:22:22:22:22:22 -i vf0_0
+# sudo ip netns exec ns_server arp -s 192.168.2.4 00:33:33:33:33:33 -i vf0_0
+
+# sudo arp -s 192.168.2.2 00:11:11:11:11:11 -i vf0_1
+# sudo arp -s 192.168.2.4 00:33:33:33:33:33 -i vf0_1
\ No newline at end of file

dispatch_system/dispatch_manager/nic/build_offload.sh

+#! /bin/bash -x
+
+compile_flag=0
+location=$(pwd)
+while getopts 'c' flag; do
+  case "${flag}" in
+    c) compile_flag=1 ;;
+  esac
+done
+
+if [[ $compile_flag -eq 1 ]]
+then
+    # compile the nfp code
+    sudo /opt/netronome/p4/bin/nfp4build -o ./p4src/echo.nffw -p ./p4src/out -4 ./p4src/echo.p4 -l lithium --nfp4c_p4_version 16 --nfp4c_p4_compiler p4c-nfp -c ./p4src/prime.c
+fi
+# move to p4 bin
+cd /opt/netronome/p4/bin/
+
+# #offload
+sudo ./rtecli design-load -f $location/p4src/echo.nffw -c $location/p4src/echo.p4cfg -p $location/p4src/out/pif_design.json
+
+# returning back to base
+cd $location
\ No newline at end of file

dispatch_system/dispatch_manager/nic/digest_listener.py

+#!/usr/bin/env python
+#
+# based on the RunTimeEnvironment-remote example
+# that comes generated by thrift
+#
+import argparse
+import struct
+import sys
+import pprint
+import collections
+import time
+
+sys.path.append('../../out/gen_thrift/py')
+
+from urlparse import urlparse
+from thrift.transport import TTransport
+from thrift.transport import TZlibTransport
+from thrift.transport import TSocket
+#from thrift.transport import TSSLSocket
+from thrift.transport import THttpClient
+from thrift.protocol import TBinaryProtocol
+
+from sdk6_rte import RunTimeEnvironment
+from sdk6_rte.ttypes import *
+    
+if __name__ == '__main__':
+
+    #
+    # Argument handling
+    #
+
+    parser = argparse.ArgumentParser()
+
+    parser.add_argument('-s','--server',
+                        dest='rpc_server', default='localhost',
+                        type=str,
+                        help="Thrift RPC host (DEFAULT: localhost)")
+
+    parser.add_argument('-p','--port',
+                        dest='rpc_port', default='20206',
+                        type=str,
+                        help="Thrift RPC port (DEFAULT: 20206)")
+    
+    parser.set_defaults()
+
+    args = parser.parse_args()
+    
+    use_zlib = 1
+    
+    host = args.rpc_server
+    port = int(args.rpc_port)
+
+    socket = TSocket.TSocket(host, port)
+    transport = TTransport.TBufferedTransport(socket)
+    if use_zlib:
+        transport = TZlibTransport.TZlibTransport(transport)
+
+    protocol = TBinaryProtocol.TBinaryProtocol(transport)
+    client = RunTimeEnvironment.Client(protocol)
+    transport.open()
+
+    try:
+        client.sys_ping()
+    except TException:
+        transport.close()
+        sys.stderr.write("Basic communication with RPC server failed\n")
+        sys.exit(1)
+
+    # grab all the digest info
+    digests = client.digest_list_all()
+
+    # a map for associating registration handle with digest data
+    digest_map = collections.OrderedDict()
+
+    # register for each digest
+    for d in digests:
+
+        # get the digest registration handle
+        dh = client.digest_register(d.id)
+        if dh < 0:
+            sys.stderr.write("Failed to register for digest %s\n" % d.name)
+            sys.exit(1)
+
+        # associate the registration handle with the digest data
+        digest_map[dh] = {'desc' : d, 'count' : 0}
+
+    print "polling for digests events"
+    # okay now periodically retrieve and dump the digest data
+    try:
+        while 1:
+            for dh, dgdata in digest_map.items():
+                values = client.digest_retrieve(dh)
+
+                if len(values) == 0: # no data
+                    continue
+
+                fldcnt = len(dgdata['desc'].fields)
+                if len(values) % fldcnt != 0:
+                    sys.stderr.write("Invalid field layout from digest %s" %
+                                     dgdata['desc'].name)
+                    sys.exit(1)
+                
+                for i in range(len(values) / fldcnt):
+                    print "digest %s (P4 ID %d, P4 fieldlist %s)[%d] {" % (
+                            dgdata['desc'].name,
+                            dgdata['desc'].app_id,
+                            dgdata['desc'].field_list_name,
+                            dgdata['count'])
+
+                    for flddesc, fielddata in zip(dgdata['desc'].fields, values[fldcnt * i:fldcnt * (i + 1)]):
+                        print "    %s : %s" % (flddesc.name, fielddata)
+                    print "}\n"
+
+                    dgdata['count'] += 1
+
+            time.sleep(2)
+    except KeyboardInterrupt: # exit on control-c
+        pass
+
+    transport.close()
+    

dispatch_system/dispatch_manager/nic/p4src/.gitignore

+
+out_dir/
+out/
+*.list

dispatch_system/dispatch_manager/nic/p4src/command_to_offload

+sudo ./pif_rte -n 0 -p 20207 -I -z -s /opt/nfp_pif/scripts/pif_ctl_nfd.sh -f ~/nilanjan/test/offload/p4src/routing_benchmark.nffw -d ~/nilanjan/test/offload/p4src/out_dir/pif_design.json -c ~/nilanjan/test/offload/p4src/echo.p4cfg --log_file /var/log/nfp-sdk6-rte.log

dispatch_system/dispatch_manager/nic/p4src/echo.p4

+#include <core.p4>
+#define V1MODEL_VERSION 20200408
+#include <v1model.p4>
+#include "includes/defines.p4"
+#include "includes/headers.p4"
+#include "includes/parsers.p4"
+
+extern void prime();
+
+control ingress(inout headers hdr, inout metadata meta, inout standard_metadata_t standard_metadata) {
+    
+    @name(".set_egress") action set_egress() {
+
+        prime();
+        
+        bit<32> temp = hdr.ipv4.dstAddr;
+        hdr.ipv4.dstAddr = hdr.ipv4.srcAddr;
+        hdr.ipv4.srcAddr = temp;
+        hdr.udp.dstPort = 9000;
+        bit<48> tempEth = hdr.ethernet.dstAddr;
+        hdr.ethernet.dstAddr = hdr.ethernet.srcAddr;
+        hdr.ethernet.srcAddr = tempEth;
+        standard_metadata.egress_spec = standard_metadata.ingress_port;
+        hdr.ipv4.ttl = hdr.ipv4.ttl - 8w1;
+    }
+
+    @name(".fwd_act") action fwd_act(bit<16> port) {
+        standard_metadata.egress_spec = port;
+    }
+
+    @name(".fwd") table fwd {
+        actions = {
+            fwd_act;
+        }
+        key = {
+            standard_metadata.ingress_port : exact;
+        }
+    }
+
+    @name(".dispatch_act") action dispatch_act(bit<32> dstAddr, bit<16> dstPort) {
+        hdr.ipv4.dstAddr = dstAddr;
+        hdr.udp.dstPort = dstPort;
+    }
+
+    @name(".dispatch") table dispatch {
+        actions = {
+            dispatch_act;
+        }
+        key = {
+            hdr.map_hdr.chain_id : exact;
+        }
+    }
+
+    apply {
+        if (hdr.ipv4.isValid() && hdr.udp.dstPort == MDS_PORT) {
+	        // hdr.map_hdr.data = 32w100;
+            
+            set_egress();
+        } else {
+            if (hdr.ipv4.isValid() && hdr.udp.dstPort == 8080) {
+                dispatch.apply();
+            }
+            fwd.apply();
+        }
+        //fwd.apply();
+    }
+}
+
+control egress(inout headers hdr, inout metadata meta, inout standard_metadata_t standard_metadata) {
+
+    // @name(".ethernet_set_mac_act") action ethernet_set_mac_act(bit<48> smac, bit<48> dmac) {
+    //     hdr.ethernet.srcAddr = smac;
+    //     hdr.ethernet.dstAddr = dmac;
+    // }
+    // @name(".ethernet_set_mac") table ethernet_set_mac {
+    //     actions = {
+    //         ethernet_set_mac_act;
+    //     }
+    //     key = {
+    //         standard_metadata.egress_port: exact;
+    //     }
+    // }
+
+    @name("fix_checksum") action fix_checksum() {
+        hdr.udp.checksum = 16w0;
+    }
+  
+
+    apply {
+        // if (hdr.udp.dstPort == MDS_PORT) {           
+        //     ethernet_set_mac.apply();
+        // }
+        fix_checksum();
+    }
+}
+
+control DeparserImpl(packet_out packet, in headers hdr) {
+    apply {
+        packet.emit<ethernet_t>(hdr.ethernet);
+        packet.emit<ipv4_t>(hdr.ipv4);
+        packet.emit<udp_t>(hdr.udp);
+        packet.emit<map_hdr_t>(hdr.map_hdr);
+    }
+}
+
+control verifyChecksum(inout headers hdr, inout metadata meta) {
+    apply {
+        verify_checksum(
+            hdr.ipv4.isValid(),
+            { hdr.ipv4.version,
+            hdr.ipv4.ihl,
+            hdr.ipv4.diffserv,
+            hdr.ipv4.totalLen,
+            hdr.ipv4.identification,
+            hdr.ipv4.flags,
+            hdr.ipv4.fragOffset,
+            hdr.ipv4.ttl,
+            hdr.ipv4.protocol,
+            hdr.ipv4.srcAddr,
+            hdr.ipv4.dstAddr },
+            hdr.ipv4.hdrChecksum,
+            HashAlgorithm.csum16);
+    }
+}
+
+control computeChecksum(inout headers hdr, inout metadata meta) {
+    apply {
+        update_checksum(
+            hdr.ipv4.isValid(),
+            { hdr.ipv4.version,
+            hdr.ipv4.ihl,
+            hdr.ipv4.diffserv,
+            hdr.ipv4.totalLen,
+            hdr.ipv4.identification,
+            hdr.ipv4.flags,
+            hdr.ipv4.fragOffset,
+            hdr.ipv4.ttl,
+            hdr.ipv4.protocol,
+            hdr.ipv4.srcAddr,
+            hdr.ipv4.dstAddr },
+            hdr.ipv4.hdrChecksum,
+            HashAlgorithm.csum16);
+    }
+}
+
+V1Switch<headers, metadata>(ParserImpl(), verifyChecksum(), ingress(), egress(), computeChecksum(), DeparserImpl()) main;

dispatch_system/dispatch_manager/nic/p4src/ethernet.p4

+action ethernet_set_mac_act (smac, dmac) {
+    modify_field (ethernet.srcAddr, smac);
+    modify_field (ethernet.dstAddr, dmac);
+}
+table ethernet_set_mac {
+    reads {
+        standard_metadata.egress_port: exact;
+    }
+    actions {
+        ethernet_set_mac_act;
+    }
+}

dispatch_system/dispatch_manager/nic/p4src/includes/checksum.p4

+field_list ipv4_field_list {
+    ipv4.version;
+    ipv4.ihl;
+    ipv4.diffserv;
+    ipv4.totalLen;
+    ipv4.identification;
+    ipv4.flags;
+    ipv4.fragOffset;
+    ipv4.ttl;
+    ipv4.protocol;
+    ipv4.srcAddr;
+    ipv4.dstAddr;
+}
+field_list_calculation ipv4_chksum_calc {
+    input {
+        ipv4_field_list;
+    }
+    algorithm : csum16;
+    output_width: 16;
+}
+calculated_field ipv4.hdrChecksum {
+    update ipv4_chksum_calc;
+}
+
+field_list udp_checksum_list {
+    // IPv4 Pseudo Header Format. Must modify for IPv6 support.
+    ipv4.srcAddr;
+    ipv4.dstAddr;
+    8'0;
+    ipv4.protocol;
+    udp.len;
+    udp.srcPort;
+    udp.dstPort;
+    udp.len;
+    // udp.checksum;
+    payload;
+}
+field_list_calculation udp_checksum {
+    input {
+        udp_checksum_list;
+    }
+    algorithm : csum16;
+    output_width : 16;
+}
+calculated_field udp.checksum {
+    update udp_checksum;
+}

dispatch_system/dispatch_manager/nic/p4src/includes/defines.p4

+#define REPLY_PORT 9000
+#define MDS_PORT 8889
+#define NUM_CACHE 128
+#define CLUSTER_COUNT 2
+#define DEAD_EGRESS_PORT 9
+#define CONTROLLER_IP 0x0a000002
+#define SERVER_IP 0x0a000001
+
+#define NC_READ_REQUEST     0
+#define NC_READ_REPLY       1
+#define NC_HOT_READ_REQUEST 2
+#define NC_WRITE_REQUEST    4
+#define NC_WRITE_REPLY      5
+#define NC_UPDATE_REQUEST   8
+#define NC_UPDATE_REPLY     9
+
+#define PKT_INSTANCE_TYPE_NORMAL 0
+#define PKT_INSTANCE_TYPE_INGRESS_CLONE 1
+#define PKT_INSTANCE_TYPE_EGRESS_CLONE 2
+#define PKT_INSTANCE_TYPE_COALESCED 3
+#define PKT_INSTANCE_TYPE_INGRESS_RECIRC 4
+#define PKT_INSTANCE_TYPE_REPLICATION 5
+#define PKT_INSTANCE_TYPE_RESUBMIT 6
\ No newline at end of file