Fix for issue 685 (#690)

* Fix the Python code to properly close the dispatcher.

Signed-off-by: Dscano <d.scano89@gmail.com>

* Fix flow rule deletion no GRPC Error

Signed-off-by: Dscano <d.scano89@gmail.com>

* Added a notification database to prevent gRPC errors caused by multiple deletions

Signed-off-by: Dscano <d.scano89@gmail.com>

* Address Andy's comments

Signed-off-by: Dscano <d.scano89@gmail.com>

---------

Signed-off-by: Dscano <d.scano89@gmail.com>

Author: Dscano

exercises/flowcache/README.md

@@ -165,6 +165,13 @@ You might notice that the rules that are printed by `mycontroller.py` contain th
 IDs rather than the table names. You can use the P4Info helper to translate these IDs
 into entry names.
 
+The current mechanism for triggering the deletion of a stale flow rule can be improved. 
+As shown in the code of `mycontroller.py`, a notification database is maintained to track 
+the flow rules. Thanks to this, the deletion of a single flow rule is performed only once. 
+However, there is a need for a mechanism to ensure that the deletion is triggered only once. 
+Without this safeguard, errors may occur when attempting to delete the same flow rule multiple 
+times or when trying to delete a flow rule that no longer exists. To address this issue, 
+you can provide and experiment with different approaches.
 
 If you are interested, you can find the protocol buffer and gRPC definitions here:
 - [P4Runtime](https://github.com/p4lang/p4runtime/blob/main/proto/p4/v1/p4runtime.proto)

exercises/flowcache/mycontroller.py

@@ -6,8 +6,11 @@
 import asyncio
 import traceback
 import time
+import ipaddress
+import pprint
 
 from collections import Counter
+from datetime import datetime, timedelta
 from scapy.all import *
 
 import grpc
@@ -34,8 +37,11 @@
 global_data["10.0.2.2"] = "08:00:00:00:02:22"
 global_data["10.0.3.3"] = "08:00:00:00:03:33"
 
+## The notification database keeps track of the received idle notifications and triggers the deletion of stale flow rules.
+notif_db = {}
+
 # The lookup table is defined to simplify reachability and provide connectivity
-# among hosts.  In a real-world scenario, however, you should use an algorithm
+# among hosts. In a real-world scenario, however, you should use an algorithm
 # to solve this problem more effectively.
 
 lookup_table = {
@@ -67,6 +73,23 @@ def ipv4ToInt(addr):
     # to add a separate check for that here.
     return int.from_bytes(bytes(bytes_), byteorder='big')
 
+def intToIpv4(n):
+    """Take an argument 'n' containing a 32-bit IPv4 address as an
+    integer in the range [0, 2^32-1], and return a string in dotted
+    decimal notation."""
+    return "%d.%d.%d.%d" % ((n >> 24) & 0xff,
+                            (n >> 16) & 0xff,
+                            (n >> 8) & 0xff,
+                            n & 0xff)
+
+def flowCacheEntryToDebugStr(table_entry, include_action=False):
+    # TODO: The match fields are hardcoded to specific indices to retrieve specific parameters, such as hdr.ipv4.srcAddr and its value.
+    src_ip = intToIpv4(int.from_bytes(table_entry.match[1].exact.value, byteorder='big'))
+    dst_ip = intToIpv4(int.from_bytes(table_entry.match[2].exact.value, byteorder='big'))
+    proto = int.from_bytes(table_entry.match[0].exact.value, byteorder='big')
+    return ("(SA=%s, DA=%s, proto=%d)"
+            "" % (src_ip, dst_ip, proto))
+
 def decodePacketInMetadata(pktin_info, packet):
     pktin_field_to_val = {}
     for md in packet.metadata:
@@ -119,6 +142,7 @@ def controllerPacketMetadataDictKeyId(p4info_obj_map, name):
         ret[md.id] = {'id': md.id, 'name': md.name, 'bitwidth': md.bitwidth}
     return ret
 
+
 def makeP4infoObjMap(p4info_data):
     p4info_obj_map = {}
     suffix_count = Counter()
@@ -178,11 +202,59 @@ def addFlowRule( ingress_sw, src_ip_addr, dst_ip_addr, protocol, port, new_dscp,
         # TODO: Add idle timeout
         )
     ingress_sw.WriteTableEntry(table_entry)
-    print("Installed ingress rule on %s" % ingress_sw.name)
 
-def deleteFlowRule(notif):
-    notif["sw"].DeleteTableEntry(notif["idle"].table_entry[0])
-    print("Deleted ingress rule on %s" % notif["sw"].name)
+def createFlowRule(notif):
+    # TODO: This function generates a flow entry to trigger deletion. 
+    # The match fields are populated using values retrieved from the IDLE notification.
+    # Hardcoded values are configured to extract specific parameters, such as hdr.ipv4.protocol, from the IDLE notification.
+    table_entry = global_data['p4info_helper'].buildTableEntry(
+        table_name="MyIngress.flow_cache",
+        match_fields={
+            "hdr.ipv4.protocol": int.from_bytes(notif["idle"].table_entry[0].match[0].exact.value,byteorder='big'),
+            "hdr.ipv4.srcAddr": int(ipaddress.IPv4Address(notif["idle"].table_entry[0].match[1].exact.value)),
+            "hdr.ipv4.dstAddr": int(ipaddress.IPv4Address(notif["idle"].table_entry[0].match[2].exact.value))
+        },
+    )
+    return table_entry
+
+def deleteFlowRule(sw, table_entry):
+    sw.DeleteTableEntry(table_entry)
+    print("Deleted flow_cache entry on %s. %s"
+          "" % (sw.name, flowCacheEntryToDebugStr(table_entry)))
+
+def addNotification(sw_name, flow_rule):
+    # Add notification to notification DB
+    notification = {
+        "timestamp": datetime.now(),
+        "flow_rule": flow_rule,
+    }
+    notif_db[sw_name].append(notification)
+
+def checkFlowRule(sw_name, flow_rule):
+    # Checks if a flow rule is already in the notification DB
+    # to avoid storing multiple notifications for the same flow rule
+    if sw_name not in notif_db:
+        return False
+
+    for notif in notif_db[sw_name]:
+        if notif["flow_rule"] == flow_rule:
+            return True
+        
+    return False
+
+def isExpired(timestamp, timeout):
+    return datetime.now() - timestamp > timedelta(seconds=timeout)
+
+def cleanExpiredNotifiction(sw_name, timeout=5):
+    # Removes expired notifications 
+    if sw_name not in notif_db:
+        return False
+    # Filter the notifications to remove expired ones
+    notif_db[sw_name] = [
+        notif for notif in notif_db[sw_name]
+        if not isExpired(notif["timestamp"], timeout)
+    ]
+    return True
 
 def packetOutMetadataList(opcode, reserved1, operand0):
     # This function does not use the generated contents of the P4Info
@@ -195,7 +267,6 @@ def packetOutMetadataList(opcode, reserved1, operand0):
 
 def sendPacketOut(sw ,payload, metadatas):
     # TODO: Implement the function logic to send a packet-out message
-    sw.PacketOut(payload, metadatas)
 
 def readTableRules(p4info_helper, sw):
     """
@@ -295,30 +366,54 @@ def processPacket(message):
                             decrement_ttl_bool,
                             dst_eth_addr)
 
-                print("For flow (SA=%s, DA=%s, proto=%d)"
+                print("For switch %s flow (SA=%s, DA=%s, proto=%d)"
                             " added table entry to send packets"
                             " to port %d with new DSCP %d"
-                            "" % (ip_sa_str, ip_da_str, ip_proto,
-                                  dest_port_int, new_dscp_int))
+                            "" % (message["sw"].name, ip_sa_str, ip_da_str,
+                                  ip_proto, dest_port_int, new_dscp_int))
 
 async def processNotif(notif_queue):
         while True:
             notif = await notif_queue.get()
-            print(notif)
+            debug_notif = False
+            if debug_notif:
+                print(notif)
+                pprint.pprint(notif_db)
             if notif["type"] == "packet-in":
                 processPacket(notif)
                 printCounter(global_data ['p4info_helper'], notif["sw"], 'MyIngress.ingressPktOutCounter', global_data ['index'])
                 printCounter(global_data ['p4info_helper'], notif["sw"], 'MyEgress.egressPktInCounter', global_data ['index'])
-                readTableRules(global_data ['p4info_helper'], notif["sw"])
+                if debug_notif:
+                    readTableRules(global_data ['p4info_helper'], notif["sw"])
             elif notif["type"] == "idle-notif":
-                deleteFlowRule(notif)
-
+                # TODO: For extra credit, you can experiment with adjusting the stale time for notifications (e.g., 10 seconds)
+                # and optimize the behavior of the notification database (notif_db).
+                if notif["sw"].name not in notif_db:
+                    notif_db[notif["sw"].name] = []
+                else:
+                    # Check if a notification is older than 10 seconds
+                    cleanExpiredNotifiction(notif["sw"].name, 10)
+
+                table_entry = createFlowRule(notif)
+
+                if not checkFlowRule(notif["sw"].name, table_entry):
+                    addNotification(notif["sw"].name, table_entry)
+                    deleteFlowRule(notif["sw"], table_entry)
+                else:
+                    print("Received idle timeout notification for switch=%s %s"
+                          "  It is duplicate of recently processed notification, so ignoring it."
+                          "" % (notif["sw"].name,
+                                flowCacheEntryToDebugStr(table_entry)))
             notif_queue.task_done()
 
 async def packetInHandler(notif_queue,sw):
     # TODO: Implement the function logic to handle a packet-in message
     while True:
         try:
+            packet_in = await asyncio.to_thread(sw.PacketIn)
+            #print(f"Received packet: {packet_in}")
+            message = {"type": "packet-in", "sw": sw, "packet-in": packet_in}
+            await notif_queue.put(message)
 
         except grpc.RpcError as e:
             print(f"[gRPC Error in packetInHandler for {sw.name}]")
@@ -336,7 +431,10 @@ async def packetInHandler(notif_queue,sw):
 async def idleTimeHandler(notif_queue,sw):
     # TODO: Implement the function logic to handle idle timeout notification
     while True:
-
+        idle_notif = await asyncio.to_thread(sw.IdleTimeoutNotification)
+        message = {"type": "idle-notif", "sw": sw, "idle": idle_notif}
+        await notif_queue.put(message)
+        await asyncio.sleep(5)
 
 def printGrpcError(e):
     print("gRPC Error:", e.details(), end=' ')
@@ -426,7 +524,9 @@ async def main(p4info_file_path, bmv2_file_path):
     except KeyboardInterrupt:
         print(" Shutting down.")
     except grpc.RpcError as e:
-        printGrpcError(e)
+        print(f"gRPC error occurred: {e}")
+        print(f"Status code: {e.code()}")  # e.g., StatusCode.UNAVAILABLE or StatusCode.INVALID_ARGUMENT
+        print(f"Details: {e.details()}")
 
     ShutdownAllSwitchConnections()