Updated Future FFI

The initial motivation for this was cancellation.  PR#1697 made it
so if an async function was cancelled we would eventually release the
resources. However, it would be better if we could immedately release
the resources.  In order to implement that, the future FFI needed to
change quite a bit and most of these changes are good.

The new FFI is simpler overall and supports cancel and drop operations.
It's actually quite close to the initial FFI that Hywan proposed. Cancel
ensures that the foreign code will resume and break out of its async
code.  Drop ensures that all resources from the wrapped future are
relased.

Note: the new FFI has a cancel method, but no bindings use it yet.  For
Python/Kotlin we don't need to because they throw cancellation
exceptions, which means cancellation support falls out from the new API
without any extra work. This cancel method could be used for Swift, but
we still need to think this through in a different PR.

The new code does not use ForeignExecutor anymore, so that code is in a
state of limbo.  I hope to repurpose it for foreign dispatch queues
(mozilla#1734).  If that doesn't work out, we can just delete it.

The FFI calls need some care since we pass a type-erased handle to the
foreign code, while RustFuture is generic over an anonymous Future type:
  - The concrete RustFuture type implements the `RustFutureFFI<F>`
    trait.
  - We give the foreign side a leaked `Box<Arc<dyn RustFutureFFI<F>>>>`.
  - We hand-monomorphize, creating a scaffolding function for each
    RustFutureFFI method, for each possible FFI type.

Updated proc macros lift arguments in 2 phases.  First we call
`try_lift` on all arguments, generating a `Result<LiftedArgsTuple>`.
Then we call the rust function using that tuple or handle the error.
This is needed because the new async code adds a `Send` bound futures.
The `Send` bound is good because futures are going to be moving across
threads as we poll/wake them.  However, the lift code won't always be Send
because it may deal with raw pointers.  To deal with that, we perform
the non-Send lifting outside of the future, then create a Send future
from the result. This means that the lift phase is executed outside of
the async context, but it should be very fast.  This change also allows
futures that return Results to attempt to downcast lift errors.

More changes:

- Updated the futures fixture tests for this to hold on to the mutex
  longer in the initial call.  This makes it so they will fail unless
  the future is dropped while the mutex is still locked.  Before they
  would only succeed as long as the mutex was dropped once the timeout
  expired.
- Updated `RustCallStatus.code` field to be an enum.  Added `Cancelled`
  as one of the variants.  `Cancelled` is only used for async functions.
- Removed the FutureCallback and invoke_future_callback from
  `FfiConverter`.
- New syncronization handling code in RustFuture that's hopefully
  clearer, more correct, and more understandable than the old stuff.
- Updated `UNIFFI_CONTRACT_VERSION` since this is an ABI change
- Removed the `RustCallStatus` param from async scaffolding functions.
  These functions can't fail, so there's no need.
- Added is_async() to the Callable trait.
- Changed `handle_failed_lift` signature. Now, if a `Result<>` is able
  to downcast the error, it returns `Self::Err` directly instead of
  serializing the error into `RustBuffer`.

Co-authored-by: Ivan Enderlin <[email protected]>
Co-authored-by: Jonas Platte <[email protected]>

docs/manual/src/futures.md

@@ -4,7 +4,7 @@ UniFFI supports exposing async Rust functions over the FFI. It can convert a Rus
 
 Check out the [examples](https://github.com/mozilla/uniffi-rs/tree/main/examples/futures) or the more terse and thorough [fixtures](https://github.com/mozilla/uniffi-rs/tree/main/fixtures/futures).
 
-Note that currently async functions are only supported by proc-macros, if you require UDL support please file a bug.
+Note that currently async functions are only supported by proc-macros, UDL support is being planned in https://github.com/mozilla/uniffi-rs/issues/1716.
 
 ## Example
 
@@ -41,97 +41,3 @@ In Rust `Future` terminology this means the foreign bindings supply the "executo
 There are [some great API docs](https://docs.rs/uniffi_core/latest/uniffi_core/ffi/rustfuture/index.html) on the implementation that are well worth a read.
 
 See the [foreign-executor fixture](https://github.com/mozilla/uniffi-rs/tree/main/fixtures/foreign-executor) for more implementation details.
-
-## How it works
-
-As [described in the documentation](https://docs.rs/uniffi_core/latest/uniffi_core/ffi/rustfuture/index.html),
-UniFFI generates code which uses callbacks from Rust futures back into that foreign "executor" to drive them to completion.
-Fortunately, each of the bindings and Rust have similar models, so the discussion below is Python, but it's almost exactly the same in Kotlin and Swift.
-
-In the above example, the generated `say_after` function looks something like:
-
-```python
-
-# A helper to work with asyncio.
-def _rust_say_after_executor(eventloop_handle, rust_task_handle):
-    event_loop = UniFFIMagic_GetExecutor(eventloop_handle)
-
-    def callback(task_handle):
-        # The event-loop has called us - call back into Rust.
-        _uniffi_say_after_executor_callback(task_handle)
-
-    # Now have the asyncio eventloop - ask it to schedule a call to help drive the Rust future.
-    eventloop.call_soon_threadsafe(callback, rust_task_handle)
-
-# A helper for say_after which creates a future and passes it Rust
-def _rust_call_say_after(callback_fn):
-    # Handle to our executor.
-    eventloop = asyncio.get_running_loop()
-    eventloop_handle = UniFFIMagic_SetExecutor(eventloop)
-
-    # Use asyncio to create a new Python future.
-    future = eventloop.create_future()
-    future_handle = UniFFIMagic_SetFuture(future)
-
-    # This is a "normal" UniFFI call across the FFI to Rust scaffoloding, but
-    # because it is an async function it has a special signature which
-    # requires the handles and the callback.
-    _uniffi_call_say_after(executor_handle, callback_fun, future_handle)
-
-    # and return the future to the caller.
-    return future
-
-def say_after_callback(future_handle, result)
-    future = UniFFIMagic_GetFuture(future_handle)
-    if future.cancelled():
-        return
-    future.set_result(result))
-
-def say_after(...):
-    return await _rust_call_say_after(say_after_callback)
-
-```
-
-And the code generated for Rust is something like:
-
-```rust
-struct SayAfterHelper {
-    rust_future: Future<>,
-    uniffi_executor_handle: ::uniffi::ForeignExecutorHandle,
-    uniffi_callback: ::uniffi::FfiConverter::FutureCallback,
-    uniffi_future_handle: ...,
-}
-
-impl SayAfterHelper {
-    fn wake(&self) {
-        match self.rust_future.poll() {
-            Some(Poll::Pending) => {
-                // ... snip executor stuff
-                self.rust_future.wake()
-            },
-            Some(Poll::Ready(v)) => {
-                // ready - tell the foreign executor
-                UniFFI_Magic_Invoke_Foreign_Callback(self.uniffi_callback, self.uniffi_future_handle)
-            },
-            None => todo!("error handling"),
-        }
-    }
-}
-
-pub extern "C" fn _uniffi_call_say_after(
-    uniffi_executor_handle: ::uniffi::ForeignExecutorHandle,
-    uniffi_callback: ::uniffi::FfiConverter::FutureCallback,
-    uniffi_future_handle: ...,
-) {
-    // Call the async function to get the Rust future.
-    let rust_future = say_after(...)
-    let helper = SayAfterHelper {
-        rust_future,
-        uniffi_executor_handle,
-        uniffi_callback,
-        uniffi_future_handle,
-    );
-    helper.wake();
-    Ok(())
-}
-```

fixtures/futures/src/lib.rs

@@ -151,6 +151,12 @@ pub async fn async_maybe_new_megaphone(y: bool) -> Option<Arc<Megaphone>> {
     }
 }
 
+/// Async function that inputs `Megaphone`.
+#[uniffi::export]
+pub async fn say_after_with_megaphone(megaphone: Arc<Megaphone>, ms: u16, who: String) -> String {
+    megaphone.say_after(ms, who).await
+}
+
 /// A megaphone. Be careful with the neighbours.
 #[derive(uniffi::Object)]
 pub struct Megaphone;

fixtures/futures/tests/bindings/test_futures.kts

@@ -92,6 +92,17 @@ runBlocking {
     assertApproximateTime(time, 200, "async methods")
 }
 
+runBlocking {
+    val megaphone = newMegaphone()
+    val time = measureTimeMillis {
+        val resultAlice = sayAfterWithMegaphone(megaphone, 200U, "Alice")
+
+        assert(resultAlice == "HELLO, ALICE!")
+    }
+
+    assertApproximateTime(time, 200, "async methods")
+}
+
 // Test async method returning optional object
 runBlocking {
     val megaphone = asyncMaybeNewMegaphone(true)
@@ -209,7 +220,7 @@ runBlocking {
 runBlocking {
     val time = measureTimeMillis {
         val job = launch {
-            useSharedResource(SharedResourceOptions(releaseAfterMs=100U, timeoutMs=1000U))
+            useSharedResource(SharedResourceOptions(releaseAfterMs=5000U, timeoutMs=100U))
         }
 
         // Wait some time to ensure the task has locked the shared resource
@@ -233,6 +244,3 @@ runBlocking {
     }
     println("useSharedResource (not canceled): ${time}ms")
 }
-
-// Test that we properly cleaned up future callback references
-assert(uniffiActiveFutureCallbacks.size == 0)

fixtures/futures/tests/bindings/test_futures.py

@@ -74,6 +74,19 @@ async def test():
 
         asyncio.run(test())
 
+    def test_async_object_param(self):
+        async def test():
+            megaphone = new_megaphone()
+            t0 = now()
+            result_alice = await say_after_with_megaphone(megaphone, 200, 'Alice')
+            t1 = now()
+
+            t_delta = (t1 - t0).total_seconds()
+            self.assertGreater(t_delta, 0.2)
+            self.assertEqual(result_alice, 'HELLO, ALICE!')
+
+        asyncio.run(test())
+
     def test_with_tokio_runtime(self):
         async def test():
             t0 = now()
@@ -150,21 +163,14 @@ async def test():
 
     # Test a future that uses a lock and that is cancelled.
     def test_shared_resource_cancellation(self):
-        # Note: Python uses the event loop to schedule calls via the `call_soon_threadsafe()`
-        # method.  This means that creating a task and cancelling it won't trigger the issue, we
-        # need to create an EventLoop and close it instead.
-        loop = asyncio.new_event_loop()
-        loop.create_task(use_shared_resource(
-            SharedResourceOptions(release_after_ms=100, timeout_ms=1000)))
-        # Wait some time to ensure the task has locked the shared resource
-        loop.call_later(0.05, loop.stop)
-        loop.run_forever()
-        # Close the EventLoop before the shared resource has been released.
-        loop.close()
-
-        # Try accessing the shared resource again using the main event loop.  The initial task
-        # should release the shared resource before the timeout expires.
-        asyncio.run(use_shared_resource(SharedResourceOptions(release_after_ms=0, timeout_ms=1000)))
+        async def test():
+            task = asyncio.create_task(use_shared_resource(
+                SharedResourceOptions(release_after_ms=5000, timeout_ms=100)))
+            # Wait some time to ensure the task has locked the shared resource
+            await asyncio.sleep(0.05)
+            task.cancel()
+            await use_shared_resource(SharedResourceOptions(release_after_ms=0, timeout_ms=1000))
+        asyncio.run(test())
 
     def test_shared_resource_no_cancellation(self):
         async def test():

fixtures/reexport-scaffolding-macro/src/lib.rs

@@ -8,7 +8,7 @@ mod tests {
     use std::ffi::CString;
     use std::os::raw::c_void;
     use std::process::{Command, Stdio};
-    use uniffi::{FfiConverter, ForeignCallback, RustBuffer, RustCallStatus};
+    use uniffi::{FfiConverter, ForeignCallback, RustBuffer, RustCallStatus, RustCallStatusCode};
     use uniffi_bindgen::ComponentInterface;
 
     struct UniFfiTag;
@@ -165,7 +165,7 @@ mod tests {
             get_symbol(&library, object_def.ffi_object_free().name());
 
         let num_alive = unsafe { get_num_alive(&mut call_status) };
-        assert_eq!(call_status.code, 0);
+        assert_eq!(call_status.code, RustCallStatusCode::Success);
         assert_eq!(num_alive, 0);
 
         let obj_id = unsafe {
@@ -174,24 +174,24 @@ mod tests {
                 &mut call_status,
             )
         };
-        assert_eq!(call_status.code, 0);
+        assert_eq!(call_status.code, RustCallStatusCode::Success);
 
         let name_buf = unsafe { coveralls_get_name(obj_id, &mut call_status) };
-        assert_eq!(call_status.code, 0);
+        assert_eq!(call_status.code, RustCallStatusCode::Success);
         assert_eq!(
             <String as FfiConverter<UniFfiTag>>::try_lift(name_buf).unwrap(),
             "TestName"
         );
 
         let num_alive = unsafe { get_num_alive(&mut call_status) };
-        assert_eq!(call_status.code, 0);
+        assert_eq!(call_status.code, RustCallStatusCode::Success);
         assert_eq!(num_alive, 1);
 
         unsafe { coveralls_free(obj_id, &mut call_status) };
-        assert_eq!(call_status.code, 0);
+        assert_eq!(call_status.code, RustCallStatusCode::Success);
 
         let num_alive = unsafe { get_num_alive(&mut call_status) };
-        assert_eq!(call_status.code, 0);
+        assert_eq!(call_status.code, RustCallStatusCode::Success);
         assert_eq!(num_alive, 0);
     }
 }

uniffi_bindgen/src/bindings/kotlin/gen_kotlin/mod.rs

@@ -198,6 +198,7 @@ pub struct KotlinWrapper<'a> {
     ci: &'a ComponentInterface,
     type_helper_code: String,
     type_imports: BTreeSet<ImportRequirement>,
+    has_async_fns: bool,
 }
 
 impl<'a> KotlinWrapper<'a> {
@@ -210,6 +211,7 @@ impl<'a> KotlinWrapper<'a> {
             ci,
             type_helper_code,
             type_imports,
+            has_async_fns: ci.has_async_fns(),
         }
     }
 
@@ -218,6 +220,10 @@ impl<'a> KotlinWrapper<'a> {
             .iter_types()
             .map(|t| KotlinCodeOracle.find(t))
             .filter_map(|ct| ct.initialization_fn())
+            .chain(
+                self.has_async_fns
+                    .then(|| "uniffiRustFutureContinuationCallback.register".into()),
+            )
             .collect()
     }
 
@@ -302,10 +308,11 @@ impl KotlinCodeOracle {
             FfiType::ForeignCallback => "ForeignCallback".to_string(),
             FfiType::ForeignExecutorHandle => "USize".to_string(),
             FfiType::ForeignExecutorCallback => "UniFfiForeignExecutorCallback".to_string(),
-            FfiType::FutureCallback { return_type } => {
-                format!("UniFfiFutureCallback{}", Self::ffi_type_label(return_type))
+            FfiType::RustFutureHandle => "Pointer".to_string(),
+            FfiType::RustFutureContinuationCallback => {
+                "UniFffiRustFutureContinuationCallbackType".to_string()
             }
-            FfiType::FutureCallbackData => "USize".to_string(),
+            FfiType::RustFutureContinuationData => "USize".to_string(),
         }
     }
 }

uniffi_bindgen/src/bindings/kotlin/templates/Async.kt

@@ -0,0 +1,44 @@
+// Async return type handlers
+
+internal const val UNIFFI_RUST_FUTURE_POLL_READY = 0.toShort()
+internal const val UNIFFI_RUST_FUTURE_POLL_MAYBE_READY = 1.toShort()
+
+internal val uniffiContinuationHandleMap = UniFfiHandleMap<CancellableContinuation<Short>>()
+
+// FFI type for Rust future continuations
+internal object uniffiRustFutureContinuationCallback: UniFffiRustFutureContinuationCallbackType {
+    override fun callback(continuationHandle: USize, pollResult: Short) {
+        uniffiContinuationHandleMap.remove(continuationHandle)?.resume(pollResult)
+    }
+
+    internal fun register(lib: _UniFFILib) {
+        lib.{{ ci.ffi_rust_future_continuation_callback_set().name() }}(this)
+    }
+}
+
+internal suspend fun<T, F, E: Exception> uniffiRustCallAsync(
+    rustFuture: Pointer,
+    pollFunc: (Pointer, USize) -> Unit,
+    completeFunc: (Pointer, RustCallStatus) -> F,
+    freeFunc: (Pointer) -> Unit,
+    liftFunc: (F) -> T,
+    errorHandler: CallStatusErrorHandler<E>
+): T {
+    try {
+        do {
+            val pollResult = suspendCancellableCoroutine<Short> { continuation ->
+                pollFunc(
+                    rustFuture,
+                    uniffiContinuationHandleMap.insert(continuation)
+                )
+            }
+        } while (pollResult != UNIFFI_RUST_FUTURE_POLL_READY);
+
+        return liftFunc(
+            rustCallWithError(errorHandler, { status -> completeFunc(rustFuture, status) })
+        )
+    } finally {
+        freeFunc(rustFuture)
+    }
+}
+