Merge branch 'master' into dan_stateless

2 files changed, 275 insertions, 37 deletions

diff --git a/scripts/automation/trex_control_plane/client_utils/jsonrpc_client.py b/scripts/automation/trex_control_plane/client_utils/jsonrpc_client.py
index ebeec77e..51bb3a14 100755
--- a/scripts/automation/trex_control_plane/client_utils/jsonrpc_client.py
+++ b/scripts/automation/trex_control_plane/client_utils/jsonrpc_client.py
@@ -17,13 +17,37 @@ class bcolors:
     BOLD = '\033[1m'
     UNDERLINE = '\033[4m'
 
+# sub class to describe a batch
+class BatchMessage(object):
+    def __init__ (self, rpc_client):
+        self.rpc_client = rpc_client
+        self.batch_list = []
 
+    def add (self, method_name, params = {}):
+
+        id, msg = self.rpc_client.create_jsonrpc_v2(method_name, params, encode = False)
+        self.batch_list.append(msg)
+
+    def invoke (self, block = False):
+        if not self.rpc_client.connected:
+            return False, "Not connected to server"
+
+        msg = json.dumps(self.batch_list)
+
+        rc, resp_list = self.rpc_client.send_raw_msg(msg, block = False)
+        if len(self.batch_list) == 1:
+            return True, [(rc, resp_list)]
+        else:
+            return rc, resp_list
+
+
+# JSON RPC v2.0 client
 class JsonRpcClient(object):
 
     def __init__ (self, default_server, default_port):
         self.verbose = False
         self.connected = False
-
+        
         # default values
         self.port   = default_port
         self.server = default_server
@@ -63,7 +87,10 @@ class JsonRpcClient(object):
         print "[verbose] " + msg
 
 
-    def create_jsonrpc_v2 (self, method_name, params = {}):
+    def create_batch (self):
+        return BatchMessage(self)
+
+    def create_jsonrpc_v2 (self, method_name, params = {}, encode = True):
         msg = {}
         msg["jsonrpc"] = "2.0"
         msg["method"]  = method_name
@@ -72,21 +99,22 @@ class JsonRpcClient(object):
 
         msg["id"] = self.id_gen.next()
 
-        return id, json.dumps(msg)
-
-    def invoke_rpc_method (self, method_name, params = {}, block = False):
-        rc, msg = self._invoke_rpc_method(method_name, params, block)
-        if not rc:
-            self.disconnect()
+        if encode:
+            return id, json.dumps(msg)
+        else:
+            return id, msg
 
-        return rc, msg
 
-    def _invoke_rpc_method (self, method_name, params = {}, block = False):
+    def invoke_rpc_method (self, method_name, params = {}, block = False):
         if not self.connected:
             return False, "Not connected to server"
 
         id, msg = self.create_jsonrpc_v2(method_name, params)
 
+        return self.send_raw_msg(msg, block)
+
+    
+    def send_raw_msg (self, msg, block = False):
         self.verbose_msg("Sending Request To Server:\n\n" + self.pretty_json(msg) + "\n")
 
         if block:
@@ -95,6 +123,7 @@ class JsonRpcClient(object):
             try:
                 self.socket.send(msg, flags = zmq.NOBLOCK)
             except zmq.error.ZMQError as e:
+                self.disconnect()
                 return False, "Failed To Get Send Message"
 
         got_response = False
@@ -112,22 +141,41 @@ class JsonRpcClient(object):
                     sleep(0.2)
 
         if not got_response:
+            self.disconnect()
             return False, "Failed To Get Server Response"
 
         self.verbose_msg("Server Response:\n\n" + self.pretty_json(response) + "\n")
 
         # decode
+
+        # batch ?
         response_json = json.loads(response)
 
+        if isinstance(response_json, list):
+            rc_list = []
+
+            for single_response in response_json:
+                rc, msg = self.process_single_response(single_response)
+                rc_list.append( (rc, msg) )
+
+            return True, rc_list
+
+        else:
+            rc, msg = self.process_single_response(response_json)
+            return rc, msg
+
+
+    def process_single_response (self, response_json):
+
         if (response_json.get("jsonrpc") != "2.0"):
             return False, "Malfromed Response ({0})".format(str(response))
 
-        if (response_json.get("id") != id):
-            return False, "Server Replied With Bad ID ({0})".format(str(response))
-
         # error reported by server
         if ("error" in response_json):
