refactor(framework) Remove asyncio from Backend definitions

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

@@ -33,8 +33,8 @@ def __init__(self, backend_config: BackendConfig, work_dir: str) -> None:
         """Construct a backend."""
 
     @abstractmethod
-    async def build(self) -> None:
-        """Build backend asynchronously.
+    def build(self) -> None:
+        """Build backend.
 
         Different components need to be in place before workers in a backend are ready
         to accept jobs. When this method finishes executing, the backend should be fully
@@ -54,11 +54,11 @@ def is_worker_idle(self) -> bool:
         """Report whether a backend worker is idle and can therefore run a ClientApp."""
 
     @abstractmethod
-    async def terminate(self) -> None:
+    def terminate(self) -> None:
         """Terminate backend."""
 
     @abstractmethod
-    async def process_message(
+    def process_message(
         self,
         app: Callable[[], ClientApp],
         message: Message,

src/py/flwr/server/superlink/fleet/vce/backend/raybackend.py

@@ -153,12 +153,12 @@ def is_worker_idle(self) -> bool:
         """Report whether the pool has idle actors."""
         return self.pool.is_actor_available()
 
-    async def build(self) -> None:
+    def build(self) -> None:
         """Build pool of Ray actors that this backend will submit jobs to."""
-        await self.pool.add_actors_to_pool(self.pool.actors_capacity)
+        self.pool.add_actors_to_pool(self.pool.actors_capacity)
         log(DEBUG, "Constructed ActorPool with: %i actors", self.pool.num_actors)
 
-    async def process_message(
+    def process_message(
         self,
         app: Callable[[], ClientApp],
         message: Message,
@@ -172,17 +172,16 @@ async def process_message(
 
         try:
             # Submit a task to the pool
-            future = await self.pool.submit(
+            future = self.pool.submit(
                 lambda a, a_fn, mssg, cid, state: a.run.remote(a_fn, mssg, cid, state),
                 (app, message, str(partition_id), context),
             )
 
-            await future
             # Fetch result
             (
                 out_mssg,
                 updated_context,
-            ) = await self.pool.fetch_result_and_return_actor_to_pool(future)
+            ) = self.pool.fetch_result_and_return_actor_to_pool(future)
 
             return out_mssg, updated_context
 
@@ -193,11 +192,11 @@ async def process_message(
                 self.__class__.__name__,
             )
             # add actor back into pool
-            await self.pool.add_actor_back_to_pool(future)
+            self.pool.add_actor_back_to_pool(future)
             raise ex
 
-    async def terminate(self) -> None:
+    def terminate(self) -> None:
         """Terminate all actors in actor pool."""
-        await self.pool.terminate_all_actors()
+        self.pool.terminate_all_actors()
         ray.shutdown()
         log(DEBUG, "Terminated %s", self.__class__.__name__)

src/py/flwr/server/superlink/fleet/vce/backend/raybackend_test.py

@@ -14,11 +14,10 @@
 # ==============================================================================
 """Test for Ray backend for the Fleet API using the Simulation Engine."""
 
-import asyncio
 from math import pi
 from pathlib import Path
 from typing import Callable, Dict, Optional, Tuple, Union
-from unittest import IsolatedAsyncioTestCase
+from unittest import TestCase
 
 import ray
 
@@ -84,18 +83,18 @@ def _load_app() -> ClientApp:
     return _load_app
 
 
-async def backend_build_process_and_termination(
+def backend_build_process_and_termination(
     backend: RayBackend,
     process_args: Optional[Tuple[Callable[[], ClientApp], Message, Context]] = None,
 ) -> Union[Tuple[Message, Context], None]:
     """Build, process job and terminate RayBackend."""
-    await backend.build()
+    backend.build()
     to_return = None
 
     if process_args:
-        to_return = await backend.process_message(*process_args)
+        to_return = backend.process_message(*process_args)
 
-    await backend.terminate()
+    backend.terminate()
 
     return to_return
 
@@ -129,20 +128,18 @@ def _create_message_and_context() -> Tuple[Message, Context, float]:
     return message, context, expected_output
 
 
-class AsyncTestRayBackend(IsolatedAsyncioTestCase):
-    """A basic class that allows runnig multliple asyncio tests."""
+class TestRayBackend(TestCase):
+    """A basic class that allows runnig multliple tests."""
 
-    async def on_cleanup(self) -> None:
+    def doCleanups(self) -> None:
         """Ensure Ray has shutdown."""
         if ray.is_initialized():
             ray.shutdown()
 
     def test_backend_creation_and_termination(self) -> None:
         """Test creation of RayBackend and its termination."""
         backend = RayBackend(backend_config={}, work_dir="")
-        asyncio.run(
-            backend_build_process_and_termination(backend=backend, process_args=None)
-        )
+        backend_build_process_and_termination(backend=backend, process_args=None)
 
     def test_backend_creation_submit_and_termination(
         self,
@@ -157,10 +154,8 @@ def test_backend_creation_submit_and_termination(
 
         message, context, expected_output = _create_message_and_context()
 
-        res = asyncio.run(
-            backend_build_process_and_termination(
-                backend=backend, process_args=(client_app_callable, message, context)
-            )
+        res = backend_build_process_and_termination(
+            backend=backend, process_args=(client_app_callable, message, context)
         )
 
         if res is None:
@@ -189,7 +184,6 @@ def test_backend_creation_submit_and_termination_non_existing_client_app(
             self.test_backend_creation_submit_and_termination(
                 client_app_loader=_load_from_module("a_non_existing_module:app")
             )
-        self.addAsyncCleanup(self.on_cleanup)
 
     def test_backend_creation_submit_and_termination_existing_client_app(
         self,
@@ -217,7 +211,6 @@ def test_backend_creation_submit_and_termination_existing_client_app_unsetworkdi
                 client_app_loader=_load_from_module("raybackend_test:client_app"),
                 workdir="/?&%$^#%@$!",
             )
-        self.addAsyncCleanup(self.on_cleanup)
 
     def test_backend_creation_with_init_arguments(self) -> None:
         """Testing whether init args are properly parsed to Ray."""
@@ -248,5 +241,3 @@ def test_backend_creation_with_init_arguments(self) -> None:
         nodes = ray.nodes()
 
         assert nodes[0]["Resources"]["CPU"] == backend_config_2["init_args"]["num_cpus"]
-
-        self.addAsyncCleanup(self.on_cleanup)