Error on mixed async/sync

Pull Request resolved: #1626


ghstack-source-id: 317845131
@exported-using-ghexport

Differential Revision: [D85090856](https://our.internmc.facebook.com/intern/diff/D85090856/)

**NOTE FOR REVIEWERS**: This PR has internal Meta-specific changes or comments, please review them on [Phabricator](https://our.internmc.facebook.com/intern/diff/D85090856/)!

Author: zdevito

python/monarch/_src/actor/actor_mesh.py

@@ -1122,6 +1122,8 @@ def __init__(
         # can be collected even without any endpoints being awaited.
         self._inner.start_supervision(context().actor_instance._as_rust())
 
+        async_endpoints = []
+        sync_endpoints = []
         for attr_name in dir(self._class):
             attr_value = getattr(self._class, attr_name, None)
             if isinstance(attr_value, EndpointProperty):
@@ -1141,6 +1143,17 @@ def __init__(
                         attr_value._explicit_response_port,
                     ),
                 )
+                if inspect.iscoroutinefunction(attr_value._method):
+                    async_endpoints.append(attr_name)
+                else:
+                    sync_endpoints.append(attr_name)
+
+        if sync_endpoints and async_endpoints:
+            raise ValueError(
+                f"{self._class} mixes both async and sync endpoints."
+                "Synchronous endpoints cannot be mixed with async endpoints because they can cause the asyncio loop to deadlock if they wait."
+                f"sync: {sync_endpoints} async: {async_endpoints}"
+            )
 
     def __getattr__(self, attr: str) -> NotAnEndpoint:
         if attr in dir(self._class):

python/tests/test_actor_error.py

@@ -423,13 +423,23 @@ class ActorFailureError(BaseException):
     pass
 
 
-class ErrorActor(Actor):
+class SyncErrorActor(Actor):
     @endpoint
     def fail_with_supervision_error(self) -> None:
         raise ActorFailureError("Simulated actor failure for supervision testing")
 
     @endpoint
-    async def fail_with_supervision_error_async(self) -> None:
+    def check(self) -> str:
+        return "this is a healthy check"
+
+    @endpoint
+    def check_with_exception(self) -> None:
+        raise RuntimeError("failed the check with app error")
+
+
+class ErrorActor(Actor):
+    @endpoint
+    async def fail_with_supervision_error(self) -> None:
         raise ActorFailureError("Simulated actor failure for supervision testing")
 
     @endpoint
@@ -579,10 +589,10 @@ async def test_actor_mesh_supervision_handling_chained_error() -> None:
     ids=["local_proc_mesh", "proc_mesh"],
 )
 @pytest.mark.parametrize(
-    "method_name",
-    ["fail_with_supervision_error", "fail_with_supervision_error_async"],
+    "error_actor_cls",
+    [ErrorActor, SyncErrorActor],
 )
-async def test_base_exception_handling(mesh, method_name) -> None:
+async def test_base_exception_handling(mesh, error_actor_cls) -> None:
     """Test that BaseException subclasses trigger supervision errors.
 
     This test verifies that both synchronous and asynchronous methods
@@ -591,10 +601,10 @@ async def test_base_exception_handling(mesh, method_name) -> None:
 
     """
     proc = mesh({"gpus": 1})
-    error_actor = proc.spawn("error", ErrorActor)
+    error_actor = proc.spawn("error", error_actor_cls)
 
     # Get the method to call based on the parameter
-    method = getattr(error_actor, method_name)
+    method = error_actor.fail_with_supervision_error
 
     # The call should raise a SupervisionError
     with pytest.raises(
@@ -807,16 +817,6 @@ async def subworker_fail(self, sleep: float | None = None) -> None:
             "Should never get here, SupervisionError should be raised by the above call"
         )
 
-    @endpoint
-    async def subworker_broadcast_fail(self) -> None:
-        self.mesh.check.broadcast()
-        # When not awaiting the result of an endpoint which experiences a
-        # failure, it should still propagate back to __supervise__.
-        self.mesh.fail_with_supervision_error.broadcast()
-        # Give time for the failure to occur before returning, so get_failures
-        # will have a non-empty result.
-        await asyncio.sleep(10)
-
     @endpoint
     async def get_failures(self) -> list[str]:
         # MeshFailure is not picklable, so we convert it to a string.

python/tests/test_debugger.py

@@ -1151,7 +1151,7 @@ def debug_func(self, func, class_closure) -> int:
         return func(class_closure)
 
     @endpoint
-    async def name(self) -> str:
+    def name(self) -> str:
         return context().actor_instance.actor_id.actor_name
 
 

python/tests/test_python_actors.py

@@ -87,9 +87,14 @@ async def incr(self):
     async def value(self) -> int:
         return self.v
 
+
+class SyncCounter(Actor):
+    def __init__(self, c):
+        self.c = c
+
     @endpoint
     def value_sync_endpoint(self) -> int:
-        return self.v
+        return self.c.value.choose().get()
 
 
 class Indirect(Actor):
@@ -112,7 +117,8 @@ async def test_choose():
 
     assert result == result2
 
-    result3 = await v.value_sync_endpoint.choose()
+    v2 = proc.spawn("sync_counter", SyncCounter, v)
+    result3 = v2.value_sync_endpoint.choose().get()
     assert_type(result, int)
     assert result2 == result3
 
@@ -335,15 +341,15 @@ def __init__(self):
         self.local.value = 0
 
     @endpoint
-    def increment(self):
+    async def increment(self):
         self.local.value += 1
 
     @endpoint
     async def increment_async(self):
         self.local.value += 1
 
     @endpoint
-    def get_value(self):
+    async def get_value(self):
         return self.local.value
 
     @endpoint
@@ -408,18 +414,6 @@ async def no_more(self) -> None:
         self.should_exit = True
 
 
-@pytest.mark.timeout(60)
-async def test_async_concurrency():
-    """Test that async endpoints will be processed concurrently."""
-    pm = this_host().spawn_procs(per_host={})
-    am = pm.spawn("async", AsyncActor)
-    fut = am.sleep.call()
-    # This call should go through and exit the sleep loop, as long as we are
-    # actually concurrently processing messages.
-    await am.no_more.call()
-    await fut
-
-
 async def awaitit(f):
     return await f
 
@@ -1543,7 +1537,7 @@ def __init__(self, root="None"):
         self._root = root
 
     @endpoint
-    def return_root(self):
+    async def return_root(self):
         return self._root
 
     @endpoint