-            return True, response_json["error"]["message"]
+            if "specific_err" in response_json["error"]:
+                return False, response_json["error"]["specific_err"]
+            else:
+                return False, response_json["error"]["message"]
 
         # if no error there should be a result
         if ("result" not in response_json):
@@ -136,17 +184,7 @@ class JsonRpcClient(object):
         return True, response_json["result"]
 
 
-    def ping_rpc_server(self):
-
-        return self.invoke_rpc_method("ping", block = False)
-
-    def get_rpc_server_status (self):
-        return self.invoke_rpc_method("get_status")
-
-    def query_rpc_server(self):
-        return self.invoke_rpc_method("get_supported_cmds")
-
-
+  
     def set_verbose(self, mode):
         self.verbose = mode
 
@@ -182,12 +220,6 @@ class JsonRpcClient(object):
 
         self.connected = True
 
-        # ping the server
-        rc, err = self.ping_rpc_server()
-        if not rc:
-            self.disconnect()
-            return rc, err
-
         return True, ""
 
 
@@ -205,11 +237,213 @@ class JsonRpcClient(object):
     def is_connected(self):
         return self.connected
 
-
     def __del__(self):
         print "Shutting down RPC client\n"
         if hasattr(self, "context"):
             self.context.destroy(linger=0)
 
-if __name__ == "__main__":
-    pass
+# MOVE THIS TO DAN'S FILE
+class TrexStatelessClient(JsonRpcClient):
+
+    def __init__ (self, server, port, user):
+
+        super(TrexStatelessClient, self).__init__(server, port)
+
+        self.user = user
+        self.port_handlers = {}
+
+        self.supported_cmds  = []
+        self.system_info     = None
+        self.server_version  = None
+        
+
+    def whoami (self):
+        return self.user
+
+    def ping_rpc_server(self):
+
+        return self.invoke_rpc_method("ping", block = False)
+
+    def get_rpc_server_version (self):
+        return self.server_version
+
+    def get_system_info (self):
+        return self.system_info
+
+    def get_supported_cmds(self):
+        return self.supported_cmds
+
+    def get_port_count (self):
+        if not self.system_info:
+            return 0
+
+        return self.system_info["port_count"]
+
+    # refresh the client for transient data
+    def refresh (self):
+
+        # get server versionrc, msg = self.get_supported_cmds()
+        rc, msg = self.invoke_rpc_method("get_version")
+        if not rc:
+            self.disconnect()
+            return rc, msg
+
+        self.server_version = msg
+
+        # get supported commands
+        rc, msg = self.invoke_rpc_method("get_supported_cmds")
+        if not rc:
+            self.disconnect()
+            return rc, msg
+
+        self.supported_cmds = [str(x) for x in msg if x]
+
+        # get system info
+        rc, msg = self.invoke_rpc_method("get_system_info")
+        if not rc:
+            self.disconnect()
+            return rc, msg
+
+        self.system_info = msg
+
+        return True, ""
+
+    def connect (self):
+        rc, err = super(TrexStatelessClient, self).connect()
+        if not rc:
+            return rc, err
+
+        return self.refresh()
+
+
+    # take ownership over ports
+    def take_ownership (self, port_id_array, force = False):
+        if not self.connected:
+            return False, "Not connected to server"
+
+        batch = self.create_batch()
+
+        for port_id in port_id_array:
+            batch.add("acquire", params = {"port_id":port_id, "user":self.user, "force":force})
+
+        rc, resp_list = batch.invoke()
+        if not rc:
+            return rc, resp_list
+
+        for i, rc in enumerate(resp_list):
+            if rc[0]:
+                self.port_handlers[port_id_array[i]] = rc[1]
+
+        return True, resp_list
+
+
+    def release_ports (self, port_id_array):
+        batch = self.create_batch()
+
+        for port_id in port_id_array:
+
+            # let the server handle un-acquired errors
+            if self.port_handlers.get(port_id):
+                handler = self.port_handlers[port_id]
+            else:
+                handler = ""
+
+            batch.add("release", params = {"port_id":port_id, "handler":handler})
+
+
+        rc, resp_list = batch.invoke()
+        if not rc:
+            return rc, resp_list
+
+        for i, rc in enumerate(resp_list):
+            if rc[0]:
+                self.port_handlers.pop(port_id_array[i])
+
+        return True, resp_list
+
+    def get_owned_ports (self):
+        return self.port_handlers.keys()
+
+    # fetch port stats
+    def get_port_stats (self, port_id_array):
+        if not self.connected:
+            return False, "Not connected to server"
+
+        batch = self.create_batch()
+
+        # empty list means all
+        if port_id_array == []:
+            port_id_array = list([x for x in xrange(0, self.system_info["port_count"])])
+
+        for port_id in port_id_array:
+
+            # let the server handle un-acquired errors
+            if self.port_handlers.get(port_id):
+                handler = self.port_handlers[port_id]
+            else:
+                handler = ""
+
+            batch.add("get_port_stats", params = {"port_id":port_id, "handler":handler})
+
+
+        rc, resp_list = batch.invoke()
+
+        return rc, resp_list
+
+    # snapshot will take a snapshot of all your owned ports for streams and etc.
+    def snapshot(self):
+        
+
+        if len(self.get_owned_ports()) == 0:
+            return {}
+
+        snap = {}
+
+        batch = self.create_batch()
+        
+        for port_id in self.get_owned_ports():
+
+            batch.add("get_port_stats", params = {"port_id": port_id, "handler": self.port_handlers[port_id]})
+            batch.add("get_stream_list", params = {"port_id": port_id, "handler": self.port_handlers[port_id]})
+
+        rc, resp_list = batch.invoke()
+        if not rc:
+            return rc, resp_list
+
+        # split the list to 2s
+        index = 0
+        for port_id in self.get_owned_ports():
+            if not resp_list[index] or not resp_list[index + 1]:
+                snap[port_id] = None
+                continue
+
+            # fetch the first two
+            stats = resp_list[index][1]
+            stream_list = resp_list[index + 1][1]
+           
+            port = {}
+            port['status'] = stats['status']
+            port['stream_list'] = []
+
+            # get all the streams
+            if len(stream_list) > 0:
+                batch = self.create_batch()
+                for stream_id in stream_list:
+                    batch.add("get_stream", params = {"port_id": port_id, "stream_id": stream_id, "handler": self.port_handlers[port_id]})
+
+                rc, stream_resp_list = batch.invoke()
+                if not rc:
+                    port = {}
+
+                port['streams'] = {}
+                for i, resp in enumerate(stream_resp_list):
+                    if resp[0]:
+                        port['streams'][stream_list[i]] = resp[1]
+
+            snap[port_id] = port
+
+            # move to next one
+            index += 2
+
+
+        return snap
diff --git a/scripts/automation/trex_control_plane/client_utils/packet_builder.py b/scripts/automation/trex_control_plane/client_utils/packet_builder.py
index 0505d7f1..1c643335 100755
--- a/scripts/automation/trex_control_plane/client_utils/packet_builder.py
+++ b/scripts/automation/trex_control_plane/client_utils/packet_builder.py
@@ -75,6 +75,7 @@ class CTRexPktBuilder(object):
 
             attr: str
                 a string representation of the sub-field to be set:
