Clean future cancellation handling (#1687)

Summary:
Pull Request resolved: #1687

The implementation of `monarch.actor.Future` uses `loop.call_soon_threadsafe` to set a value/exception on an underlying asyncio future. But if the caller cancelled the asyncio future before `call_soon_threadsafe` executes the callback, then `asyncio.exceptions.InvalidStateError` will be logged, which is noisy and misleading.

This diff solves the issue by checking whether the future is already resolved/cancelled from inside the callback.
ghstack-source-id: 319407775
exported-using-ghexport

Reviewed By: zdevito

Differential Revision: D85693875

fbshipit-source-id: eb515cd65e78fcbb853909af3ba96196c1315c98

Author: samlurye

python/monarch/_src/actor/future.py

@@ -138,12 +138,20 @@ def __await__(self) -> Generator[Any, Any, R]:
                     fut = loop.create_future()
                     self._status = _Asyncio(fut)
 
+                    def set_result(fut, value):
+                        if not fut.cancelled():
+                            fut.set_result(value)
+
+                    def set_exception(fut, e):
+                        if not fut.cancelled():
+                            fut.set_exception(e)
+
                     async def mark_complete():
                         try:
-                            func, value = fut.set_result, await coro
+                            func, value = set_result, await coro
                         except Exception as e:
-                            func, value = fut.set_exception, e
-                        loop.call_soon_threadsafe(func, value)
+                            func, value = set_exception, e
+                        loop.call_soon_threadsafe(func, fut, value)
 
                     PythonTask.from_coroutine(mark_complete()).spawn()
                     return fut.__await__()