From 1dbc83f6da5ea42ef1af9b56420ed57a8f8b1ae1 Mon Sep 17 00:00:00 2001 From: Murph Murphy Date: Wed, 16 Sep 2026 15:38:55 -0600 Subject: [PATCH 1/2] Make cancellation more robust, document spots we can't improve past Also expand test coverage with more edge cases. --- CHANGELOG.md | 2 +- Cargo.lock | 57 +++++ Cargo.toml | 15 +- README.md | 1 + fixtures/async-lifecycle/Cargo.toml | 13 ++ fixtures/async-lifecycle/src/lib.rs | 142 ++++++++++++ src/gen_java/mod.rs | 55 ++++- src/templates/Async.java | 102 ++++----- src/templates/macros.java | 10 +- .../TestAsyncLifecycle.java | 90 ++++++++ .../uniffi/async_lifecycle/AsyncContract.java | 209 ++++++++++++++++++ tests/tests.rs | 28 ++- 12 files changed, 645 insertions(+), 79 deletions(-) create mode 100644 fixtures/async-lifecycle/Cargo.toml create mode 100644 fixtures/async-lifecycle/src/lib.rs create mode 100644 tests/scripts/TestAsyncLifecycle/TestAsyncLifecycle.java create mode 100644 tests/scripts/TestAsyncLifecycle/uniffi/async_lifecycle/AsyncContract.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ca7ba9..764823a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,6 @@ ## 0.5.2 -- fix a narrow race in async cancellation. If `cancel()` on a `CompletableFuture` returned by an async function landed in the few instructions between the pipeline's `isCancelled()` check and its own `rust_future_free`, or between a wake and the re-poll it triggers, the Rust future was freed twice or polled after being freed. Seen once in CI as a glibc `tcache_thread_shutdown()` abort. Every use of the handle now runs under the future's monitor and stops once it has been freed. No measurable change to async call overhead. +- fix a narrow race in async cancellation. If `cancel()` on a `CompletableFuture` returned by an async function landed in the few instructions between the pipeline's `isCancelled()` check and its own `rust_future_free`, or between a wake and the re-poll it triggers, the Rust future was freed twice or polled after being freed. Seen once in CI as a glibc `tcache_thread_shutdown()` abort. `cancel()` now signals Rust with `rust_future_cancel` and the poll/complete pipeline is the only thing that frees. The Rust future is dropped once the in-flight poll's continuation fires, on an executor thread rather than the cancelling one. ## 0.5.1 diff --git a/Cargo.lock b/Cargo.lock index 889149e..aae96a0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1017,6 +1017,15 @@ version = "0.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae" +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + [[package]] name = "log" version = "0.4.34" @@ -1081,6 +1090,29 @@ version = "2.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + [[package]] name = "paste" version = "1.0.15" @@ -1219,6 +1251,15 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + [[package]] name = "regex" version = "1.13.1" @@ -1282,6 +1323,12 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + [[package]] name = "scroll" version = "0.12.0" @@ -1605,6 +1652,7 @@ dependencies = [ "uniffi-example-rondpoint", "uniffi-example-sprites", "uniffi-example-todolist", + "uniffi-fixture-async-lifecycle", "uniffi-fixture-benchmarks", "uniffi-fixture-coverall", "uniffi-fixture-enum-types", @@ -1685,6 +1733,15 @@ dependencies = [ "uniffi", ] +[[package]] +name = "uniffi-fixture-async-lifecycle" +version = "0.1.0" +dependencies = [ + "async-trait", + "parking_lot", + "uniffi", +] + [[package]] name = "uniffi-fixture-benchmarks" version = "0.22.0" diff --git a/Cargo.toml b/Cargo.toml index 48a4ca8..c4ae6cb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,17 +24,17 @@ bench = false [dependencies] anyhow = "1" askama = { version = "0.16", default-features = false, features = [ - "config", - "derive", - "alloc", + "config", + "derive", + "alloc", ] } camino = "1.1.6" cargo_metadata = "0.23" clap = { version = "4", default-features = false, features = [ - "derive", - "help", - "std", - "cargo", + "derive", + "help", + "std", + "cargo", ] } heck = "0.5" once_cell = "1.19.0" @@ -56,6 +56,7 @@ uniffi-example-geometry = { git = "https://github.com/mozilla/uniffi-rs.git", br uniffi-example-rondpoint = { git = "https://github.com/mozilla/uniffi-rs.git", branch = "release-v0.32.x" } uniffi-example-sprites = { git = "https://github.com/mozilla/uniffi-rs.git", branch = "release-v0.32.x" } uniffi-example-todolist = { git = "https://github.com/mozilla/uniffi-rs.git", branch = "release-v0.32.x" } +uniffi-fixture-async-lifecycle = { path = "fixtures/async-lifecycle" } uniffi-fixture-benchmarks = { git = "https://github.com/mozilla/uniffi-rs.git", branch = "release-v0.32.x" } uniffi-fixture-coverall = { git = "https://github.com/mozilla/uniffi-rs.git", branch = "release-v0.32.x" } uniffi-fixture-enum-types = { git = "https://github.com/mozilla/uniffi-rs.git", branch = "release-v0.32.x" } diff --git a/README.md b/README.md index e4a9de8..c6e4dc1 100644 --- a/README.md +++ b/README.md @@ -236,6 +236,7 @@ scope). - a Rust `&[u8]` argument is borrowed by Rust for the duration of the call rather than copied, and takes a **direct** `java.nio.ByteBuffer` (as in Kotlin) instead of `byte[]`. Build one with `ByteBuffer.allocateDirect(arr.length).put(arr).flip()`; a heap buffer throws `IllegalArgumentException`. Reuse the buffer across calls where you can, and don't let another thread write to it while a call is in flight. `Vec` is unaffected and still maps to `byte[]`. - failures in CompletableFutures will cause them to `completeExceptionally`. The error that caused the failure can be checked with `e.getCause()`. When implementing an async Rust trait in Java, you'll need to `completeExceptionally` instead of throwing. See `TestFixtureFutures.java` for an example trait implementation with errors. +- every async function and method has an overload taking a `java.util.concurrent.Executor`; the other overload uses `ForkJoinPool.commonPool()`. The executor runs each poll and the completion of the returned `CompletableFuture`. Any executor that hands tasks to its own threads works: the common pool, `Executors.newFixedThreadPool`, `newCachedThreadPool`, `newSingleThreadExecutor`, or a virtual-thread executor. Rust invokes the continuation from inside `Waker::wake()`, so an executor that runs tasks on the submitting thread (such as `Runnable::run`) polls the future from within its own waker and deadlocks any future that holds a lock while waking. Cancelling the returned future signals Rust with `rust_future_cancel`; the Rust future is dropped on the executor once its in-flight poll completes, not on the cancelling thread. - all primitives are signed in Java by default. Rust correctly interprets the a signed primitive value from Java as unsigned when told to. Callers of Uniffi functions need to be aware when making comparisons (`compareUnsigned`) or printing when a value is actually unsigned to code around footguns on this side. - this is an internal note for development but because Enum variants are not cases/hanging off their parent in Java, their named standalone, they can conflict with any/all `java.lang` types. We could do extensive checking and forced renaming around this, but instead we use fully qualified names for all `java.lang` types in all templates. Ensure that when you're making changes you're not dropping those qualified names or adding generated code without them. diff --git a/fixtures/async-lifecycle/Cargo.toml b/fixtures/async-lifecycle/Cargo.toml new file mode 100644 index 0000000..e67c396 --- /dev/null +++ b/fixtures/async-lifecycle/Cargo.toml @@ -0,0 +1,13 @@ +[package] +name = "uniffi-fixture-async-lifecycle" +version = "0.1.0" +edition = "2024" + +[lib] +crate-type = ["cdylib", "lib"] +name = "uniffi_fixture_async_lifecycle" + +[dependencies] +async-trait = "0.1" +parking_lot = "0.12" +uniffi = { git = "https://github.com/mozilla/uniffi-rs.git", branch = "release-v0.32.x" } diff --git a/fixtures/async-lifecycle/src/lib.rs b/fixtures/async-lifecycle/src/lib.rs new file mode 100644 index 0000000..631c2f8 --- /dev/null +++ b/fixtures/async-lifecycle/src/lib.rs @@ -0,0 +1,142 @@ +/* This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at http://mozilla.org/MPL/2.0/. */ + +//! Async functions whose futures count their own construction and drop. + +use parking_lot::Mutex; +use std::{ + future::Future, + pin::Pin, + sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + }, + task::{Context, Poll, Waker}, + thread, + time::Duration, +}; + +uniffi::setup_scaffolding!("async_lifecycle"); + +static CREATED: AtomicU64 = AtomicU64::new(0); +static DROPPED: AtomicU64 = AtomicU64::new(0); + +struct DropTracker; + +impl DropTracker { + fn new() -> Self { + CREATED.fetch_add(1, Ordering::SeqCst); + Self + } +} + +impl Drop for DropTracker { + fn drop(&mut self) { + DROPPED.fetch_add(1, Ordering::SeqCst); + } +} + +#[uniffi::export] +pub fn created_count() -> u64 { + CREATED.load(Ordering::SeqCst) +} + +#[uniffi::export] +pub fn dropped_count() -> u64 { + DROPPED.load(Ordering::SeqCst) +} + +#[uniffi::export] +pub fn reset_counts() { + CREATED.store(0, Ordering::SeqCst); + DROPPED.store(0, Ordering::SeqCst); +} + +struct TimerState { + completed: bool, + waker: Option, +} + +/// Completes after `duration` on a spawned thread, after `spurious_wakes` wakes that leave the +/// future pending; the foreign side re-polls once per wake. +struct Timer { + state: Arc>, +} + +impl Timer { + fn new(duration: Duration, spurious_wakes: u16) -> Self { + let state = Arc::new(Mutex::new(TimerState { + completed: false, + waker: None, + })); + let thread_state = Arc::clone(&state); + // Wakers are invoked with the lock released: an inline foreign executor re-polls from + // inside `wake()`, on this thread. + thread::spawn(move || { + for _ in 0..spurious_wakes { + thread::sleep(Duration::from_millis(1)); + let waker = thread_state.lock().waker.take(); + if let Some(waker) = waker { + waker.wake(); + } + } + thread::sleep(duration); + let waker = { + let mut state = thread_state.lock(); + state.completed = true; + state.waker.take() + }; + if let Some(waker) = waker { + waker.wake(); + } + }); + Self { state } + } +} + +impl Future for Timer { + type Output = (); + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { + let mut state = self.state.lock(); + if state.completed { + Poll::Ready(()) + } else { + state.waker = Some(cx.waker().clone()); + Poll::Pending + } + } +} + +#[uniffi::export] +pub async fn tracked_ready() { + let _tracker = DropTracker::new(); +} + +#[uniffi::export] +pub async fn tracked_sleep(ms: u16) { + let _tracker = DropTracker::new(); + Timer::new(Duration::from_millis(ms.into()), 0).await; +} + +#[uniffi::export] +pub async fn tracked_wake_storm(ms: u16, spurious_wakes: u16) { + let _tracker = DropTracker::new(); + Timer::new(Duration::from_millis(ms.into()), spurious_wakes).await; +} + +#[uniffi::export(with_foreign)] +#[async_trait::async_trait] +pub trait Notifier: Send + Sync { + async fn on_ready(&self); +} + +/// The sleep guarantees the foreign caller holds the future before `on_ready` is upcalled from +/// a re-poll. +#[uniffi::export] +pub async fn sleep_then_notify(ms: u16, notifier: Arc) { + let _tracker = DropTracker::new(); + Timer::new(Duration::from_millis(ms.into()), 0).await; + notifier.on_ready().await; +} diff --git a/src/gen_java/mod.rs b/src/gen_java/mod.rs index 827728b..be9ebf6 100644 --- a/src/gen_java/mod.rs +++ b/src/gen_java/mod.rs @@ -1906,6 +1906,16 @@ mod filters { Ok(format!("(_allocator, future, continuation) -> {call}")) } + #[askama::filter_fn] + pub fn async_cancel( + callable: impl Callable, + _v: &dyn askama::Values, + ci: &ComponentInterface, + ) -> Result { + let ffi_func = callable.ffi_rust_future_cancel(ci); + Ok(format!("(future) -> UniffiLib.{ffi_func}(future)")) + } + #[askama::filter_fn] pub fn async_free( callable: impl Callable, @@ -1933,11 +1943,48 @@ mod filters { _v: &dyn askama::Values, spaces: &i32, ) -> Result { - let middle = textwrap::indent(&textwrap::dedent(docstring.as_ref()), " * "); - let wrapped = format!("/**\n{middle}\n */"); + Ok(javadoc(&textwrap::dedent(docstring.as_ref()), *spaces)) + } - let spaces = usize::try_from(*spaces).unwrap_or_default(); - Ok(textwrap::indent(&wrapped, &" ".repeat(spaces))) + /// Javadoc for one overload of an async callable. + #[askama::filter_fn] + pub fn async_docstring( + callable: impl Callable, + _v: &dyn askama::Values, + spaces: &i32, + with_executor: bool, + ) -> Result { + let mut body = String::new(); + if let Some(docstring) = callable.docstring() { + body.push_str(&textwrap::dedent(docstring)); + body.push_str("\n\n"); + } + body.push_str(if with_executor { + ASYNC_EXECUTOR_PARAM_DOC + } else { + ASYNC_COMMON_POOL_DOC + }); + Ok(javadoc(&body, *spaces)) + } + + const ASYNC_COMMON_POOL_DOC: &str = "\ +Polls on {@link java.util.concurrent.ForkJoinPool#commonPool()}. See the overload taking an +{@link java.util.concurrent.Executor} for the contract an executor must meet."; + + const ASYNC_EXECUTOR_PARAM_DOC: &str = "\ +@param uniffiExecutor runs each poll and the completion of the returned future. Any executor + that hands tasks to its own threads works: {@link java.util.concurrent.ForkJoinPool#commonPool()}, + a fixed, cached or single-thread pool, or a virtual-thread executor. Rust invokes the + continuation from inside {@code Waker::wake()}, so an executor that runs tasks on the + submitting thread, such as {@code Runnable::run}, polls the future from within its own waker + and deadlocks any future that holds a lock while waking. Cancelling the returned future + signals Rust; the Rust future is dropped on this executor once its in-flight poll completes."; + + fn javadoc(body: &str, spaces: i32) -> String { + let middle = textwrap::indent(body, " * "); + let wrapped = format!("/**\n{middle}\n */"); + let spaces = usize::try_from(spaces).unwrap_or_default(); + textwrap::indent(&wrapped, &" ".repeat(spaces)) } /// Returns the type name suitable for use in field declarations, method parameters, and return types. diff --git a/src/templates/Async.java b/src/templates/Async.java index 5641e5a..d94d93b 100644 --- a/src/templates/Async.java +++ b/src/templates/Async.java @@ -44,67 +44,52 @@ interface AsyncCompleteVoidFunction { void apply(java.lang.foreign.SegmentAllocator allocator, long rustFuture, java.lang.foreign.MemorySegment status); } - // Every use of the handle holds the monitor and checks `freed` first. cancel() frees on the - // calling thread while the poll/complete pipeline may be mid-call on an executor thread. + // The pipeline is the sole freer; cancel() only signals Rust, which fires the in-flight poll's + // continuation and makes later polls return Ready. `lock` covers only rust_future_cancel and + // rust_future_free, neither of which runs user code. A failed tryLock means free() is in + // progress and there is nothing left to cancel. static final class UniffiFreeingFuture extends java.util.concurrent.CompletableFuture { private final long rustFuture; + private final java.util.function.Consumer cancelFunc; private final java.util.function.Consumer freeFunc; + private final java.util.concurrent.locks.ReentrantLock lock = new java.util.concurrent.locks.ReentrantLock(); private boolean freed; - UniffiFreeingFuture(long rustFuture, java.util.function.Consumer freeFunc) { + UniffiFreeingFuture( + long rustFuture, + java.util.function.Consumer cancelFunc, + java.util.function.Consumer freeFunc + ) { this.rustFuture = rustFuture; + this.cancelFunc = cancelFunc; this.freeFunc = freeFunc; } @Override public boolean cancel(boolean mayInterruptIfRunning) { boolean cancelled = super.cancel(mayInterruptIfRunning); - if (cancelled) { - free(); + if (cancelled && lock.tryLock()) { + try { + if (!freed) { + cancelFunc.accept(rustFuture); + } + } finally { + lock.unlock(); + } } return cancelled; } - synchronized void free() { - if (freed) { - return; - } - freed = true; - freeFunc.accept(rustFuture); - } - - // Returns false, without polling, once freed. - synchronized boolean poll(PollingFunction pollFunc, long continuationHandle) { - if (freed) { - return false; - } - pollFunc.apply(rustFuture, CONTINUATION_CALLBACK_STUB, continuationHandle); - return true; - } - - // Returns null once freed, which the pipeline only reaches after cancel(). - synchronized F completeRust( - AsyncCompleteFunction completeFunc, - UniffiRustCallStatusErrorHandler errorHandler - ) throws E { - if (freed) { - return null; - } - return UniffiHelpers.uniffiRustCallWithError(errorHandler, (_allocator, status) -> { - return completeFunc.apply(_allocator, rustFuture, status); - }); - } - - synchronized void completeRustVoid( - AsyncCompleteVoidFunction completeFunc, - UniffiRustCallStatusErrorHandler errorHandler - ) throws E { - if (freed) { - return; + void free() { + lock.lock(); + try { + if (!freed) { + freed = true; + freeFunc.accept(rustFuture); + } + } finally { + lock.unlock(); } - UniffiHelpers.uniffiRustCallWithError(errorHandler, (_allocator, status) -> { - completeFunc.apply(_allocator, rustFuture, status); - }); } } @@ -132,15 +117,16 @@ static java.util.concurrent.CompletableFut long rustFuture, PollingFunction pollFunc, AsyncCompleteFunction completeFunc, + java.util.function.Consumer cancelFunc, java.util.function.Consumer freeFunc, java.util.function.Function liftFunc, UniffiRustCallStatusErrorHandler errorHandler ){ - UniffiFreeingFuture future = new UniffiFreeingFuture<>(rustFuture, freeFunc); + UniffiFreeingFuture future = new UniffiFreeingFuture<>(rustFuture, cancelFunc, freeFunc); java.util.concurrent.CompletableFuture pollChain; try { - pollChain = pollUntilReady(future, pollFunc, uniffiExecutor); + pollChain = pollUntilReady(rustFuture, pollFunc, uniffiExecutor); } catch (java.lang.Exception e) { future.completeExceptionally(e); future.free(); @@ -152,8 +138,10 @@ static java.util.concurrent.CompletableFut return null; } try { - F result = future.completeRust(completeFunc, errorHandler); - return result == null ? null : liftFunc.apply(result); + F result = UniffiHelpers.uniffiRustCallWithError(errorHandler, (_allocator, status) -> { + return completeFunc.apply(_allocator, rustFuture, status); + }); + return liftFunc.apply(result); } catch (java.lang.Exception e) { throw new java.util.concurrent.CompletionException(e); } @@ -184,15 +172,16 @@ static java.util.concurrent.CompletableFuture cancelFunc, java.util.function.Consumer freeFunc, java.lang.Runnable liftFunc, UniffiRustCallStatusErrorHandler errorHandler ){ - UniffiFreeingFuture future = new UniffiFreeingFuture<>(rustFuture, freeFunc); + UniffiFreeingFuture future = new UniffiFreeingFuture<>(rustFuture, cancelFunc, freeFunc); java.util.concurrent.CompletableFuture pollChain; try { - pollChain = pollUntilReady(future, pollFunc, uniffiExecutor); + pollChain = pollUntilReady(rustFuture, pollFunc, uniffiExecutor); } catch (java.lang.Exception e) { future.completeExceptionally(e); future.free(); @@ -204,7 +193,9 @@ static java.util.concurrent.CompletableFuture { + completeFunc.apply(_allocator, rustFuture, status); + }); } catch (java.lang.Exception e) { throw new java.util.concurrent.CompletionException(e); } @@ -228,18 +219,15 @@ static java.util.concurrent.CompletableFuture pollUntilReady(UniffiFreeingFuture future, PollingFunction pollFunc, java.util.concurrent.Executor uniffiExecutor) { + private static java.util.concurrent.CompletableFuture pollUntilReady(long rustFuture, PollingFunction pollFunc, java.util.concurrent.Executor uniffiExecutor) { java.util.concurrent.CompletableFuture pollFuture = new java.util.concurrent.CompletableFuture<>(); var handle = uniffiContinuationHandleMap.insert(pollFuture); - if (!future.poll(pollFunc, handle)) { - uniffiContinuationHandleMap.remove(handle); - return java.util.concurrent.CompletableFuture.completedFuture(null); - } + pollFunc.apply(rustFuture, CONTINUATION_CALLBACK_STUB, handle); return pollFuture.thenComposeAsync(pollResult -> { if (pollResult == UNIFFI_RUST_FUTURE_POLL_READY) { return java.util.concurrent.CompletableFuture.completedFuture(null); } else { - return pollUntilReady(future, pollFunc, uniffiExecutor); + return pollUntilReady(rustFuture, pollFunc, uniffiExecutor); } }, uniffiExecutor); } diff --git a/src/templates/macros.java b/src/templates/macros.java index 4804d37..acb8239 100644 --- a/src/templates/macros.java +++ b/src/templates/macros.java @@ -73,11 +73,11 @@ {%- endmacro -%} {%- macro func_decl(func_decl, annotation, callable, indent) %} - {%- call docstring(callable, indent) %}{% endcall %} + {%- if callable.is_async() %} + {{ callable|async_docstring(indent, false) }} {%- if annotation != "" %} @{{ annotation }} {% endif %} - {%- if callable.is_async() %} {#- Async methods use CompletableFuture which requires boxed types -#} {#- No-executor overload - defaults to ForkJoinPool.commonPool(), delegates to Executor version -#} {{ func_decl }} java.util.concurrent.CompletableFuture<{% match callable.return_type() -%}{%- when Some with (return_type) -%}{{ return_type|boxed_type_name(ci, config) }}{%- when None %}java.lang.Void{%- endmatch %}> {{ callable.name()|fn_name }}( @@ -87,12 +87,17 @@ } {#- With-executor overload - does the actual async work -#} + {{ callable|async_docstring(indent, true) }} {{ func_decl }} java.util.concurrent.CompletableFuture<{% match callable.return_type() -%}{%- when Some with (return_type) -%}{{ return_type|boxed_type_name(ci, config) }}{%- when None %}java.lang.Void{%- endmatch %}> {{ callable.name()|fn_name }}( {%- call arg_list(callable, !callable.self_type().is_some()) %}{% endcall -%}{% if !callable.arguments().is_empty() %}, {% endif %}java.util.concurrent.Executor uniffiExecutor ){ return {% call call_async(callable) %}{% endcall %}; } {%- else -%} + {%- call docstring(callable, indent) %}{% endcall %} + {%- if annotation != "" %} + @{{ annotation }} + {% endif %} {#- Sync methods can use primitives for return types -#} {{ func_decl }} {% match callable.return_type() -%}{%- when Some with (return_type) -%}{{ return_type|type_name_for_field(ci, config) }}{%- when None %}void{%- endmatch %} {{ callable.name()|fn_name }}( {%- call arg_list(callable, !callable.self_type().is_some()) %}{% endcall -%} @@ -148,6 +153,7 @@ {%- endmatch %} {{ callable|async_poll(ci) }}, {{ callable|async_complete(ci, config) }}, + {{ callable|async_cancel(ci) }}, {{ callable|async_free(ci) }}, // lift function {%- match callable.return_type() %} diff --git a/tests/scripts/TestAsyncLifecycle/TestAsyncLifecycle.java b/tests/scripts/TestAsyncLifecycle/TestAsyncLifecycle.java new file mode 100644 index 0000000..c8519fb --- /dev/null +++ b/tests/scripts/TestAsyncLifecycle/TestAsyncLifecycle.java @@ -0,0 +1,90 @@ +import uniffi.async_lifecycle.*; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicReference; + +public class TestAsyncLifecycle { + // Only a leak exhausts the bound. + static void awaitAllDropped(String label) throws Exception { + long deadline = System.nanoTime() + 10_000_000_000L; + while (AsyncLifecycle.droppedCount() != AsyncLifecycle.createdCount()) { + if (System.nanoTime() > deadline) { + throw new AssertionError(label + ": created=" + AsyncLifecycle.createdCount() + + " dropped=" + AsyncLifecycle.droppedCount()); + } + Thread.sleep(5); + } + } + + public static void main(String[] args) throws Exception { + { + AsyncLifecycle.resetCounts(); + AsyncLifecycle.trackedReady().get(); + AsyncLifecycle.trackedSleep((short)1).get(); + AsyncLifecycle.trackedWakeStorm((short)1, (short)3).get(); + AsyncLifecycle.trackedSleep((short)50).cancel(true); + AsyncLifecycle.trackedWakeStorm((short)50, (short)5).cancel(true); + awaitAllDropped("single futures"); + assert AsyncLifecycle.createdCount() == 5 : "expected 5 created, got " + AsyncLifecycle.createdCount(); + System.out.println("drop accounting (5 futures) ... ok"); + } + + { + AsyncLifecycle.resetCounts(); + int threads = 8, iters = 5_000; + var failure = new AtomicReference(); + Thread[] workers = new Thread[threads]; + for (int t = 0; t < threads; t++) { + final int seed = t; + workers[t] = new Thread(() -> { + try { + for (int i = 0; i < iters; i++) { + CompletableFuture job = switch ((i + seed) % 3) { + case 0 -> AsyncLifecycle.trackedSleep((short)1); + case 1 -> AsyncLifecycle.trackedWakeStorm((short)1, (short)1); + default -> AsyncLifecycle.trackedReady(); + }; + if (i % 2 == 0) { + Thread.yield(); + } + job.cancel(true); + } + } catch (Throwable e) { + failure.compareAndSet(null, e); + } + }); + workers[t].start(); + } + for (Thread w : workers) { + w.join(); + } + assert failure.get() == null : "cancel storm threw: " + failure.get(); + awaitAllDropped("cancel storm"); + assert AsyncLifecycle.createdCount() == (long) threads * iters; + System.out.println("drop accounting after cancel storm (" + threads + " x " + iters + ") ... ok"); + } + + // cancel() from inside an async callback, i.e. on the executor thread while rust_future_poll + // is on the stack. Freeing here would deadlock on the future's own Rust mutex. + { + AsyncLifecycle.resetCounts(); + var job = new AtomicReference>(); + Notifier cancelSelf = () -> { + boolean cancelled = job.get().cancel(true); + assert cancelled : "cancel from callback returned false"; + return CompletableFuture.completedFuture(null); + }; + var f = AsyncLifecycle.sleepThenNotify((short)10, cancelSelf); + job.set(f); + try { + f.get(); + throw new AssertionError("expected cancellation"); + } catch (java.util.concurrent.CancellationException expected) { + } + awaitAllDropped("cancel from callback"); + System.out.println("cancel from inside async callback ... ok"); + } + + uniffi.async_lifecycle.AsyncContract.run(); + } +} diff --git a/tests/scripts/TestAsyncLifecycle/uniffi/async_lifecycle/AsyncContract.java b/tests/scripts/TestAsyncLifecycle/uniffi/async_lifecycle/AsyncContract.java new file mode 100644 index 0000000..ddbff26 --- /dev/null +++ b/tests/scripts/TestAsyncLifecycle/uniffi/async_lifecycle/AsyncContract.java @@ -0,0 +1,209 @@ +package uniffi.async_lifecycle; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +// Asserts the handle contract uniffi_core documents: free exactly once, and no poll, complete or +// cancel on a handle after it has been freed. +public final class AsyncContract { + static final class Ledger { + final AtomicInteger polls = new AtomicInteger(); + final AtomicInteger completes = new AtomicInteger(); + final AtomicInteger cancels = new AtomicInteger(); + final AtomicInteger frees = new AtomicInteger(); + volatile boolean freed; + // Pipeline stage in progress: at most one thread, re-entrant on an inline executor. + volatile Thread inFlight; + int depth; + } + + // Handle values are heap addresses and are reused after free, so ledgers are per call. + static final ConcurrentLinkedQueue ledgers = new ConcurrentLinkedQueue<>(); + static final ConcurrentLinkedQueue violations = new ConcurrentLinkedQueue<>(); + + static void violate(String what, long handle) { + violations.add(what + " (handle " + Long.toHexString(handle) + ", " + Thread.currentThread().getName() + ")"); + } + + // Returns whether the native call may proceed; a call on a freed handle is skipped. + static boolean enter(Ledger l, String stage, long handle) { + boolean live = !l.freed; + if (!live) { + violate(stage + " after free", handle); + } + Thread me = Thread.currentThread(); + synchronized (l) { + if (l.inFlight != null && l.inFlight != me) { + violate(stage + " concurrent with another pipeline stage", handle); + } + l.inFlight = me; + l.depth++; + } + return live; + } + + static void exit(Ledger l) { + synchronized (l) { + if (--l.depth == 0) { + l.inFlight = null; + } + } + } + + static CompletableFuture call(long handle, Executor executor) { + Ledger l = new Ledger(); + ledgers.add(l); + return UniffiAsyncHelpers.uniffiRustCallAsync( + executor, + handle, + (future, callback, continuation) -> { + boolean live = enter(l, "poll", future); + try { + l.polls.incrementAndGet(); + // Native calls are serialized per handle only after the overlap has been + // recorded, so a violation is reported instead of executed. + synchronized (l) { + if (live && !l.freed) { + UniffiLib.ffi_uniffi_fixture_async_lifecycle_rust_future_poll_void(future, callback, continuation); + } else { + // Keeps the pipeline draining so the run ends and reports. + UniffiAsyncHelpers.continuationCallback(continuation, UniffiAsyncHelpers.UNIFFI_RUST_FUTURE_POLL_READY); + } + } + } finally { + exit(l); + } + }, + (_allocator, future, status) -> { + boolean live = enter(l, "complete", future); + try { + if (l.completes.incrementAndGet() > 1) { + violate("complete twice", future); + live = false; + } + synchronized (l) { + if (live && !l.freed) { + UniffiLib.ffi_uniffi_fixture_async_lifecycle_rust_future_complete_void(future, status); + } + } + } finally { + exit(l); + } + }, + (future) -> { + l.cancels.incrementAndGet(); + synchronized (l) { + if (l.freed) { + violate("cancel after free", future); + return; + } + UniffiLib.ffi_uniffi_fixture_async_lifecycle_rust_future_cancel_void(future); + } + }, + (future) -> { + if (l.frees.incrementAndGet() > 1) { + violate("free twice", future); + return; + } + if (l.inFlight != null && l.inFlight != Thread.currentThread()) { + violate("free while a pipeline stage is in flight on another thread", future); + } + synchronized (l) { + l.freed = true; + UniffiLib.ffi_uniffi_fixture_async_lifecycle_rust_future_free_void(future); + } + }, + () -> {}, + new UniffiNullRustCallStatusErrorHandler() + ); + } + + static long newHandle(int kind) { + return switch (kind % 3) { + case 0 -> UniffiLib.uniffi_uniffi_fixture_async_lifecycle_fn_func_tracked_ready(); + case 1 -> UniffiLib.uniffi_uniffi_fixture_async_lifecycle_fn_func_tracked_sleep((short) 1); + default -> UniffiLib.uniffi_uniffi_fixture_async_lifecycle_fn_func_tracked_wake_storm((short) 1, (short) 2); + }; + } + + // Cancel timings vary so cancel lands at every pipeline stage. + static void storm(String name, Executor executor, int threads, int iters, ScheduledExecutorService canceller) throws Exception { + ledgers.clear(); + violations.clear(); + List workers = new ArrayList<>(); + List> futures = new java.util.concurrent.CopyOnWriteArrayList<>(); + for (int t = 0; t < threads; t++) { + final int seed = t; + Thread w = new Thread(() -> { + var rnd = ThreadLocalRandom.current(); + for (int i = 0; i < iters; i++) { + var f = call(newHandle(i + seed), executor); + futures.add(f); + switch (rnd.nextInt(4)) { + case 0 -> {} + case 1 -> f.cancel(true); + case 2 -> { + Thread.yield(); + f.cancel(true); + } + default -> canceller.schedule(() -> f.cancel(true), rnd.nextInt(3), TimeUnit.MILLISECONDS); + } + } + }, name + "-worker-" + t); + workers.add(w); + w.start(); + } + for (Thread w : workers) { + w.join(); + } + for (var f : futures) { + try { + f.get(10, TimeUnit.SECONDS); + } catch (java.util.concurrent.CancellationException | java.util.concurrent.ExecutionException expected) { + } + } + // free runs after the future completes. + long deadline = System.nanoTime() + 10_000_000_000L; + while (ledgers.stream().anyMatch(l -> l.frees.get() == 0)) { + if (System.nanoTime() > deadline) { + long leaked = ledgers.stream().filter(l -> l.frees.get() == 0).count(); + throw new AssertionError(name + ": " + leaked + " handles never freed"); + } + Thread.sleep(5); + } + if (!violations.isEmpty()) { + throw new AssertionError(name + ": " + violations.size() + " contract violations, first: " + violations.peek()); + } + System.out.println("async handle contract [" + name + "] (" + ledgers.size() + " futures) ... ok"); + } + + public static void run() throws Exception { + ScheduledExecutorService canceller = Executors.newScheduledThreadPool(2); + ExecutorService single = Executors.newSingleThreadExecutor(); + ScheduledExecutorService jitterPool = Executors.newScheduledThreadPool(4); + Executor jittery = r -> jitterPool.schedule(r, ThreadLocalRandom.current().nextInt(3), TimeUnit.MILLISECONDS); + Executor rejecting = r -> { throw new RejectedExecutionException("rejecting executor"); }; + try { + storm("common pool", java.util.concurrent.ForkJoinPool.commonPool(), 8, 5_000, canceller); + storm("inline executor", Runnable::run, 8, 2_000, canceller); + storm("single thread executor", single, 4, 2_000, canceller); + storm("jittery executor", jittery, 4, 500, canceller); + storm("rejecting executor", rejecting, 4, 500, canceller); + } finally { + canceller.shutdownNow(); + single.shutdownNow(); + jitterPool.shutdownNow(); + } + } +} diff --git a/tests/tests.rs b/tests/tests.rs index 894d0a0..52e01cc 100644 --- a/tests/tests.rs +++ b/tests/tests.rs @@ -116,9 +116,12 @@ fn run_test(fixture_name: &str, test_file: &str) -> Result<()> { let jar_file = build_jar(fixture_name, &out_dir)?; // compile test + // Helper sources in a package directory beside the test compile from here, e.g. to reach + // package-private generated helpers. + let test_dir = test_path.parent().unwrap().to_path_buf(); let status = Command::new("javac") .arg("-classpath") - .arg(calc_classpath(vec![&out_dir, &jar_file])) + .arg(calc_classpath(vec![&out_dir, &jar_file, &test_dir])) // Our tests should not produce any warnings. .arg("-Werror") .arg(&test_path) @@ -132,7 +135,7 @@ fn run_test(fixture_name: &str, test_file: &str) -> Result<()> { // run resulting test let compiled_path = test_path.file_stem().unwrap(); - let run_status = Command::new("java") + let run_status = java_command() // allow for runtime assertions .arg("-ea") // Enable FFM native access @@ -140,11 +143,7 @@ fn run_test(fixture_name: &str, test_file: &str) -> Result<()> { // Set native library path so System.loadLibrary can find the cdylib .arg(format!("-Djava.library.path={}", native_lib_dir)) .arg("-classpath") - .arg(calc_classpath(vec![ - &out_dir, - &jar_file, - &test_path.parent().unwrap().to_path_buf(), - ])) + .arg(calc_classpath(vec![&out_dir, &jar_file, &test_dir])) .arg(compiled_path) .spawn() .context("Failed to spawn `java` to run Java test")? @@ -222,7 +221,7 @@ fn run_test_with_library_override( // Run with library override set to an absolute path and NO java.library.path, // so this can only work if the generated code uses System.load() for absolute paths. let compiled_path = test_path.file_stem().unwrap(); - let run_status = Command::new("java") + let run_status = java_command() .arg("-ea") .arg("--enable-native-access=ALL-UNNAMED") .arg(format!( @@ -248,6 +247,18 @@ fn run_test_with_library_override( Ok(()) } +/// `java` with allocator debug checks that abort on a stale write into freed native memory. Each +/// platform ignores the other's variables. +fn java_command() -> Command { + let mut cmd = Command::new("java"); + // glibc: fill freed memory with a pattern. + cmd.env("GLIBC_TUNABLES", "glibc.malloc.perturb=165"); + // macOS libmalloc: fill freed memory and guard large allocations. + cmd.env("MallocScribble", "1"); + cmd.env("MallocGuardEdges", "1"); + cmd +} + /// Get the uniffi_toml of the fixture if it exists. /// It looks for it in the root directory of the project `name`. fn find_uniffi_toml(name: &str) -> Result> { @@ -391,6 +402,7 @@ fixture_tests! { (test_proc_macro, "uniffi-fixture-proc-macro", "scripts/TestProcMacro.java"), (test_rename, "uniffi-fixture-rename", "scripts/TestRename/TestRename.java"), (test_primitive_arrays, "uniffi-fixture-primitive-arrays", "scripts/TestPrimitiveArrays.java"), + (test_async_lifecycle, "uniffi-fixture-async-lifecycle", "scripts/TestAsyncLifecycle/TestAsyncLifecycle.java"), (test_zero_copy, "uniffi-fixture-zero-copy", "scripts/TestZeroCopy.java"), } From 7d1cb82610858a2d8ed3ff7a95d75aa2dfe486ce Mon Sep 17 00:00:00 2001 From: Murph Murphy Date: Wed, 16 Sep 2026 17:28:49 -0600 Subject: [PATCH 2/2] Another review pass --- CHANGELOG.md | 2 +- Cargo.lock | 48 ------------------- Cargo.toml | 14 +++--- README.md | 2 +- fixtures/async-lifecycle/Cargo.toml | 1 - fixtures/async-lifecycle/src/lib.rs | 13 +++-- src/gen_java/mod.rs | 21 ++++---- src/templates/Async.java | 15 ++++-- .../uniffi/async_lifecycle/AsyncContract.java | 19 ++++---- tests/tests.rs | 8 ++-- 10 files changed, 53 insertions(+), 90 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c9b313c..1167c9a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,6 @@ ## 0.5.2 -- fix a narrow race in async cancellation. If `cancel()` on a `CompletableFuture` returned by an async function landed in the few instructions between the pipeline's `isCancelled()` check and its own `rust_future_free`, or between a wake and the re-poll it triggers, the Rust future was freed twice or polled after being freed. Seen once in CI as a glibc `tcache_thread_shutdown()` abort. `cancel()` now signals Rust with `rust_future_cancel` and the poll/complete pipeline is the only thing that frees. The Rust future is dropped once the in-flight poll's continuation fires, on an executor thread rather than the cancelling one. +- fix a narrow race in async cancellation. If `cancel()` on a `CompletableFuture` returned by an async function landed in the few instructions between the pipeline's `isCancelled()` check and its own `rust_future_free`, or between a wake and the re-poll it triggers, the Rust future was freed twice or polled after being freed. Seen once in CI as a glibc `tcache_thread_shutdown()` abort. `cancel()` now signals Rust with `rust_future_cancel` and the poll/complete pipeline is the only thing that frees. The Rust future is dropped once the in-flight poll's continuation fires, on whichever thread the executor runs that continuation on; with a pool that is an executor thread rather than the cancelling one. Cancelling with an executor that accepts a task and never runs it now leaks the Rust future, where before `cancel()` freed it inline; see the executor notes in the README. - fix generated Java failing to compile when a Rust field or parameter name matched a name the generator used in the same scope, such as an enum variant field named `value` ([#63](https://github.com/IronCoreLabs/uniffi-bindgen-java/issues/63)). Enum and error variant field names, and function, method, constructor and callback-interface parameter names can no longer collide with anything the Java generator emits. Names uniffi-rs itself reserves on the Rust side, such as a field named `buf` or a callback-interface parameter named `uniffi_handle`, are rejected by the Rust derive before bindings are generated and remain unavailable. ## 0.5.1 diff --git a/Cargo.lock b/Cargo.lock index 783821f..529cb76 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1017,15 +1017,6 @@ version = "0.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae" -[[package]] -name = "lock_api" -version = "0.4.14" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" -dependencies = [ - "scopeguard", -] - [[package]] name = "log" version = "0.4.34" @@ -1090,29 +1081,6 @@ version = "2.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" -[[package]] -name = "parking_lot" -version = "0.12.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" -dependencies = [ - "lock_api", - "parking_lot_core", -] - -[[package]] -name = "parking_lot_core" -version = "0.9.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" -dependencies = [ - "cfg-if", - "libc", - "redox_syscall", - "smallvec", - "windows-link", -] - [[package]] name = "paste" version = "1.0.15" @@ -1251,15 +1219,6 @@ dependencies = [ "crossbeam-utils", ] -[[package]] -name = "redox_syscall" -version = "0.5.18" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" -dependencies = [ - "bitflags", -] - [[package]] name = "regex" version = "1.13.1" @@ -1323,12 +1282,6 @@ dependencies = [ "winapi-util", ] -[[package]] -name = "scopeguard" -version = "1.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" - [[package]] name = "scroll" version = "0.12.0" @@ -1739,7 +1692,6 @@ name = "uniffi-fixture-async-lifecycle" version = "0.1.0" dependencies = [ "async-trait", - "parking_lot", "uniffi", ] diff --git a/Cargo.toml b/Cargo.toml index a47f5d9..7f05bfc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,17 +24,17 @@ bench = false [dependencies] anyhow = "1" askama = { version = "0.16", default-features = false, features = [ - "config", - "derive", - "alloc", + "config", + "derive", + "alloc", ] } camino = "1.1.6" cargo_metadata = "0.23" clap = { version = "4", default-features = false, features = [ - "derive", - "help", - "std", - "cargo", + "derive", + "help", + "std", + "cargo", ] } heck = "0.5" once_cell = "1.19.0" diff --git a/README.md b/README.md index c6e4dc1..aa7f289 100644 --- a/README.md +++ b/README.md @@ -236,7 +236,7 @@ scope). - a Rust `&[u8]` argument is borrowed by Rust for the duration of the call rather than copied, and takes a **direct** `java.nio.ByteBuffer` (as in Kotlin) instead of `byte[]`. Build one with `ByteBuffer.allocateDirect(arr.length).put(arr).flip()`; a heap buffer throws `IllegalArgumentException`. Reuse the buffer across calls where you can, and don't let another thread write to it while a call is in flight. `Vec` is unaffected and still maps to `byte[]`. - failures in CompletableFutures will cause them to `completeExceptionally`. The error that caused the failure can be checked with `e.getCause()`. When implementing an async Rust trait in Java, you'll need to `completeExceptionally` instead of throwing. See `TestFixtureFutures.java` for an example trait implementation with errors. -- every async function and method has an overload taking a `java.util.concurrent.Executor`; the other overload uses `ForkJoinPool.commonPool()`. The executor runs each poll and the completion of the returned `CompletableFuture`. Any executor that hands tasks to its own threads works: the common pool, `Executors.newFixedThreadPool`, `newCachedThreadPool`, `newSingleThreadExecutor`, or a virtual-thread executor. Rust invokes the continuation from inside `Waker::wake()`, so an executor that runs tasks on the submitting thread (such as `Runnable::run`) polls the future from within its own waker and deadlocks any future that holds a lock while waking. Cancelling the returned future signals Rust with `rust_future_cancel`; the Rust future is dropped on the executor once its in-flight poll completes, not on the cancelling thread. +- every async function and method has an overload taking a `java.util.concurrent.Executor`; the other overload uses `ForkJoinPool.commonPool()`. The first poll runs on the calling thread, before the method returns; the executor runs every re-poll and the completion of the returned `CompletableFuture`. Any executor that hands tasks to its own threads works: the common pool, `Executors.newFixedThreadPool`, `newCachedThreadPool`, `newSingleThreadExecutor`, or a virtual-thread executor. Rust invokes the continuation from inside `Waker::wake()`, so an executor that runs tasks on the submitting thread (such as `Runnable::run`) polls the future from within its own waker and deadlocks any future that holds a lock while waking. Cancelling the returned future signals Rust with `rust_future_cancel` and the poll/complete pipeline frees the Rust future once the in-flight poll's continuation fires, on whichever thread the executor runs that continuation on. Two consequences: an executor that accepts a task and never runs it (`ThreadPoolExecutor.DiscardPolicy`, or a queue drained by `shutdownNow()`) leaks the Rust future and everything it owns, and with an inline executor the free happens on the cancelling thread. - all primitives are signed in Java by default. Rust correctly interprets the a signed primitive value from Java as unsigned when told to. Callers of Uniffi functions need to be aware when making comparisons (`compareUnsigned`) or printing when a value is actually unsigned to code around footguns on this side. - this is an internal note for development but because Enum variants are not cases/hanging off their parent in Java, their named standalone, they can conflict with any/all `java.lang` types. We could do extensive checking and forced renaming around this, but instead we use fully qualified names for all `java.lang` types in all templates. Ensure that when you're making changes you're not dropping those qualified names or adding generated code without them. diff --git a/fixtures/async-lifecycle/Cargo.toml b/fixtures/async-lifecycle/Cargo.toml index e67c396..ec922c9 100644 --- a/fixtures/async-lifecycle/Cargo.toml +++ b/fixtures/async-lifecycle/Cargo.toml @@ -9,5 +9,4 @@ name = "uniffi_fixture_async_lifecycle" [dependencies] async-trait = "0.1" -parking_lot = "0.12" uniffi = { git = "https://github.com/mozilla/uniffi-rs.git", branch = "release-v0.32.x" } diff --git a/fixtures/async-lifecycle/src/lib.rs b/fixtures/async-lifecycle/src/lib.rs index 631c2f8..89dfe6e 100644 --- a/fixtures/async-lifecycle/src/lib.rs +++ b/fixtures/async-lifecycle/src/lib.rs @@ -4,12 +4,11 @@ //! Async functions whose futures count their own construction and drop. -use parking_lot::Mutex; use std::{ future::Future, pin::Pin, sync::{ - Arc, + Arc, Mutex, atomic::{AtomicU64, Ordering}, }, task::{Context, Poll, Waker}, @@ -58,8 +57,8 @@ struct TimerState { waker: Option, } -/// Completes after `duration` on a spawned thread, after `spurious_wakes` wakes that leave the -/// future pending; the foreign side re-polls once per wake. +/// Completes after `duration` on a spawned thread, having woken the future `spurious_wakes` times +/// first. A wake with no waker stored yet is dropped. struct Timer { state: Arc>, } @@ -76,14 +75,14 @@ impl Timer { thread::spawn(move || { for _ in 0..spurious_wakes { thread::sleep(Duration::from_millis(1)); - let waker = thread_state.lock().waker.take(); + let waker = thread_state.lock().unwrap().waker.take(); if let Some(waker) = waker { waker.wake(); } } thread::sleep(duration); let waker = { - let mut state = thread_state.lock(); + let mut state = thread_state.lock().unwrap(); state.completed = true; state.waker.take() }; @@ -99,7 +98,7 @@ impl Future for Timer { type Output = (); fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> { - let mut state = self.state.lock(); + let mut state = self.state.lock().unwrap(); if state.completed { Poll::Ready(()) } else { diff --git a/src/gen_java/mod.rs b/src/gen_java/mod.rs index 0493f39..ce87672 100644 --- a/src/gen_java/mod.rs +++ b/src/gen_java/mod.rs @@ -1971,17 +1971,20 @@ mod filters { } const ASYNC_COMMON_POOL_DOC: &str = "\ -Polls on {@link java.util.concurrent.ForkJoinPool#commonPool()}. See the overload taking an -{@link java.util.concurrent.Executor} for the contract an executor must meet."; +Re-polls and completes on {@link java.util.concurrent.ForkJoinPool#commonPool()}. See the overload +taking an {@link java.util.concurrent.Executor} for the contract an executor must meet."; const ASYNC_EXECUTOR_PARAM_DOC: &str = "\ -@param uniffiExecutor runs each poll and the completion of the returned future. Any executor - that hands tasks to its own threads works: {@link java.util.concurrent.ForkJoinPool#commonPool()}, - a fixed, cached or single-thread pool, or a virtual-thread executor. Rust invokes the - continuation from inside {@code Waker::wake()}, so an executor that runs tasks on the - submitting thread, such as {@code Runnable::run}, polls the future from within its own waker - and deadlocks any future that holds a lock while waking. Cancelling the returned future - signals Rust; the Rust future is dropped on this executor once its in-flight poll completes."; +@param _uniffiExecutor runs every re-poll and the completion of the returned future. The first + poll runs on the calling thread, before this method returns. Any executor that hands tasks to + its own threads works: {@link java.util.concurrent.ForkJoinPool#commonPool()}, a fixed, cached + or single-thread pool, or a virtual-thread executor. Rust invokes the continuation from inside + {@code Waker::wake()}, so an executor that runs tasks on the submitting thread, such as + {@code Runnable::run}, polls the future from within its own waker and deadlocks any future + that holds a lock while waking. Cancelling the returned future signals Rust; the pipeline then + frees the Rust future, so an executor that accepts a task and never runs it, such as one using + {@link java.util.concurrent.ThreadPoolExecutor.DiscardPolicy} or one whose queue was drained + by {@code shutdownNow()}, leaks the Rust future and everything it owns."; fn javadoc(body: &str, spaces: i32) -> String { let middle = textwrap::indent(body, " * "); diff --git a/src/templates/Async.java b/src/templates/Async.java index d94d93b..042297e 100644 --- a/src/templates/Async.java +++ b/src/templates/Async.java @@ -45,9 +45,10 @@ interface AsyncCompleteVoidFunction { } // The pipeline is the sole freer; cancel() only signals Rust, which fires the in-flight poll's - // continuation and makes later polls return Ready. `lock` covers only rust_future_cancel and - // rust_future_free, neither of which runs user code. A failed tryLock means free() is in - // progress and there is nothing left to cancel. + // continuation and makes later polls return Ready. A failed tryLock means free() is in progress + // and there is nothing left to cancel. Under an inline executor rust_future_cancel invokes that + // continuation synchronously, so the critical section extends through the completion pipeline + // and a reentrant free(). static final class UniffiFreeingFuture extends java.util.concurrent.CompletableFuture { private final long rustFuture; private final java.util.function.Consumer cancelFunc; @@ -222,7 +223,13 @@ static java.util.concurrent.CompletableFuture pollUntilReady(long rustFuture, PollingFunction pollFunc, java.util.concurrent.Executor uniffiExecutor) { java.util.concurrent.CompletableFuture pollFuture = new java.util.concurrent.CompletableFuture<>(); var handle = uniffiContinuationHandleMap.insert(pollFuture); - pollFunc.apply(rustFuture, CONTINUATION_CALLBACK_STUB, handle); + try { + pollFunc.apply(rustFuture, CONTINUATION_CALLBACK_STUB, handle); + } catch (java.lang.Throwable e) { + // Rust never took the handle, so nothing else will remove it. + uniffiContinuationHandleMap.remove(handle); + throw e; + } return pollFuture.thenComposeAsync(pollResult -> { if (pollResult == UNIFFI_RUST_FUTURE_POLL_READY) { return java.util.concurrent.CompletableFuture.completedFuture(null); diff --git a/tests/scripts/TestAsyncLifecycle/uniffi/async_lifecycle/AsyncContract.java b/tests/scripts/TestAsyncLifecycle/uniffi/async_lifecycle/AsyncContract.java index ddbff26..c2cc813 100644 --- a/tests/scripts/TestAsyncLifecycle/uniffi/async_lifecycle/AsyncContract.java +++ b/tests/scripts/TestAsyncLifecycle/uniffi/async_lifecycle/AsyncContract.java @@ -3,7 +3,6 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; @@ -18,9 +17,7 @@ // cancel on a handle after it has been freed. public final class AsyncContract { static final class Ledger { - final AtomicInteger polls = new AtomicInteger(); final AtomicInteger completes = new AtomicInteger(); - final AtomicInteger cancels = new AtomicInteger(); final AtomicInteger frees = new AtomicInteger(); volatile boolean freed; // Pipeline stage in progress: at most one thread, re-entrant on an inline executor. @@ -70,11 +67,14 @@ static CompletableFuture call(long handle, Executor executor) { (future, callback, continuation) -> { boolean live = enter(l, "poll", future); try { - l.polls.incrementAndGet(); // Native calls are serialized per handle only after the overlap has been // recorded, so a violation is reported instead of executed. synchronized (l) { - if (live && !l.freed) { + if (live && l.freed) { + violate("poll after free, won by free inside enter", future); + live = false; + } + if (live) { UniffiLib.ffi_uniffi_fixture_async_lifecycle_rust_future_poll_void(future, callback, continuation); } else { // Keeps the pipeline draining so the run ends and reports. @@ -93,7 +93,11 @@ static CompletableFuture call(long handle, Executor executor) { live = false; } synchronized (l) { - if (live && !l.freed) { + if (live && l.freed) { + violate("complete after free, won by free inside enter", future); + live = false; + } + if (live) { UniffiLib.ffi_uniffi_fixture_async_lifecycle_rust_future_complete_void(future, status); } } @@ -102,7 +106,6 @@ static CompletableFuture call(long handle, Executor executor) { } }, (future) -> { - l.cancels.incrementAndGet(); synchronized (l) { if (l.freed) { violate("cancel after free", future); @@ -142,7 +145,7 @@ static void storm(String name, Executor executor, int threads, int iters, Schedu ledgers.clear(); violations.clear(); List workers = new ArrayList<>(); - List> futures = new java.util.concurrent.CopyOnWriteArrayList<>(); + ConcurrentLinkedQueue> futures = new ConcurrentLinkedQueue<>(); for (int t = 0; t < threads; t++) { final int seed = t; Thread w = new Thread(() -> { diff --git a/tests/tests.rs b/tests/tests.rs index 7730a12..10b7b15 100644 --- a/tests/tests.rs +++ b/tests/tests.rs @@ -247,13 +247,13 @@ fn run_test_with_library_override( Ok(()) } -/// `java` with allocator debug checks that abort on a stale write into freed native memory. Each -/// platform ignores the other's variables. +/// `java` with allocator poisoning, so a use-after-free of native memory reads a fill pattern +/// instead of plausible data. Each platform ignores the other's variables. fn java_command() -> Command { let mut cmd = Command::new("java"); - // glibc: fill freed memory with a pattern. + // glibc: fill allocated and freed memory with 0xa5 and its complement. cmd.env("GLIBC_TUNABLES", "glibc.malloc.perturb=165"); - // macOS libmalloc: fill freed memory and guard large allocations. + // macOS libmalloc: fill freed memory, and abort on access past a guarded large allocation. cmd.env("MallocScribble", "1"); cmd.env("MallocGuardEdges", "1"); cmd