+
                 + "src" for source
                 + "dst" for destination
 
@@ -84,6 +85,7 @@ class CTRexPktBuilder(object):
 
             ip_type : str
                 a string representation of the IP version to be set:
+
                 + "ipv4" for IPv4
                 + "ipv6" for IPv6
 
@@ -115,6 +117,7 @@ class CTRexPktBuilder(object):
 
             attr: str
                 a string representation of the sub-field to be set:
+
                 + "src" for source
                 + "dst" for destination
 
@@ -227,6 +230,7 @@ class CTRexPktBuilder(object):
             val : int
                 value of attribute.
                 This value will be set "ontop" of the existing value using bitwise "OR" operation.
+
                 .. tip:: It is very useful to use dpkt constants to define the values of these fields.
 
         :raises:
@@ -408,9 +412,9 @@ class CTRexPktBuilder(object):
         trim_size = val_size*2
         hdr_offset, field_abs_offset = self._calc_offset(layer_name, hdr_field, val_size)
         self.vm.add_flow_man_inst(range_name, size=val_size, operation=operation,
-                                  init_value=str(init_val),
-                                  min_value=str(start_val),
-                                  max_value=str(end_val))
+                                  init_value=init_val,
+                                  min_value=start_val,
+                                  max_value=end_val)
         self.vm.add_write_flow_inst(range_name, field_abs_offset)
         self.vm.set_vm_off_inst_field(range_name, "add_value", add_val)
         self.vm.set_vm_off_inst_field(range_name, "is_big_endian", is_big_endian)