Error on mixed async/sync

Pull Request resolved: #1626


ghstack-source-id: 317483496
@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

@@ -1114,6 +1114,9 @@ def __init__(
         self._inner: "ActorMeshProtocol" = inner
         self._shape = shape
         self._proc_mesh = proc_mesh
+
+        async_endpoints = []
+        sync_endpoints = []
         for attr_name in dir(self._class):
             attr_value = getattr(self._class, attr_name, None)
             if isinstance(attr_value, EndpointProperty):
@@ -1133,6 +1136,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_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