Merge pull request #10 from beba-eu/stats_state_compression

Stats state compression

Author: DavideSanvito

ryu/app/beba/ddos_use_case/simple_monitoring.py

@@ -89,8 +89,7 @@ def _state_stats_reply_handler(self, ev):
 
         if (msg.body.experimenter == 0XBEBABEBA):
             if (msg.body.exp_type == bebaproto.OFPMP_EXP_STATE_STATS_AND_DELETE):
-                data = msg.body.data
-                state_stats_list = bebaparser.OFPStateStats.parser(data, 0)
+                state_stats_list = bebaparser.OFPStateStats.parser(msg, 0)
                 if (state_stats_list != 0):
                     for index in range(len(state_stats_list)):
                         if (state_stats_list[index].entry.state != 0):

ryu/app/beba/ddos_use_case/simple_monitoring_1.py

@@ -477,7 +477,7 @@ def _state_stats_reply_handler(self, ev):
         if (msg.body.experimenter == 0XBEBABEBA):
             if (msg.body.exp_type == bebaproto.OFPMP_EXP_STATE_STATS_AND_DELETE):
                 data = msg.body.data
-                state_stats_list = bebaparser.OFPStateStats.parser(data, 0)
+                state_stats_list = bebaparser.OFPStateStats.parser(msg, 0)
                 if (state_stats_list != 0):
                     self.replies += 1
                     for index in range(len(state_stats_list)):

ryu/app/beba/state_sync/get_flow_state.py

@@ -90,11 +90,11 @@ def switch_features_handler(self, event):
 	@set_ev_cls(ofp_event.EventOFPExperimenterStatsReply, MAIN_DISPATCHER)
 	def packet_in_handler(self, event):
 		msg = event.msg
-		
+
 		# State Sync: Parse the response from the switch
 		if (msg.body.experimenter == 0xBEBABEBA):
 			if (msg.body.exp_type == bebaproto.OFPMP_EXP_STATE_STATS):
-				state_entry_list = bebaparser.OFPStateStats.parser(msg.body.data)
+				state_entry_list = bebaparser.OFPStateStats.parser(msg)
 				for state_entry in state_entry_list:
 					print 'State :',state_entry.entry.state
 					print 'Key   :',bebaparser.state_entry_key_to_str(state_entry)

ryu/app/beba/state_sync/get_flows_in_state.py

@@ -91,7 +91,7 @@ def packet_in_handler(self, event):
 
 		if (msg.body.experimenter == 0xBEBABEBA):
 			if(msg.body.exp_type == bebaproto.OFPMP_EXP_STATE_STATS):
-				state_entry_list = bebaparser.OFPStateStats.parser(msg.body.data)
+				state_entry_list = bebaparser.OFPStateStats.parser(msg)
 				if (state_entry_list==[]):
 					print "No key for this state"
 				else:

ryu/app/beba/tcp_flag_test.py

@@ -80,9 +80,8 @@ def switch_features_handler(self, event):
 
     @set_ev_cls(ofp_event.EventOFPExperimenterStatsReply, MAIN_DISPATCHER)
     def state_stats_reply_handler(self, ev):
-       #Extract buffer 
-        buff = ev.msg.body.data
-        tmp_list = osparser.OFPStateStats.parser(buff,0)    
+       #Extract buffer
+        tmp_list = osparser.OFPStateStats.parser(ev.msg,0)
         for i in tmp_list:
             out_str = "----> ROUND: " + str(self.cnt) + " PAYLOAD:" + str(i) + " STATE: " + str(i.entry.state) + " KEY: " + str(i.entry.key) + "KEY_CNT: " + str(i.entry.key_count)
             LOG.info(out_str)

ryu/ofproto/beba_v1_0.py

@@ -62,14 +62,14 @@
         OFP_STATE_STATS_REQUEST_0_SIZE)
 
 # struct ofp_state_stats
