Revert recent switch.py changes (#688)

* Move optional steps for testing VMs out of README.md file
This may help avoid new users being confused that they must follow
those instructions.

Signed-off-by: Andy Fingerhut <andy_fingerhut@alum.wustl.edu>

* Add 2025-Jan-30 versions of source code for install.sh

Signed-off-by: Andy Fingerhut <andy_fingerhut@alum.wustl.edu>

* Revert recent changes made to utils/p4runtime_lib/switch.py
They cause Python exceptions to be raised when initializing other
exercises, e.g. basic and ecn at least.

Signed-off-by: Andy Fingerhut <andy_fingerhut@alum.wustl.edu>

---------

Signed-off-by: Andy Fingerhut <andy_fingerhut@alum.wustl.edu>

Author: jafingerhut

utils/p4runtime_lib/switch.py

@@ -16,7 +16,6 @@
 from abc import abstractmethod
 from datetime import datetime
 from queue import Queue
-import threading
 
 import grpc
 from p4.tmp import p4config_pb2
@@ -31,36 +30,6 @@ def ShutdownAllSwitchConnections():
     for c in connections:
         c.shutdown()
 
-class StreamDispatcher:
-    def __init__(self, stream):
-        self.stream = stream
-        self.running = True
-        # Queues for each message type
-        self.arbitration_queue = Queue()
-        self.packet_in_queue = Queue()
-        self.timeout_queue = Queue()
-        self.error_queue = Queue()
-
-        self.thread = threading.Thread(target=self._dispatch_loop, daemon=True)
-        self.thread.start()
-
-    def _dispatch_loop(self):
-        for msg in self.stream:
-            if not self.running:
-                break
-            if msg.HasField("arbitration"):
-                self.arbitration_queue.put(msg.arbitration)
-            elif msg.HasField("packet"):
-                self.packet_in_queue.put(msg.packet)
-            elif msg.HasField("idle_timeout_notification"):
-                self.timeout_queue.put(msg.idle_timeout_notification)
-            elif msg.HasField("error"):
-                self.error_queue.put(msg.error)
-            else:
-                print("Unknown StreamMessageResponse:", msg)
-
-    def stop(self):
-        self.running = False
 class SwitchConnection(object):
 
     def __init__(self, name=None, address='127.0.0.1:50051', device_id=0,
@@ -76,7 +45,6 @@ def __init__(self, name=None, address='127.0.0.1:50051', device_id=0,
         self.client_stub = p4runtime_pb2_grpc.P4RuntimeStub(self.channel)
         self.requests_stream = IterableQueue()
         self.stream_msg_resp = self.client_stub.StreamChannel(iter(self.requests_stream))
-        self.dispatcher = StreamDispatcher(self.stream_msg_resp)
         self.proto_dump_file = proto_dump_file
         connections.append(self)
 
@@ -98,7 +66,8 @@ def MasterArbitrationUpdate(self, dry_run=False, **kwargs):
             print("P4Runtime MasterArbitrationUpdate: ", request)
         else:
             self.requests_stream.put(request)
-            return self.dispatcher.arbitration_queue.get()
+            for item in self.stream_msg_resp:
+                return item # just one
 
     def SetForwardingPipelineConfig(self, p4info, dry_run=False, **kwargs):
         device_config = self.buildDeviceConfig(**kwargs)
@@ -131,18 +100,6 @@ def WriteTableEntry(self, table_entry, dry_run=False):
         else:
             self.client_stub.Write(request)
 
-    def DeleteTableEntry(self, table_entry, dry_run=False):
-        request = p4runtime_pb2.WriteRequest()
-        request.device_id = self.device_id
-        request.election_id.low = 1
-        update = request.updates.add()
-        update.type = p4runtime_pb2.Update.DELETE
-        update.entity.table_entry.CopyFrom(table_entry)
-        if dry_run:
-            print("P4Runtime Write:", request)
-        else:
-            self.client_stub.Write(request)
-
     def ReadTableEntries(self, table_id=None, dry_run=False):
         request = p4runtime_pb2.ReadRequest()
         request.device_id = self.device_id
@@ -188,11 +145,14 @@ def WritePREEntry(self, pre_entry, dry_run=False):
             self.client_stub.Write(request)
 
     def PacketIn(self, dry_run=False):
-        request = self.dispatcher.packet_in_queue.get()
+        request = p4runtime_pb2.StreamMessageRequest()
         if dry_run:
             print("P4 Runtime PacketIn: ", request)
         else:
-            return request
+            self.requests_stream.put(request)
+            for item in self.stream_msg_resp:
+                print(item)
+                return item
 
     def PacketOut(self, payload, metadatas):
         packet_out = p4runtime_pb2.PacketOut()
@@ -212,12 +172,10 @@ def PacketOut(self, payload, metadatas):
         request.packet.CopyFrom(packet_out)
         self.requests_stream.put(request)
 
-    def IdleTimeoutNotification(self, dry_run=False):
-        msg = self.dispatcher.timeout_queue.get()
-        if dry_run:
-            print("P4 Runtime PacketIn: ", msg)
-        else:
-            return msg
+    def IdleTimeoutNotification(self):
+        msg = p4runtime_pb2.IdleTimeoutNotification()
+        return msg
+
 class GrpcRequestLogger(grpc.UnaryUnaryClientInterceptor,
                         grpc.UnaryStreamClientInterceptor):
     """Implementation of a gRPC interceptor that logs request to a file"""