refactor(framework) Replace asyncio.Event with threading.Event

src/py/flwr/server/superlink/fleet/vce/vce_api.py

@@ -15,7 +15,6 @@
 """Fleet Simulation Engine API."""
 
 
-import asyncio
 import json
 import sys
 import threading
@@ -63,7 +62,7 @@ def worker(
     taskres_queue: "Queue[TaskRes]",
     node_states: Dict[int, NodeState],
     backend: Backend,
-    f_stop: asyncio.Event,
+    f_stop: threading.Event,
 ) -> None:
     """Get TaskIns from queue and pass it to an actor in the pool to execute it."""
     while not f_stop.is_set():
@@ -123,7 +122,7 @@ def add_taskins_to_queue(
     state: State,
     queue: "Queue[TaskIns]",
     nodes_mapping: NodeToPartitionMapping,
-    f_stop: asyncio.Event,
+    f_stop: threading.Event,
 ) -> None:
     """Put TaskIns in a queue from State."""
     while not f_stop.is_set():
@@ -153,7 +152,7 @@ def run(
     nodes_mapping: NodeToPartitionMapping,
     state_factory: StateFactory,
     node_states: Dict[int, NodeState],
-    f_stop: asyncio.Event,
+    f_stop: threading.Event,
 ) -> None:
     """Run the VCE."""
     taskins_queue: "Queue[TaskIns]" = Queue()
@@ -232,7 +231,7 @@ def start_vce(
     backend_name: str,
     backend_config_json_stream: str,
     app_dir: str,
-    f_stop: asyncio.Event,
+    f_stop: threading.Event,
     client_app: Optional[ClientApp] = None,
     client_app_attr: Optional[str] = None,
     num_supernodes: Optional[int] = None,

src/py/flwr/server/superlink/fleet/vce/vce_api_test.py

@@ -15,7 +15,6 @@
 """Test Fleet Simulation Engine API."""
 
 
-import asyncio
 import threading
 import time
 from itertools import cycle
@@ -46,7 +45,7 @@
 from flwr.server.superlink.state import InMemoryState, StateFactory
 
 
-def terminate_simulation(f_stop: asyncio.Event, sleep_duration: int) -> None:
+def terminate_simulation(f_stop: threading.Event, sleep_duration: int) -> None:
     """Set event to terminate Simulation Engine after `sleep_duration` seconds."""
     sleep(sleep_duration)
     f_stop.set()
@@ -148,15 +147,15 @@ def start_and_shutdown(
 ) -> None:
     """Start Simulation Engine and terminate after specified number of seconds.
 
-    Some tests need to be terminated by triggering externally an asyncio.Event. This
+    Some tests need to be terminated by triggering externally an threading.Event. This
     is enabled when passing `duration`>0.
     """
-    f_stop = asyncio.Event()
+    f_stop = threading.Event()
 
     if duration:
 
         # Setup thread that will set the f_stop event, triggering the termination of all
-        # asyncio logic in the Simulation Engine. It will also terminate the Backend.
+        # logic in the Simulation Engine. It will also terminate the Backend.
         termination_th = threading.Thread(
             target=terminate_simulation, args=(f_stop, duration)
         )

src/py/flwr/simulation/run_simulation.py

@@ -129,7 +129,7 @@ def run_serverapp_th(
     server_app_run_config: Dict[str, str],
     driver: Driver,
     app_dir: str,
-    f_stop: asyncio.Event,
+    f_stop: threading.Event,
     has_exception: threading.Event,
     enable_tf_gpu_growth: bool,
     delay_launch: int = 3,
@@ -138,7 +138,7 @@ def run_serverapp_th(
 
     def server_th_with_start_checks(
         tf_gpu_growth: bool,
-        stop_event: asyncio.Event,
+        stop_event: threading.Event,
         exception_event: threading.Event,
         _driver: Driver,
         _server_app_dir: str,
@@ -225,7 +225,7 @@ def _main_loop(
     # Initialize StateFactory
     state_factory = StateFactory(":flwr-in-memory-state:")
 
-    f_stop = asyncio.Event()
+    f_stop = threading.Event()
     # A Threading event to indicate if an exception was raised in the ServerApp thread
     server_app_thread_has_exception = threading.Event()
     serverapp_th = None