-OFP_STATE_STATS_0_PACK_STR = '!HBxIII'
-OFP_STATE_STATS_0_SIZE = 16
-assert calcsize(OFP_STATE_STATS_0_PACK_STR) == OFP_STATE_STATS_0_SIZE
-OFP_STATE_STATS_ENTRY_SIZE = 56
-OFP_STATE_STATS_1_PACK_STR = '!IIII'
-OFP_STATE_STATS_1_SIZE = 16
-assert calcsize(OFP_STATE_STATS_1_PACK_STR) == OFP_STATE_STATS_1_SIZE
-OFP_STATE_STATS_SIZE = OFP_STATE_STATS_0_SIZE + 4*MAX_FIELD_COUNT + OFP_STATE_STATS_ENTRY_SIZE + OFP_STATE_STATS_1_SIZE
+OFP_STATE_STATS_PACK_STR = '!HBBIIIIIII'
+OFP_STATE_STATS_SIZE = 32
+assert calcsize(OFP_STATE_STATS_PACK_STR) == OFP_STATE_STATS_SIZE
+
+# struct ofp_state_stats_short
+OFP_STATE_STATS_SHORT_PACK_STR = '!HBBI'
+OFP_STATE_STATS_SHORT_SIZE = 8
+assert calcsize(OFP_STATE_STATS_SHORT_PACK_STR) == OFP_STATE_STATS_SHORT_SIZE
 
 #struct ofp_exp_msg_pkttmp_mod
 OFP_EXP_PKTTMP_MOD_PACK_STR='!Bx'
@@ -124,6 +124,8 @@
 OFPMP_EXP_STATE_STATS = 0
 OFPMP_EXP_GLOBAL_STATE_STATS = 1
 OFPMP_EXP_STATE_STATS_AND_DELETE = 2
+OFPMP_EXP_STATE_STATS_SHORT = 3
+OFPMP_EXP_STATE_STATS_AND_DELETE_SHORT = 4
 
 # enum ofp_error_type
 OFPET_EXPERIMENTER = 0xffff

ryu/ofproto/beba_v1_0_parser.py

@@ -87,6 +87,14 @@ def OFPExpMsgKeyExtract(datapath, command, fields, table_id,bit=0):
             data+=struct.pack(field_extract_format,fields[f])
 
     exp_type=bebaproto.OFPT_EXP_STATE_MOD
+
+    # store lookup-scopes
+    if command == bebaproto.OFPSC_EXP_SET_L_EXTRACTOR:
+        if hasattr(datapath, 'lookup_scope'):
+            datapath.lookup_scope[table_id] = fields
+        else:
+            datapath.lookup_scope = {table_id: fields}
+
     return ofproto_parser.OFPExperimenter(datapath=datapath, experimenter=0xBEBABEBA, exp_type=exp_type, data=data)
 
 def OFPExpMsgSetFlowState(datapath, state, keys, table_id, idle_timeout=0, idle_rollback=0, hard_timeout=0, hard_rollback=0, state_mask=0xffffffff):
@@ -140,43 +148,40 @@ def OFPExpResetGlobalState(datapath):
     exp_type=bebaproto.OFPT_EXP_STATE_MOD
     return ofproto_parser.OFPExperimenter(datapath=datapath, experimenter=0xBEBABEBA, exp_type=exp_type, data=data)
 
-def OFPExpStateStatsMultipartRequest(datapath, table_id=ofproto.OFPTT_ALL, state=None, match=None):
-    flags = 0 # Zero or ``OFPMPF_REQ_MORE``
+def OFPExpStateStatsMultipartRequestBase(datapath, exp_type, table_id=ofproto.OFPTT_ALL, state=None, match=None):
+    flags = 0  # Zero or ``OFPMPF_REQ_MORE``
     get_from_state = 1
     if state is None:
         get_from_state = 0
         state = 0
-        
+
     if match is None:
         match = ofproto_parser.OFPMatch()
 
-    data=bytearray()
+    data = bytearray()
     msg_pack_into(bebaproto.OFP_STATE_STATS_REQUEST_0_PACK_STR, data, 0, table_id, get_from_state, state)
-    
-    offset=bebaproto.OFP_STATE_STATS_REQUEST_0_SIZE
+
+    offset = bebaproto.OFP_STATE_STATS_REQUEST_0_SIZE
     match.serialize(data, offset)
 
-    exp_type=bebaproto.OFPMP_EXP_STATE_STATS
-    return ofproto_parser.OFPExperimenterStatsRequest(datapath=datapath, flags=flags, experimenter=0xBEBABEBA, exp_type=exp_type, data=data)
+    return ofproto_parser.OFPExperimenterStatsRequest(datapath=datapath, flags=flags, experimenter=0xBEBABEBA,
+                                                      exp_type=exp_type, data=data)
+
+def OFPExpStateStatsMultipartRequest(datapath, table_id=ofproto.OFPTT_ALL, state=None, match=None):
+    return OFPExpStateStatsMultipartRequestBase(datapath=datapath, exp_type=bebaproto.OFPMP_EXP_STATE_STATS,
+                                                table_id=table_id, state=state, match=match)
 
 def OFPExpStateStatsMultipartRequestAndDelete(datapath, table_id=ofproto.OFPTT_ALL, state=None, match=None):
-    flags = 0 # Zero or ``OFPMPF_REQ_MORE``
-    get_from_state = 1
-    if state is None:
-        get_from_state = 0
-        state = 0
-        
-    if match is None:
-        match = ofproto_parser.OFPMatch()
+    return OFPExpStateStatsMultipartRequestBase(datapath=datapath, exp_type=bebaproto.OFPMP_EXP_STATE_STATS_AND_DELETE,
+                                                table_id=table_id, state=state, match=match)
 
-    data=bytearray()
-    msg_pack_into(bebaproto.OFP_STATE_STATS_REQUEST_0_PACK_STR, data, 0, table_id, get_from_state, state)
-    
-    offset=bebaproto.OFP_STATE_STATS_REQUEST_0_SIZE
-    match.serialize(data, offset)
+def OFPExpStateStatsMultipartRequestShort(datapath, table_id=ofproto.OFPTT_ALL, state=None, match=None):
+    return OFPExpStateStatsMultipartRequestBase(datapath=datapath, exp_type=bebaproto.OFPMP_EXP_STATE_STATS_SHORT,
+                                                table_id=table_id, state=state, match=match)
 
-    exp_type=bebaproto.OFPMP_EXP_STATE_STATS_AND_DELETE
-    return ofproto_parser.OFPExperimenterStatsRequest(datapath=datapath, flags=flags, experimenter=0xBEBABEBA, exp_type=exp_type, data=data)
+def OFPExpStateStatsMultipartRequestAndDeleteShort(datapath, table_id=ofproto.OFPTT_ALL, state=None, match=None):
+    return OFPExpStateStatsMultipartRequestBase(datapath=datapath, exp_type=bebaproto.OFPMP_EXP_STATE_STATS_AND_DELETE_SHORT,
+                                                table_id=table_id, state=state, match=match)
 
 """ 
 State Sync: Controller asks for flows in a specific state
@@ -306,28 +311,22 @@ def __init__(self, key_count=None, key=None, state=None):
         self.state = state
 
     @classmethod
-    def parser(cls, buf, offset):
-        entry = OFPStateEntry()
+    def parser(cls, buf, offset, key_count, state):
+        entry = OFPStateEntry(key_count=key_count, state=state)
 
-        key_count = struct.unpack_from('!I', buf, offset)
-        entry.key_count = key_count[0]
-        offset += 4
-        entry.key=[]
+        entry.key = []
         if entry.key_count <= bebaproto.MAX_KEY_LEN:
             for f in range(entry.key_count):
                 key=struct.unpack_from('!B',buf,offset,)
                 entry.key.append(key[0])
                 offset +=1
-        offset += (bebaproto.MAX_KEY_LEN - entry.key_count)
 
-        state = struct.unpack_from('!I', buf, offset)
-        entry.state=state[0]
-        offset += 4
+        offset += (utils.round_up(entry.key_count, 8) - entry.key_count)
 
-        return entry
+        return entry, offset
 
 class OFPStateStats(StringifyMixin):
-    def __init__(self, table_id=None, dur_sec=None, dur_nsec=None, field_count=None, fields=None, 
+    def __init__(self, table_id=None, dur_sec=None, dur_nsec=None, field_count=None, fields=None,
         entry=None,length=None, hard_rb=None, idle_rb=None, hard_to=None, idle_to=None):
         super(OFPStateStats, self).__init__()
         self.length = 0
@@ -343,35 +342,73 @@ def __init__(self, table_id=None, dur_sec=None, dur_nsec=None, field_count=None,
         self.hard_rb = hard_rb
 
     @classmethod
-    def parser(cls, buf, offset=0):
+    def parser(cls, msg, offset=0):
         state_stats_list = []
-        
-        for i in range(len(buf)/bebaproto.OFP_STATE_STATS_SIZE):
+
+        buf = msg.body.data
+
+        while len(buf)-offset > 0:
+            state_stats = cls()
+
+            (state_stats.length, state_stats.table_id, state_stats.key_len, state_stats.dur_sec,
+                state_stats.dur_nsec, state_stats.hard_rb, state_stats.idle_rb,state_stats.hard_to,
+                state_stats.idle_to, state_stats.state) = struct.unpack_from(
+                    bebaproto.OFP_STATE_STATS_PACK_STR, buf, offset)
+            offset += bebaproto.OFP_STATE_STATS_SIZE
+
+            if state_stats.table_id not in msg.datapath.lookup_scope:
+                continue
+            state_stats.fields = msg.datapath.lookup_scope[state_stats.table_id]
+            state_stats.field_count = len(state_stats.fields)
+
+            state_stats.entry, offset = OFPStateEntry.parser(buf, offset, state_stats.key_len, state_stats.state)
+
+            state_stats_list.append(state_stats)
+
+        return state_stats_list
+
+
+class OFPStateStatsShort(StringifyMixin):
+    def __init__(self, table_id=None, dur_sec=None, dur_nsec=None, field_count=None, fields=None,
+                 entry=None, length=None, hard_rb=None, idle_rb=None, hard_to=None, idle_to=None):
+        super(OFPStateStatsShort, self).__init__()
+        self.length = 0
+        self.table_id = table_id
+        self.dur_sec = dur_sec
+        self.dur_nsec = dur_nsec
+        self.field_count = field_count
+        self.fields = fields
+        self.entry = entry
+        self.hard_to = hard_to
+        self.hard_rb = hard_rb
+        self.idle_to = idle_to
+        self.hard_rb = hard_rb
+
+    @classmethod
+    def parser(cls, msg, offset=0):
+        state_stats_list = []
+
+        buf = msg.body.data
+
+        while len(buf) - offset > 0:
             state_stats = cls()
 
-            (state_stats.length, state_stats.table_id, state_stats.dur_sec,
-                state_stats.dur_nsec, state_stats.field_count) = struct.unpack_from(
-                bebaproto.OFP_STATE_STATS_0_PACK_STR, buf, offset)
-            offset += bebaproto.OFP_STATE_STATS_0_SIZE
-
-            state_stats.fields=[]
-            field_extract_format='!I'
-            if state_stats.field_count <= bebaproto.MAX_FIELD_COUNT:
-                for f in range(state_stats.field_count):
-                    field=struct.unpack_from(field_extract_format,buf,offset)
-                    state_stats.fields.append(field[0])
-                    offset +=4
-            offset += ((bebaproto.MAX_FIELD_COUNT-state_stats.field_count)*4)
-            state_stats.entry = OFPStateEntry.parser(buf, offset)
-            offset += bebaproto.OFP_STATE_STATS_ENTRY_SIZE
-
-            (state_stats.hard_rb, state_stats.idle_rb,state_stats.hard_to, state_stats.idle_to) = struct.unpack_from(
-                bebaproto.OFP_STATE_STATS_1_PACK_STR, buf, offset)
-            offset += bebaproto.OFP_STATE_STATS_1_SIZE
+            (state_stats.length, state_stats.table_id, state_stats.key_len, state_stats.state) = struct.unpack_from(
+                bebaproto.OFP_STATE_STATS_SHORT_PACK_STR, buf, offset)
+            offset += bebaproto.OFP_STATE_STATS_SHORT_SIZE
+
+            if state_stats.table_id not in msg.datapath.lookup_scope:
+                continue
+            state_stats.fields = msg.datapath.lookup_scope[state_stats.table_id]
+            state_stats.field_count = len(state_stats.fields)
+
+            state_stats.entry, offset = OFPStateEntry.parser(buf, offset, state_stats.key_len, state_stats.state)
+
             state_stats_list.append(state_stats)
 
         return state_stats_list
 
+
 class OFPGlobalStateStats(StringifyMixin):
     def __init__(self, global_state=None):
         super(OFPGlobalStateStats, self).__init__()