diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c403c4d..0257cd6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -61,6 +61,9 @@ jobs: distribution: 'temurin' java-version: '25' + - name: Set up Kotlin + run: python tests/kotlin/install_kotlin.py --github-path + - name: Cache Cargo dependencies uses: Swatinem/rust-cache@v2 diff --git a/.github/workflows/update-nightly.yml b/.github/workflows/update-nightly.yml index 2f74df7..d1b699c 100644 --- a/.github/workflows/update-nightly.yml +++ b/.github/workflows/update-nightly.yml @@ -95,6 +95,10 @@ jobs: distribution: temurin java-version: '25' + - name: Set up Kotlin + if: steps.latest.outputs.changed == 'true' + run: python tests/kotlin/install_kotlin.py --github-path + - name: Cache Cargo dependencies if: steps.latest.outputs.changed == 'true' uses: Swatinem/rust-cache@v2 diff --git a/Readme.md b/Readme.md index 298776e..3ac18aa 100644 --- a/Readme.md +++ b/Readme.md @@ -334,6 +334,7 @@ The following example programs live in `tests/`, are compiled with the standard |---|---| | **[Lambda Callbacks](tests/integration/lambda_callbacks/Main.java)** | Passing native Java lambdas directly into Rust functions expecting `Fn` closures. | | **[Trait Implementors](tests/integration/trait_implementors/Main.java)** | Implementing a Rust trait on a Java class and passing it to Rust dynamic dispatch (`&dyn Trait`). | +| **[Kotlin Async](tests/kotlin/async_interop/Main.kt)** | Awaiting Rust `async` functions from Kotlin `suspend` code. | ## Features & Standard Library Support @@ -355,7 +356,7 @@ Beyond `core` and `alloc`, a large amount of the `std` is supported too, though | Subsystem | Status | Details | |---|---|---| | **Threads & Sync** | **Supported** | Thread spawning, scoped threads, Mutex, RwLock, Condvar, TLS | -| **Async & Futures** | **Supported** | Async functions, blocks, closures and trait methods; boxed/recursive `dyn Future`; cancellation | +| **Async & Futures** | **Supported** | Async functions, blocks, closures and trait methods; boxed/recursive `dyn Future`; Kotlin `suspend` interop; cancellation | | **Panic Unwinding** | **Supported** | Complete unwinding stack, `catch_unwind`, panic hooks, and abort-on-double-panic semantics | | **Stdio & Env** | **Supported** | `println!`, `eprintln!`, `stdin`, `env::args`, `env::vars` | | **Time & Random** | **Supported** | `SystemTime`, `Instant`, standard entropy seeds | @@ -497,6 +498,8 @@ boolean equal = Root.eq(root, new Leaf.A(42)); - **Git** - **cargo-jvm** - see install instructions in [its README](cargo-jvm/README.md) +The Kotlin compiler is only needed for `tests/kotlin`; CI installs the pinned version with `tests/kotlin/install_kotlin.py`. + ## Usage [`cargo-jvm`](cargo-jvm/README.md) is used to make building and running Rust projects on the JVM as seamless as possible. It wraps the standard Cargo workflow, forwarding all ordinary Cargo selection and feature arguments. For instructions on installing `cargo-jvm`, see [its README](cargo-jvm/README.md). diff --git a/Tester.py b/Tester.py index 08a2317..c25173b 100644 --- a/Tester.py +++ b/Tester.py @@ -5,12 +5,15 @@ import json import os import re +import shutil import subprocess import sys from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from test_harness import ( + ROOT, + TEST_TARGET_DIR, TestCase, build_test, cargo_jobs, @@ -197,6 +200,82 @@ def run_java( logs.append(f"|---- ❌ Invalid Java process test input: {error}") return False + if test.kind == "kotlin": + kotlin_files = sorted(test.directory.glob("*.kt")) + kotlin_files.extend(sorted((ROOT / "runtime" / "kotlin").glob("*.kt"))) + if not kotlin_files: + logs.append("|---- ❌ No .kt files found for Kotlin interop test") + return False + + executable_name = "kotlinc.bat" if os.name == "nt" else "kotlinc" + configured = os.environ.get("KOTLINC") + kotlinc = ( + Path(configured) + if configured + else Path(shutil.which("kotlinc") or "") + ) + if not kotlinc.is_file(): + local = ( + ROOT + / "target" + / "tools" + / "kotlin-2.4.10" + / "kotlinc" + / "bin" + / executable_name + ) + kotlinc = local if local.is_file() else Path() + if not kotlinc.is_file(): + logs.append( + "|---- ❌ kotlinc is required for tests/kotlin; install Kotlin or set KOTLINC" + ) + return False + + profile = "release" if release else "debug" + kotlin_jar = TEST_TARGET_DIR / "kotlin" / profile / f"{test.name}.jar" + kotlin_jar.parent.mkdir(parents=True, exist_ok=True) + command = [ + str(kotlinc), + "-jvm-target", + "1.8", + "-classpath", + str(jar), + "-include-runtime", + "-d", + str(kotlin_jar), + *(str(path) for path in kotlin_files), + ] + if os.name == "nt" and kotlinc.suffix.lower() == ".bat": + command = [os.environ.get("COMSPEC", "cmd.exe"), "/d", "/c", *command] + + logs.append("|--- 🟣 Compiling Kotlin interop test source...") + proc = run_command(command) + if proc.returncode != 0: + write_failure(test.directory / "kotlinc-fail.generated", proc) + logs.append(f"|---- ❌ kotlinc exited with code {proc.returncode}") + ci_diagnostic(logs, f"STDOUT:\n{proc.stdout}\nSTDERR:\n{proc.stderr}") + return False + + logs.append("|--- 🤖 Running Kotlin suspend entry point...") + classpath = os.pathsep.join([str(kotlin_jar), str(jar)]) + proc, diagnostics = run_java_command( + ["java", "-cp", classpath, "MainKt", *java_arguments], + timeout=java_timeout, + input_text=stdin, + ) + if diagnostics is not None: + output = ( + f"Kotlin process exceeded the {java_timeout:g}s timeout.\n\n" + f"{diagnostics}\n\nSTDOUT:\n{proc.stdout}\n\nSTDERR:\n{proc.stderr}" + ) + (test.directory / "java-timeout.generated").write_text( + output, encoding="utf-8" + ) + logs.append(f"|---- ❌ Kotlin process timed out after {java_timeout:g}s") + ci_diagnostic(logs, output) + return False + return check_results(proc, test, release, logs) + if test.kind != "integration": logs.append("|--- 🤖 Running with Java...") proc, diagnostics = run_java_command( diff --git a/runtime/kotlin/RustFuture.kt b/runtime/kotlin/RustFuture.kt new file mode 100644 index 0000000..895d070 --- /dev/null +++ b/runtime/kotlin/RustFuture.kt @@ -0,0 +1,110 @@ +package org.rustlang.runtime + +import kotlin.coroutines.Continuation +import kotlin.coroutines.resume +import kotlin.coroutines.resumeWithException +import kotlin.coroutines.suspendCoroutine + +private sealed class PollSignal + +private class Ready(val value: Any?) : PollSignal() + +private object Repoll : PollSignal() + +/** + * Arbitrates wakes which race with a poll. A wake observed during the poll is + * deferred until its result is known, so a future which wakes and returns + * Ready is never polled again after completion. + */ +private class PollGate( + private val continuation: Continuation, +) : Runnable { + private var polling = true + private var wakeRequested = false + private var resolved = false + + override fun run() { + val resume = synchronized(this) { + when { + resolved -> false + polling -> { + wakeRequested = true + false + } + else -> { + resolved = true + true + } + } + } + if (resume) { + continuation.resume(Repoll) + } + } + + fun finish(result: Any?) { + val signal = synchronized(this) { + polling = false + when { + resolved -> null + result !== KotlinFutureInterop.PENDING -> { + resolved = true + Ready(result) + } + wakeRequested -> { + resolved = true + Repoll + } + else -> null + } + } + if (signal != null) { + continuation.resume(signal) + } + } + + fun fail(failure: Throwable) { + val resume = synchronized(this) { + polling = false + if (resolved) { + false + } else { + resolved = true + true + } + } + if (resume) { + continuation.resumeWithException(failure) + } + } +} + +private suspend fun RustFuture.pollAfterWake(): PollSignal = + suspendCoroutine { continuation -> + val gate = PollGate(continuation) + try { + gate.finish(poll(gate)) + } catch (failure: Throwable) { + gate.fail(failure) + } + } + +/** + * Awaits a Rust `async` value. Every repoll is dispatched through the current + * Kotlin coroutine context; no global Java executor is imposed by the bridge. + */ +@Suppress("UNCHECKED_CAST") +suspend fun RustFuture.await(): T { + try { + while (true) { + when (val signal = pollAfterWake()) { + Repoll -> continue + is Ready -> return ( + if (signal.value === KotlinFutureInterop.UNIT) Unit else signal.value + ) as T + } + } + } finally { + KotlinFutureInterop.dropRustFuture(this) + } +} diff --git a/runtime/src/KotlinFutureInterop.java b/runtime/src/KotlinFutureInterop.java new file mode 100644 index 0000000..d1cdb2d --- /dev/null +++ b/runtime/src/KotlinFutureInterop.java @@ -0,0 +1,291 @@ +package org.rustlang.runtime; + +import java.lang.reflect.Constructor; +import java.lang.reflect.Field; +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.lang.reflect.Modifier; +import java.lang.reflect.Proxy; +import java.util.concurrent.ConcurrentHashMap; + +/** Runtime support for awaiting generated Rust futures from Kotlin. */ +public final class KotlinFutureInterop { + /** Returned by {@link RustFuture#poll(Runnable)} while the future is pending. */ + public static final Object PENDING = new Object(); + + /** Represents Rust's zero-sized unit result without depending on kotlin.Unit. */ + public static final Object UNIT = new Object(); + + private static final ConcurrentHashMap, PollAdapter> POLL_ADAPTERS = + new ConcurrentHashMap, PollAdapter>(); + + private KotlinFutureInterop() {} + + /** Called by generated RawWaker functions. */ + public static void wake(Pointer data) { + Object value = data.getObject(); + if (!(value instanceof Runnable)) { + throw new IllegalStateException("Rust async waker does not contain a Runnable"); + } + ((Runnable) value).run(); + } + + /** Invoked by the small method generated on every Rust coroutine class. */ + public static Object pollRustFuture( + Object future, + Runnable wake, + int futureSize, + String futureCodec, + int futureAlignment, + int wakerVtableSize, + String wakerVtableCodec, + int wakerVtableAlignment) { + Class futureClass = future.getClass(); + PollAdapter adapter = POLL_ADAPTERS.get(futureClass); + if (adapter == null) { + PollAdapter created = new PollAdapter( + futureClass, + futureSize, + futureCodec, + futureAlignment, + wakerVtableSize, + wakerVtableCodec, + wakerVtableAlignment); + PollAdapter existing = POLL_ADAPTERS.putIfAbsent(futureClass, created); + adapter = existing == null ? created : existing; + } + return adapter.poll(future, wake); + } + + /** Drops a completed or abandoned Rust async state machine exactly once. */ + public static void dropRustFuture(RustFuture future) { + Pointer.dropRustValue(future); + } + + private static void throwUnchecked(Throwable failure) { + KotlinFutureInterop.throwAny(failure); + } + + @SuppressWarnings("unchecked") + private static void throwAny(Throwable failure) throws T { + throw (T) failure; + } + + /** Cached reflection needed to enter a monomorphized Rust coroutine. */ + private static final class PollAdapter { + private final Constructor rawWakerConstructor; + private final Constructor wakerConstructor; + private final Constructor contextConstructor; + private final Constructor pinConstructor; + private final Method resume; + private final Pointer vtablePointer; + private final int futureSize; + private final String futureCodec; + private final int futureAlignment; + + private PollAdapter( + Class futureClass, + int futureSize, + String futureCodec, + int futureAlignment, + int wakerVtableSize, + String wakerVtableCodec, + int wakerVtableAlignment) { + try { + ClassLoader loader = futureClass.getClassLoader(); + Class rawWakerClass = Class.forName( + "org.rustlang.core.task.wake.RawWaker", true, loader); + Class rawWakerVTableClass = Class.forName( + "org.rustlang.core.task.wake.RawWakerVTable", true, loader); + Class wakerClass = Class.forName( + "org.rustlang.core.task.wake.Waker", true, loader); + Class contextClass = Class.forName( + "org.rustlang.core.task.wake.Context", true, loader); + Class pinClass = Class.forName( + "org.rustlang.core.pin.Pin_MutRef" + futureClass.getSimpleName(), + true, + loader); + + rawWakerConstructor = rawWakerClass.getConstructor(Pointer.class, Pointer.class); + Constructor vtableConstructor = constructorWithArity(rawWakerVTableClass, 4); + Class[] vtableParameters = vtableConstructor.getParameterTypes(); + final Pointer[] vtableHolder = new Pointer[1]; + + Object cloneFunction = Proxy.newProxyInstance( + loader, + new Class[] {vtableParameters[0]}, + new InvocationHandler() { + @Override + public Object invoke(Object proxy, Method method, Object[] args) + throws Throwable { + Object objectMethod = handleObjectMethod(proxy, method, args); + if (objectMethod != NOT_AN_OBJECT_METHOD) { + return objectMethod; + } + return rawWakerConstructor.newInstance(args[0], vtableHolder[0]); + } + }); + InvocationHandler wakeHandler = + new InvocationHandler() { + @Override + public Object invoke(Object proxy, Method method, Object[] args) { + Object objectMethod = handleObjectMethod(proxy, method, args); + if (objectMethod != NOT_AN_OBJECT_METHOD) { + return objectMethod; + } + KotlinFutureInterop.wake((Pointer) args[0]); + return null; + } + }; + Object wakeFunction = Proxy.newProxyInstance( + loader, new Class[] {vtableParameters[1]}, wakeHandler); + Object wakeByRefFunction = Proxy.newProxyInstance( + loader, new Class[] {vtableParameters[2]}, wakeHandler); + Object noopFunction = Proxy.newProxyInstance( + loader, + new Class[] {vtableParameters[3]}, + new InvocationHandler() { + @Override + public Object invoke(Object proxy, Method method, Object[] args) { + Object objectMethod = handleObjectMethod(proxy, method, args); + return objectMethod == NOT_AN_OBJECT_METHOD ? null : objectMethod; + } + }); + Object vtable = vtableConstructor.newInstance( + cloneFunction, wakeFunction, wakeByRefFunction, noopFunction); + vtablePointer = Pointer.cellAligned( + vtable, + wakerVtableSize, + wakerVtableCodec, + wakerVtableAlignment); + vtableHolder[0] = vtablePointer; + + wakerConstructor = wakerClass.getConstructor(rawWakerClass); + contextConstructor = rustConstructor(contextClass); + pinConstructor = pinClass.getConstructor(Pointer.class); + resume = findResumeMethod(futureClass, pinClass); + this.futureSize = futureSize; + this.futureCodec = futureCodec; + this.futureAlignment = futureAlignment; + } catch (ReflectiveOperationException error) { + throw new IllegalStateException( + "Could not prepare Rust async adapter for " + futureClass.getName(), error); + } + } + + private Object poll(Object future, Runnable wake) { + try { + Pointer wakePointer = Pointer.cell(wake); + Object rawWaker = rawWakerConstructor.newInstance(wakePointer, vtablePointer); + Object waker = wakerConstructor.newInstance(rawWaker); + Pointer wakerPointer = Pointer.cell(waker); + Class[] contextParameters = contextConstructor.getParameterTypes(); + Object[] contextArguments = new Object[contextParameters.length]; + contextArguments[0] = wakerPointer; + contextArguments[1] = wakerPointer; + for (int index = 2; index < contextParameters.length; index++) { + contextArguments[index] = defaultRustValue(contextParameters[index]); + } + Object context = contextConstructor.newInstance(contextArguments); + Pointer futurePointer = Pointer.receiverCellAligned( + future, futureSize, futureCodec, futureAlignment); + Object pin = pinConstructor.newInstance(futurePointer); + Object result = resume.invoke(null, pin, Pointer.cell(context)); + + if (result.getClass().getName().endsWith("$Pending")) { + return PENDING; + } + try { + Field output = result.getClass().getField("field0"); + return output.get(result); + } catch (NoSuchFieldException noOutput) { + return UNIT; + } + } catch (InvocationTargetException error) { + throwUnchecked(PanicSupport.detachForForeignBoundary(error.getCause())); + return null; + } catch (ReflectiveOperationException error) { + throw new IllegalStateException("Could not poll Rust async value", error); + } + } + + private static Constructor constructorWithArity(Class type, int arity) + throws NoSuchMethodException { + for (Constructor constructor : type.getConstructors()) { + if (constructor.getParameterTypes().length == arity) { + return constructor; + } + } + throw new NoSuchMethodException(type.getName() + " constructor with arity " + arity); + } + + private static Constructor rustConstructor(Class type) throws NoSuchMethodException { + for (Constructor constructor : type.getConstructors()) { + Class[] parameters = constructor.getParameterTypes(); + if (parameters.length >= 2 + && parameters[0] == Pointer.class + && parameters[1] == Pointer.class + && (parameters.length == 2 || parameters[2] != long.class)) { + return constructor; + } + } + throw new NoSuchMethodException(type.getName() + " Rust-value constructor"); + } + + private static Object defaultRustValue(Class type) throws ReflectiveOperationException { + if (type.isInterface()) { + Class none = Class.forName( + type.getName() + "$None", true, type.getClassLoader()); + return none.getConstructor().newInstance(); + } + try { + return type.getConstructor().newInstance(); + } catch (NoSuchMethodException noDefaultConstructor) { + Constructor constructor = constructorWithArity(type, 1); + return constructor.newInstance(defaultRustValue(constructor.getParameterTypes()[0])); + } + } + + private static Method findResumeMethod(Class futureClass, Class pinClass) + throws ReflectiveOperationException { + String packageName = futureClass.getPackage().getName(); + String crateName = packageName.contains(".") + ? packageName.substring(0, packageName.indexOf('.')) + : packageName; + Class moduleClass = Class.forName( + crateName + "." + crateName, true, futureClass.getClassLoader()); + for (Method method : moduleClass.getMethods()) { + Class[] parameters = method.getParameterTypes(); + if (Modifier.isStatic(method.getModifiers()) + && parameters.length == 2 + && parameters[0] == pinClass + && parameters[1] == Pointer.class + && method.getReturnType().getName().contains(".task.poll.Poll_")) { + return method; + } + } + throw new NoSuchMethodException( + "Rust coroutine resume method for " + futureClass.getName()); + } + } + + private static final Object NOT_AN_OBJECT_METHOD = new Object(); + + private static Object handleObjectMethod(Object proxy, Method method, Object[] args) { + if (method.getDeclaringClass() != Object.class) { + return NOT_AN_OBJECT_METHOD; + } + String name = method.getName(); + if ("toString".equals(name)) { + return "Rust async RawWaker function"; + } + if ("hashCode".equals(name)) { + return System.identityHashCode(proxy); + } + if ("equals".equals(name)) { + return proxy == args[0]; + } + return NOT_AN_OBJECT_METHOD; + } +} diff --git a/runtime/src/PanicSupport.java b/runtime/src/PanicSupport.java index ae7455f..72f6984 100644 --- a/runtime/src/PanicSupport.java +++ b/runtime/src/PanicSupport.java @@ -66,6 +66,14 @@ public static boolean isRustPanic(Pointer caughtException) { return caughtThrowable(caughtException) instanceof RustPanic; } + /** Ends Rust's tracked unwind when a panic is deliberately exported as a JVM exception. */ + public static Throwable detachForForeignBoundary(Throwable failure) { + if (failure instanceof RustPanic && ACTIVE_PANIC.get() == failure) { + ACTIVE_PANIC.remove(); + } + return failure; + } + public static Pointer foreignFailureMessage(Pointer caughtException) { byte[] message = describe(caughtThrowable(caughtException)).getBytes(StandardCharsets.UTF_8); return Pointer.array(message, 0, 1); diff --git a/runtime/src/Pointer.java b/runtime/src/Pointer.java index a415722..45f7536 100644 --- a/runtime/src/Pointer.java +++ b/runtime/src/Pointer.java @@ -5255,6 +5255,25 @@ public static Pointer field( return field(owner, fieldName, checkedArrayLength(size), codecClassName); } + /** Returns a stable field pointer with the exact alignment of its Rust field type. */ + public static Pointer fieldAligned( + Object owner, + String fieldName, + long size, + String codecClassName, + long alignment) { + int checkedAlignment = Math.toIntExact(alignment); + if (checkedAlignment <= 0 + || (checkedAlignment & (checkedAlignment - 1)) != 0) { + throw new IllegalArgumentException("Rust field alignment must be a power of two"); + } + Pointer pointer = field(owner, fieldName, size, codecClassName); + if (pointer.allocation != null) { + recordAlignment(pointer.allocation, checkedAlignment); + } + return pointer; + } + /** * Projects a field through a replaceable root cell. The field follows the * cell's current aggregate, so field mutations are immediately visible diff --git a/runtime/src/RustFuture.java b/runtime/src/RustFuture.java new file mode 100644 index 0000000..4c3d842 --- /dev/null +++ b/runtime/src/RustFuture.java @@ -0,0 +1,10 @@ +package org.rustlang.runtime; + +/** A Rust async value that can be driven by a JVM coroutine adapter. */ +public interface RustFuture { + /** + * Polls the Rust future once. The returned value is either the completed + * result or one of the markers exposed by {@link KotlinFutureInterop}. + */ + Object poll(Runnable wake); +} diff --git a/src/async_interop.rs b/src/async_interop.rs new file mode 100644 index 0000000..bb467b8 --- /dev/null +++ b/src/async_interop.rs @@ -0,0 +1,126 @@ +//! JVM-facing adapters for Rust coroutine values. + +use crate::oomir::{ + self, BasicBlock, CodeBlock, DataTypeMethod, Function, Instruction, Operand, Signature, Type, +}; +use rustc_hash::FxHashMap as HashMap; + +const RUST_FUTURE_INTERFACE: &str = "org/rustlang/runtime/RustFuture"; +const KOTLIN_INTEROP_CLASS: &str = "org/rustlang/runtime/KotlinFutureInterop"; + +#[derive(Clone)] +pub struct PointerLayout { + pub size: i32, + pub alignment: i32, + pub codec: Operand, +} + +fn poll_function( + coroutine_class: &str, + future: PointerLayout, + waker_vtable: PointerLayout, +) -> Function { + let object_type = Type::Class("java/lang/Object".to_string()); + let runnable_type = Type::Class("java/lang/Runnable".to_string()); + let string_type = Type::java_string(); + let instructions = vec![ + Instruction::InvokeStatic { + dest: Some("result".to_string()), + class_name: KOTLIN_INTEROP_CLASS.to_string(), + method_name: "pollRustFuture".to_string(), + method_ty: Signature { + params: vec![ + ("future".to_string(), object_type.clone()), + ("wake".to_string(), runnable_type.clone()), + ("size".to_string(), Type::I32), + ("codec".to_string(), string_type), + ("alignment".to_string(), Type::I32), + ("waker_vtable_size".to_string(), Type::I32), + ("waker_vtable_codec".to_string(), Type::java_string()), + ("waker_vtable_alignment".to_string(), Type::I32), + ], + ret: Box::new(object_type.clone()), + is_static: true, + }, + args: vec![ + Operand::Variable { + name: "_1".to_string(), + ty: Type::Class(coroutine_class.to_string()), + }, + Operand::Variable { + name: "_2".to_string(), + ty: runnable_type.clone(), + }, + Operand::Constant(oomir::Constant::I32(future.size)), + future.codec, + Operand::Constant(oomir::Constant::I32(future.alignment)), + Operand::Constant(oomir::Constant::I32(waker_vtable.size)), + waker_vtable.codec, + Operand::Constant(oomir::Constant::I32(waker_vtable.alignment)), + ], + }, + Instruction::Return { + operand: Some(Operand::Variable { + name: "result".to_string(), + ty: object_type.clone(), + }), + }, + ]; + + Function { + name: "poll".to_string(), + owner_class: Some(coroutine_class.to_string()), + signature: Signature { + params: vec![ + ("self".to_string(), Type::Class(coroutine_class.to_string())), + ("wake".to_string(), runnable_type), + ], + ret: Box::new(object_type), + is_static: false, + }, + debug_variables: Vec::new(), + body: CodeBlock { + entry: "entry".to_string(), + basic_blocks: HashMap::from_iter([( + "entry".to_string(), + BasicBlock { + label: "entry".to_string(), + instructions, + }, + )]), + }, + } +} + +/// Makes one generated Rust coroutine implement the small runtime future ABI. +/// +/// Layout information is embedded directly in the generated bridge. Runtime +/// adapters therefore never need to discover pointer codecs by scanning JARs. +pub fn add_rust_future_bridge( + data_types: &mut HashMap, + class_name: &str, + future: PointerLayout, + waker_vtable: PointerLayout, +) { + let Some(oomir::DataType::Class { + methods, + interfaces, + .. + }) = data_types.get_mut(class_name) + else { + return; + }; + if methods.contains_key("poll") { + return; + } + if !interfaces + .iter() + .any(|interface| interface == RUST_FUTURE_INTERFACE) + { + interfaces.push(RUST_FUTURE_INTERFACE.to_string()); + } + methods.insert( + "poll".to_string(), + DataTypeMethod::Function(poll_function(class_name, future, waker_vtable)), + ); +} diff --git a/src/lib.rs b/src/lib.rs index 5687ae3..c219988 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -85,6 +85,7 @@ use std::{ sync::{Arc, Mutex, mpsc}, }; +mod async_interop; mod lower1; mod lower2; mod metrics; @@ -330,18 +331,8 @@ impl CanonicalDataTypeRegistry { { schema = variants .iter() - .find_map(|variant| match variant { - oomir::DataType::Interface { - methods, - interfaces, - is_enum, - } => Some(oomir::DataType::Interface { - methods: methods.clone(), - interfaces: interfaces.clone(), - is_enum: *is_enum, - }), - _ => None, - }) + .find(|variant| matches!(variant, oomir::DataType::Interface { .. })) + .map(Self::schema_for) .expect("an interface variant was observed"); } schemas.insert(name.clone(), schema); @@ -393,6 +384,54 @@ impl CanonicalDataTypeRegistry { } } +#[cfg(test)] +mod canonical_data_type_registry_tests { + use super::*; + + #[test] + fn mixed_interface_schema_keeps_only_method_stubs() { + let interface_method = oomir::DataTypeMethod::AdtHelperMethod { + kind: oomir::AdtHelperKind::StaticPartialEqEnum { + enum_class: "example/Mixed".to_string(), + variants: Vec::new(), + }, + }; + let variants = HashMap::from_iter([( + "example/Mixed".to_string(), + vec![ + oomir::DataType::Class { + is_abstract: false, + super_class: None, + fields: Vec::new(), + methods: HashMap::default(), + interfaces: Vec::new(), + }, + oomir::DataType::Interface { + methods: HashMap::from_iter([("eq".to_string(), interface_method)]), + interfaces: Vec::new(), + is_enum: true, + }, + ], + )]); + + let schemas = CanonicalDataTypeRegistry::shared_schemas(&variants); + let oomir::DataType::Interface { + methods, is_enum, .. + } = &schemas["example/Mixed"] + else { + panic!("the interface schema must take precedence"); + }; + assert!(*is_enum); + assert_eq!( + methods.get("eq"), + Some(&oomir::DataTypeMethod::SimpleConstantReturn( + oomir::Type::Void, + None, + )) + ); + } +} + fn combine_class_bundles(path: &Path, bundles: &[(String, PathBuf)]) -> std::io::Result<()> { let mut output = BufWriter::new(std::fs::File::create(path)?); output.write_all(CLASS_BUNDLE_MAGIC)?; @@ -1177,6 +1216,11 @@ fn synthetic_drop_callees_for_ty<'tcx>( (!pointee.has_escaping_bound_vars() && pointee.needs_drop(tcx, typing_env)) .then(|| Instance::resolve_drop_glue(tcx, pointee)) } + TyKind::Coroutine(..) + if !ty.has_escaping_bound_vars() && ty.needs_drop(tcx, typing_env) => + { + Some(Instance::resolve_drop_glue(tcx, ty)) + } _ => None, } }) @@ -1468,14 +1512,37 @@ fn lower_public_library_exports<'tcx>( return; } - let function_defs = java_public_surface_def_ids(tcx, JavaPublicSurface::Exported); - - let function_roots = function_defs + let function_defs = java_public_surface_def_ids(tcx, JavaPublicSurface::Exported) .into_iter() .filter(|def_id| is_lowerable_java_public_function(tcx, *def_id)) - .map(|def_id| Instance::mono(tcx, def_id)); + .collect::>(); + let mut function_roots = function_defs + .iter() + .copied() + .map(|def_id| Instance::mono(tcx, def_id)) + .collect::>(); + for function_def in &function_defs { + let Some(local_def) = function_def.as_local() else { + continue; + }; + for nested_def in tcx.nested_bodies_within(local_def) { + let coroutine_def = nested_def.to_def_id(); + if !tcx.coroutine_is_async(coroutine_def) { + continue; + } + let coroutine_ty = tcx + .type_of(coroutine_def) + .instantiate_identity() + .skip_norm_wip(); + let TyKind::Coroutine(_, args) = coroutine_ty.kind() else { + continue; + }; + function_roots.push(Instance::new_raw(coroutine_def, args)); + } + } // Rustc's collector owns ordinary Rust reachability. Java exports are - // additional roots, so only they need a supplemental MIR call walk. + // additional roots. Async state-machine bodies are also roots because a + // JVM caller polls them through RustFuture rather than an ordinary MIR call. lower_supplemental_instance_closure( tcx, function_roots, diff --git a/src/lower1.rs b/src/lower1.rs index 307f5b2..2016cfc 100644 --- a/src/lower1.rs +++ b/src/lower1.rs @@ -95,7 +95,7 @@ impl LocalBitSet { if let Some(last) = result.words.last_mut() && !local_count.is_multiple_of(u64::BITS as usize) { - *last = (1 << (local_count % u64::BITS as usize)) - 1; + *last = (1u64 << (local_count % u64::BITS as usize)) - 1; } result } diff --git a/src/lower1/control_flow.rs b/src/lower1/control_flow.rs index cef8995..956873a 100644 --- a/src/lower1/control_flow.rs +++ b/src/lower1/control_flow.rs @@ -1071,6 +1071,64 @@ pub(super) fn emit_rust_drop_value<'tcx>( dest: None, }); } + TyKind::Coroutine(..) => { + let pointer_ty = oomir::Type::Pointer(Box::new(oomir_ty.clone())); + let pointer_name = format!("{temp_prefix}_coroutine_pointer"); + let size = super::types::layout_size_bytes(tcx, rust_ty) + .unwrap_or_else(|error| panic!("could not size coroutine drop value: {error}")); + let alignment = super::types::layout_align_bytes(tcx, rust_ty) + .unwrap_or_else(|error| panic!("could not align coroutine drop value: {error}")); + let codec = + super::types::pointer_memory_codec_operand(rust_ty, tcx, data_types, instance); + instructions.push(oomir::Instruction::InvokeStatic { + dest: Some(pointer_name.clone()), + class_name: oomir::POINTER_CLASS.to_string(), + method_name: "receiverCellAligned".to_string(), + method_ty: oomir::Signature { + params: vec![ + ( + "value".to_string(), + oomir::Type::Class("java/lang/Object".to_string()), + ), + ("size".to_string(), oomir::Type::I32), + ("codec".to_string(), oomir::Type::java_string()), + ("alignment".to_string(), oomir::Type::I32), + ], + ret: Box::new(pointer_ty.clone()), + is_static: true, + }, + args: vec![ + value, + oomir::Operand::Constant(oomir::Constant::I32( + i32::try_from(size) + .expect("coroutine drop value exceeds the JVM address space"), + )), + codec, + oomir::Operand::Constant(oomir::Constant::I32( + i32::try_from(alignment) + .expect("coroutine drop alignment exceeds the JVM address space"), + )), + ], + }); + let drop_instance = Instance::resolve_drop_glue(tcx, rust_ty); + let target = super::naming::mono_fn_name_from_instance(tcx, drop_instance); + instructions.push(oomir::Instruction::InvokeStatic { + class_name: target + .class_to_call_on + .expect("coroutine drop glue has a JVM owner"), + method_name: target.method_name, + method_ty: oomir::Signature { + params: vec![("coroutine".to_string(), pointer_ty.clone())], + ret: Box::new(oomir::Type::Void), + is_static: true, + }, + args: vec![oomir::Operand::Variable { + name: pointer_name, + ty: pointer_ty, + }], + dest: None, + }); + } TyKind::Dynamic(..) => { instructions.push(oomir::Instruction::InvokeStatic { class_name: oomir::POINTER_CLASS.to_string(), diff --git a/src/lower1/control_flow/rvalue.rs b/src/lower1/control_flow/rvalue.rs index d77a648..8376154 100644 --- a/src/lower1/control_flow/rvalue.rs +++ b/src/lower1/control_flow/rvalue.rs @@ -58,6 +58,19 @@ fn rust_layout_size_operand<'tcx>( )) } +fn rust_layout_alignment<'tcx>( + ty: rustc_middle::ty::Ty<'tcx>, + tcx: TyCtxt<'tcx>, + instance: Instance<'tcx>, +) -> usize { + let ty = EarlyBinder::bind(tcx, ty) + .instantiate(tcx, instance.args) + .skip_norm_wip(); + super::super::types::layout_align_bytes(tcx, ty).unwrap_or_else(|error| { + panic!("could not determine pointer layout alignment for {ty:?}: {error}") + }) +} + fn pointer_view_size_operand<'tcx>( pointer_ty: rustc_middle::ty::Ty<'tcx>, tcx: TyCtxt<'tcx>, @@ -2005,6 +2018,7 @@ fn emit_pointer_to_place<'tcx>( instructions, ); if matches!(managed_base_rust_ty.kind(), TyKind::Tuple(_)) + || matches!(managed_base_rust_ty.kind(), TyKind::Closure(..)) || matches!(managed_base_rust_ty.kind(), TyKind::Adt(adt_def, _) if adt_def.is_struct()) { let base_layout = tcx @@ -2066,34 +2080,49 @@ fn emit_pointer_to_place<'tcx>( instructions, ); let result_name = format!("{temp_prefix}_dst_field_pointer"); + let field_alignment = rust_layout_alignment(field_rust_ty, tcx, instance); + let explicitly_aligned = field_alignment > 16; + let mut field_params = vec![ + ( + "owner".to_string(), + oomir::Type::Class("java/lang/Object".to_string()), + ), + ("field_name".to_string(), oomir::Type::java_string()), + ("size".to_string(), oomir::Type::U64), + ("codec".to_string(), oomir::Type::java_string()), + ]; + let mut field_args = vec![ + base_value, + oomir::Operand::Constant(oomir::Constant::String(field_name)), + rust_layout_size_operand(field_rust_ty, tcx, instance), + super::super::types::pointer_view_codec_operand( + field_rust_ty, + tcx, + data_types, + instance, + ), + ]; + if explicitly_aligned { + field_params.push(("alignment".to_string(), oomir::Type::U64)); + field_args.push(oomir::Operand::Constant(oomir::Constant::U64( + u64::try_from(field_alignment) + .expect("Rust layout alignment exceeds u64"), + ))); + } instructions.push(oomir::Instruction::InvokeStatic { dest: Some(result_name.clone()), class_name: oomir::POINTER_CLASS.to_string(), - method_name: "field".to_string(), + method_name: if explicitly_aligned { + "fieldAligned".to_string() + } else { + "field".to_string() + }, method_ty: oomir::Signature { - params: vec![ - ( - "owner".to_string(), - oomir::Type::Class("java/lang/Object".to_string()), - ), - ("field_name".to_string(), oomir::Type::java_string()), - ("size".to_string(), oomir::Type::U64), - ("codec".to_string(), oomir::Type::java_string()), - ], + params: field_params, ret: Box::new(pointer_ty.clone()), is_static: true, }, - args: vec![ - base_value, - oomir::Operand::Constant(oomir::Constant::String(field_name)), - rust_layout_size_operand(field_rust_ty, tcx, instance), - super::super::types::pointer_view_codec_operand( - field_rust_ty, - tcx, - data_types, - instance, - ), - ], + args: field_args, }); return oomir::Operand::Variable { name: result_name, diff --git a/src/lower1/naming.rs b/src/lower1/naming.rs index 2d3f2c1..b0ddcb8 100644 --- a/src/lower1/naming.rs +++ b/src/lower1/naming.rs @@ -4,7 +4,9 @@ use super::jvm_names; use rustc_hash::FxHashMap as HashMap; use rustc_hir::{attrs::lang_items::LangItem, def::DefKind}; use rustc_middle::middle::codegen_fn_attrs::CodegenFnAttrFlags; -use rustc_middle::ty::{GenericArg, Instance, InstanceKind, TyCtxt, TyKind, TypeVisitableExt}; +use rustc_middle::ty::{ + GenericArg, Instance, InstanceKind, ShimKind, TyCtxt, TyKind, TypeVisitableExt, +}; use rustc_span::sym; const MAX_MONO_FN_NAME_LEN: usize = 128; @@ -621,6 +623,22 @@ pub fn mono_fn_name_from_instance<'tcx>(tcx: TyCtxt<'tcx>, instance: Instance<'t }; } + if let InstanceKind::Shim(ShimKind::DropGlue(_, Some(drop_ty))) = instance.def + && let TyKind::Coroutine(def_id, args) = drop_ty.kind() + { + let base = jvm_names::method_for_function(tcx, instance.def_id()); + let coroutine = jvm_names::coroutine_class_for_args(tcx, *def_id, args, instance); + return FnNameData { + class_to_call_on: Some(mono_owner_class(tcx, instance)), + method_name: crate::stable_hash::readable_or_hashed_name( + &base, + &super::types::sanitize_name_token(&coroutine), + &super::types::stable_def_identity(tcx, instance.def_id()), + MAX_MONO_FN_NAME_LEN, + ), + }; + } + let class = Some(mono_owner_class(tcx, instance)); let external_runtime_generic = is_external_runtime_generic(tcx, instance); diff --git a/src/lower1/types.rs b/src/lower1/types.rs index 0d109e8..51c2edb 100644 --- a/src/lower1/types.rs +++ b/src/lower1/types.rs @@ -8,10 +8,10 @@ use rustc_hashes::Hash64; use rustc_middle::ty::layout::TyAndLayout; use rustc_middle::ty::print::{with_no_trimmed_paths, with_resolve_crate_name}; use rustc_middle::ty::{ - AdtDef, EarlyBinder, ExistentialPredicate, FloatTy, GenericArgsRef, IntTy, Region, Ty, TyCtxt, - TyKind, TypeFoldable, TypeFolder, TypeVisitableExt, TypingEnv, UintTy, + AdtDef, EarlyBinder, ExistentialPredicate, FloatTy, GenericArgs, GenericArgsRef, IntTy, Region, + Ty, TyCtxt, TyKind, TypeFoldable, TypeFolder, TypeVisitableExt, TypingEnv, UintTy, }; -use rustc_span::{def_id::DefId, sym}; +use rustc_span::{Symbol, def_id::DefId, sym}; use std::{ cell::RefCell, sync::{LazyLock, Mutex}, @@ -35,6 +35,26 @@ pub(crate) fn reset_type_lowering_cache() { ENUM_TYPES_IN_PROGRESS.with(|types| types.borrow_mut().clear()); } +fn raw_waker_vtable_ty<'tcx>(tcx: TyCtxt<'tcx>) -> Ty<'tcx> { + let waker_def_id = tcx + .get_diagnostic_item(Symbol::intern("Waker")) + .expect("core::task::Waker diagnostic item is unavailable"); + let waker_args = GenericArgs::identity_for_item(tcx, waker_def_id); + let raw_waker_ty = tcx.adt_def(waker_def_id).non_enum_variant().fields[FieldIdx::from_usize(0)] + .ty(tcx, waker_args) + .skip_norm_wip(); + let TyKind::Adt(raw_waker_def, raw_waker_args) = raw_waker_ty.kind() else { + panic!("core::task::Waker no longer contains a RawWaker"); + }; + let vtable_ref_ty = raw_waker_def.non_enum_variant().fields[FieldIdx::from_usize(1)] + .ty(tcx, raw_waker_args) + .skip_norm_wip(); + let TyKind::Ref(_, vtable_ty, _) = vtable_ref_ty.kind() else { + panic!("core::task::RawWaker no longer contains a RawWakerVTable reference"); + }; + *vtable_ty +} + pub const UNION_BYTES_FIELD: &str = "_bytes"; pub const UNION_OBJECTS_FIELD: &str = "_objects"; pub const MANAGED_OBJECT_POINTER_VIEW_CODEC: &str = "@managed-object"; @@ -7477,6 +7497,73 @@ fn ty_to_oomir_type_resolved<'tcx>( { *existing_fields = fields; } + let needs_managed_drop = resolved_ty.needs_drop(tcx, TypingEnv::fully_monomorphized()) + && matches!( + data_types.get(&safe_name), + Some(oomir::DataType::Class { methods, .. }) + if !methods.contains_key(MANAGED_DROP_METHOD) + ); + if needs_managed_drop { + if let Some(oomir::DataType::Class { + methods, + interfaces, + .. + }) = data_types.get_mut(&safe_name) + { + methods.insert( + MANAGED_DROP_METHOD.to_string(), + DataTypeMethod::SimpleConstantReturn(oomir::Type::Void, None), + ); + interfaces.push(MANAGED_DROP_INTERFACE.to_string()); + } + let drop_method = managed_drop_glue_function( + resolved_ty, + &safe_name, + tcx, + data_types, + instance_context, + ); + if let Some(oomir::DataType::Class { methods, .. }) = data_types.get_mut(&safe_name) + { + methods.insert( + MANAGED_DROP_METHOD.to_string(), + DataTypeMethod::Function(drop_method), + ); + } + } + if tcx.coroutine_is_async(*def_id) { + let future_size = layout_size_bytes(tcx, resolved_ty) + .unwrap_or_else(|error| panic!("could not size Rust async value: {error}")); + let future_alignment = layout_align_bytes(tcx, resolved_ty) + .unwrap_or_else(|error| panic!("could not align Rust async value: {error}")); + let future_codec = + pointer_memory_codec_operand(resolved_ty, tcx, data_types, instance_context); + let vtable_ty = raw_waker_vtable_ty(tcx); + let vtable_size = layout_size_bytes(tcx, vtable_ty) + .unwrap_or_else(|error| panic!("could not size RawWakerVTable: {error}")); + let vtable_alignment = layout_align_bytes(tcx, vtable_ty) + .unwrap_or_else(|error| panic!("could not align RawWakerVTable: {error}")); + let vtable_codec = + pointer_memory_codec_operand(vtable_ty, tcx, data_types, instance_context); + crate::async_interop::add_rust_future_bridge( + data_types, + &safe_name, + crate::async_interop::PointerLayout { + size: i32::try_from(future_size) + .expect("Rust async value exceeds the JVM address space"), + alignment: i32::try_from(future_alignment) + .expect("Rust async alignment exceeds the JVM address space"), + codec: future_codec, + }, + crate::async_interop::PointerLayout { + size: i32::try_from(vtable_size) + .expect("RawWakerVTable exceeds the JVM address space"), + alignment: i32::try_from(vtable_alignment) + .expect("RawWakerVTable alignment exceeds the JVM address space"), + codec: vtable_codec, + }, + ); + } oomir::Type::Class(safe_name) } rustc_middle::ty::TyKind::FnPtr(_, _) => { diff --git a/src/lower2.rs b/src/lower2.rs index 8cdec94..93baef7 100644 --- a/src/lower2.rs +++ b/src/lower2.rs @@ -1470,7 +1470,10 @@ pub fn oomir_to_jvm_bytecode( } else { function.signature.clone() }; - for (name, _) in &emitted_signature.params { + for (name, param_ty) in emitted_signature.explicit_jvm_params() { + if !param_ty.has_jvm_value() { + continue; + } let name_index = main_cp.add_utf8(name)?; parameters_for_attribute.push(jvm::attributes::MethodParameter { name_index, diff --git a/src/lower2/jvm_gen.rs b/src/lower2/jvm_gen.rs index 381ad38..339d7fc 100644 --- a/src/lower2/jvm_gen.rs +++ b/src/lower2/jvm_gen.rs @@ -1514,14 +1514,14 @@ fn append_field_equality_check( Type::Class(class_name) if has_generated_eq(module, class_name) => { let class_idx = cp.add_class(class_name)?; if matches!( - module.data_types.get(class_name), + module.data_type(class_name), Some(oomir::DataType::Interface { is_enum: true, .. }) ) { let eq_desc = format!("(L{class_name};L{class_name};)Z"); let eq_ref = cp.add_interface_method_ref(class_idx, "eq", &eq_desc)?; instructions.push(Instruction::Invokestatic(eq_ref)); } else if matches!( - module.data_types.get(class_name), + module.data_type(class_name), Some(oomir::DataType::Interface { .. }) ) { let eq_desc = format!("(L{class_name};)Z"); @@ -1537,7 +1537,7 @@ fn append_field_equality_check( Type::Interface(interface_name) if has_generated_eq(module, interface_name) => { let interface_idx = cp.add_class(interface_name)?; if matches!( - module.data_types.get(interface_name), + module.data_type(interface_name), Some(oomir::DataType::Interface { is_enum: true, .. }) ) { let eq_desc = format!("(L{interface_name};L{interface_name};)Z"); diff --git a/test_harness.py b/test_harness.py index a0a21f8..d76f48a 100644 --- a/test_harness.py +++ b/test_harness.py @@ -21,7 +21,7 @@ TEST_TARGET_DIR = ROOT / "target" / "test-suite" TEST_CONFIG = ROOT / "config.toml" CORE_BUILD_MANIFEST = ROOT / "tests" / "support" / "core_build" / "Cargo.toml" -TEST_TYPES = ("binary", "multicrate", "integration", "cargo_jvm") +TEST_TYPES = ("binary", "multicrate", "integration", "kotlin", "cargo_jvm") CACHE_TAG = ( "Signature: 8a477f597d28d172789f06886806bc55\n" "# This file is a cache directory tag created by rustc_codegen_jvm.\n" @@ -259,7 +259,7 @@ def build_test( command.append("--release") return run_command(command, env=stdlib_build_environment(env)) - if test.kind == "integration": + if test.kind in ("integration", "kotlin"): cargo_jvm = ROOT / "cargo-jvm" / "target" / "release" / ( "cargo-jvm.exe" if os.name == "nt" else "cargo-jvm" ) diff --git a/tests/integration/inner_classes/Main.java b/tests/integration/inner_classes/Main.java index 364670e..9ad8b7a 100644 --- a/tests/integration/inner_classes/Main.java +++ b/tests/integration/inner_classes/Main.java @@ -65,6 +65,20 @@ public static void main(String[] args) { throw new AssertionError("Nested struct payloads with different fields should not compare equal"); } + inner_classes.WrapperChoice remoteSeven = inner_classes.inner_classes.wrap_remote_choice( + inner_classes.inner_classes.make_remote_choice(7)); + inner_classes.WrapperChoice anotherRemoteSeven = inner_classes.inner_classes.wrap_remote_choice( + inner_classes.inner_classes.make_remote_choice(7)); + inner_classes.WrapperChoice remoteEight = inner_classes.inner_classes.wrap_remote_choice( + inner_classes.inner_classes.make_remote_choice(8)); + + if (!inner_classes.WrapperChoice.eq(remoteSeven, anotherRemoteSeven)) { + throw new AssertionError("Enum equality should use shared schemas across datatype shards"); + } + if (inner_classes.WrapperChoice.eq(remoteSeven, remoteEight)) { + throw new AssertionError("Cross-shard enum equality should compare payload fields"); + } + inner_classes.LeafEvent leaf = inner_classes.inner_classes.make_leaf_number(41); inner_classes.Event promoted = inner_classes.inner_classes.promote_leaf(leaf); if (promoted.getClass() != leaf.getClass()) { diff --git a/tests/integration/inner_classes/src/lib.rs b/tests/integration/inner_classes/src/lib.rs index 1fa707a..e5f0cb2 100644 --- a/tests/integration/inner_classes/src/lib.rs +++ b/tests/integration/inner_classes/src/lib.rs @@ -22,6 +22,19 @@ pub enum NestedEquality { Boxed(NumberBox), } +// These names intentionally land in different canonical datatype shards. +// Equality for the outer enum must still discover that its payload is an enum +// interface through the crate-wide shared schema. +pub enum RemoteChoice { + Value(i32), + Empty, +} + +pub enum WrapperChoice { + Remote(RemoteChoice), + Direct(i32), +} + pub enum LeafEvent { Number(i32), Empty, @@ -109,6 +122,14 @@ pub fn wrap_boxed(value: NumberBox) -> NestedEquality { NestedEquality::Boxed(value) } +pub fn make_remote_choice(value: i32) -> RemoteChoice { + RemoteChoice::Value(value) +} + +pub fn wrap_remote_choice(value: RemoteChoice) -> WrapperChoice { + WrapperChoice::Remote(value) +} + pub fn make_leaf_number(value: i32) -> LeafEvent { LeafEvent::Number(value) } diff --git a/tests/kotlin/async_interop/Cargo.lock b/tests/kotlin/async_interop/Cargo.lock new file mode 100644 index 0000000..6102b8c --- /dev/null +++ b/tests/kotlin/async_interop/Cargo.lock @@ -0,0 +1,7 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "async_interop" +version = "0.1.0" diff --git a/tests/kotlin/async_interop/Cargo.toml b/tests/kotlin/async_interop/Cargo.toml new file mode 100644 index 0000000..bd80763 --- /dev/null +++ b/tests/kotlin/async_interop/Cargo.toml @@ -0,0 +1,6 @@ +[package] +name = "async_interop" +version = "0.1.0" +edition = "2024" + +[dependencies] diff --git a/tests/kotlin/async_interop/Main.kt b/tests/kotlin/async_interop/Main.kt new file mode 100644 index 0000000..cddb246 --- /dev/null +++ b/tests/kotlin/async_interop/Main.kt @@ -0,0 +1,174 @@ +import async_interop.AsyncOutcome +import org.rustlang.runtime.await +import org.rustlang.runtime.KotlinFutureInterop +import org.rustlang.runtime.RustFuture +import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executor +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import kotlin.coroutines.AbstractCoroutineContextElement +import kotlin.coroutines.Continuation +import kotlin.coroutines.ContinuationInterceptor +import kotlin.coroutines.CoroutineContext +import kotlin.coroutines.startCoroutine + +private const val DISPATCHER_THREAD = "kotlin-rust-async-dispatcher" + +private class ExecutorDispatcher( + private val executor: Executor, +) : AbstractCoroutineContextElement(ContinuationInterceptor), ContinuationInterceptor { + val dispatchCount = AtomicInteger() + + override fun interceptContinuation(continuation: Continuation): Continuation = + object : Continuation { + override val context: CoroutineContext = continuation.context + + override fun resumeWith(result: Result) { + dispatchCount.incrementAndGet() + executor.execute { continuation.resumeWith(result) } + } + } +} + +private fun runSuspend( + coroutineContext: CoroutineContext, + block: suspend () -> T, +): T { + val finished = CountDownLatch(1) + var value: Any? = null + var failure: Throwable? = null + block.startCoroutine( + object : Continuation { + override val context: CoroutineContext = coroutineContext + + override fun resumeWith(result: Result) { + result.fold( + onSuccess = { value = it }, + onFailure = { failure = it }, + ) + finished.countDown() + } + }, + ) + check(finished.await(20, TimeUnit.SECONDS)) { "Kotlin async suite timed out" } + failure?.let { throw it } + @Suppress("UNCHECKED_CAST") + return value as T +} + +private fun assertDispatcherThread(stage: String) { + check(Thread.currentThread().name == DISPATCHER_THREAD) { + "$stage ran on ${Thread.currentThread().name}, not the Kotlin dispatcher" + } +} + +private class WakeAndReadyKotlinFuture : RustFuture { + var polls = 0 + + override fun poll(wake: Runnable): Any { + polls += 1 + check(polls == 1) { "completed Kotlin test future was repolled" } + wake.run() + wake.run() + return 99 + } +} + +private suspend fun exerciseAsyncInterop() { + assertDispatcherThread("suite entry") + + val immediate = async_interop.async_interop.immediate(41).await() + check(immediate == 42) + + val yielded = async_interop.async_interop.add_after_yield(19, 23).await() + check(yielded == 42) + + val chained = async_interop.async_interop.chained(40).await() + check(chained == 42) + + async_interop.async_interop.unit_after_yield().await() + + // Hundreds of synchronous, repeated wakes exercise coalescing without + // growing the Kotlin or Java call stack. + val stormPolls = async_interop.async_interop.synchronous_wake_storm(256).await() + check(stormPolls == 257) + + // A wake racing with Ready must not cause an illegal post-completion poll. + check(async_interop.async_interop.wake_and_ready().await() == 73) + val kotlinRace = WakeAndReadyKotlinFuture() + check(kotlinRace.await() == 99) + check(kotlinRace.polls == 1) + + // The wake originates on a Rust-created thread, but Kotlin must resume and + // repoll through its own ContinuationInterceptor. + val delayed = async_interop.async_interop.delayed_from_rust_thread(123).await() + check(delayed == 123) + assertDispatcherThread("delayed Rust wake") + + // Multiple Rust threads concurrently update and wake the same future. + val concurrent = async_interop.async_interop.concurrent_rust_wakers(6).await() + check(concurrent == 6) + assertDispatcherThread("concurrent Rust wakes") + + // The captured value has 64-byte Rust alignment and survives several + // suspension points, exercising the compiler-emitted future layout. + check(async_interop.async_interop.aligned_capture(32).await() == 42) + + // Rich Rust structs and enums travel through a deeply nested async flow. + val outcome = async_interop.async_interop.extensive_workflow(12, false) + .await() + check(outcome is AsyncOutcome.Completed) + val report = outcome.field0 + check(report.seed == 12) + check(report.stages == 4) + check(report.checksum == 81) + check(report.aligned) + + val rejected = async_interop.async_interop.extensive_workflow(-7, true) + .await() + check(rejected is AsyncOutcome.Rejected) + check(rejected.field0 == -7) + + // Completion and exceptional exit both drop captured Rust state once. + async_interop.async_interop.reset_drop_count() + check(async_interop.async_interop.tracked_completion(314).await() == 314) + check(async_interop.async_interop.drop_count() == 1L) + + async_interop.async_interop.silence_panics() + async_interop.async_interop.reset_drop_count() + val panic = try { + async_interop.async_interop.panic_after_yield().await() + null + } catch (failure: Throwable) { + failure + } + check(panic != null) { "Rust async panic should reach Kotlin as an exception" } + check(async_interop.async_interop.drop_count() == 1L) + + // Direct abandonment verifies that a pending state machine can be dropped + // without waiting for a wake that will never arrive. + async_interop.async_interop.reset_drop_count() + val abandoned = async_interop.async_interop.abandoned_future() + check(abandoned.poll(Runnable {}) === KotlinFutureInterop.PENDING) + KotlinFutureInterop.dropRustFuture(abandoned) + check(async_interop.async_interop.drop_count() == 1L) + + assertDispatcherThread("suite completion") +} + +fun main() { + val executor = Executors.newSingleThreadExecutor { task -> + Thread(task, DISPATCHER_THREAD).apply { isDaemon = true } + } + val dispatcher = ExecutorDispatcher(executor) + try { + runSuspend(dispatcher) { exerciseAsyncInterop() } + check(dispatcher.dispatchCount.get() >= 4) { + "Kotlin continuation interceptor was not used for asynchronous wakes" + } + } finally { + executor.shutdownNow() + } + println("Kotlin async interop passed") +} diff --git a/tests/kotlin/async_interop/java-output.expected b/tests/kotlin/async_interop/java-output.expected new file mode 100644 index 0000000..bb5f2f3 --- /dev/null +++ b/tests/kotlin/async_interop/java-output.expected @@ -0,0 +1,3 @@ +STDOUT: +Kotlin async interop passed +STDERR: diff --git a/tests/kotlin/async_interop/src/lib.rs b/tests/kotlin/async_interop/src/lib.rs new file mode 100644 index 0000000..d2ee0d9 --- /dev/null +++ b/tests/kotlin/async_interop/src/lib.rs @@ -0,0 +1,286 @@ +use core::future::Future; +use core::pin::Pin; +use core::task::{Context, Poll}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::Duration; + +static DROP_COUNT: AtomicUsize = AtomicUsize::new(0); + +struct YieldOnce(bool); + +impl Future for YieldOnce { + type Output = (); + + fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll { + if self.0 { + Poll::Ready(()) + } else { + self.0 = true; + context.waker().wake_by_ref(); + Poll::Pending + } + } +} + +struct WakeStorm { + remaining: u32, + polls: i32, +} + +impl Future for WakeStorm { + type Output = i32; + + fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll { + self.polls += 1; + if self.remaining == 0 { + Poll::Ready(self.polls) + } else { + self.remaining -= 1; + // Repeated wakes for one pending poll must collapse into one repoll. + context.waker().wake_by_ref(); + context.waker().wake_by_ref(); + context.waker().wake_by_ref(); + Poll::Pending + } + } +} + +struct WakeAndReady(bool); + +impl Future for WakeAndReady { + type Output = i32; + + fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll { + assert!(!self.0, "wake-and-ready future was polled after completion"); + self.0 = true; + context.waker().wake_by_ref(); + Poll::Ready(73) + } +} + +struct DelayedWake { + ready: Arc, + spawned: bool, + value: i32, +} + +impl Future for DelayedWake { + type Output = i32; + + fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll { + if self.ready.load(Ordering::Acquire) != 0 { + return Poll::Ready(self.value); + } + if !self.spawned { + self.spawned = true; + let ready = self.ready.clone(); + let waker = context.waker().clone(); + thread::Builder::new() + .name("rust-async-waker".to_string()) + .spawn(move || { + thread::sleep(Duration::from_millis(15)); + ready.store(1, Ordering::Release); + waker.wake_by_ref(); + }) + .expect("could not spawn delayed Rust waker"); + } + Poll::Pending + } +} + +struct ConcurrentWakeState { + completed: AtomicUsize, + current_waker: Mutex>, +} + +struct ConcurrentWakes { + state: Arc, + started: bool, + workers: usize, +} + +impl Future for ConcurrentWakes { + type Output = i32; + + fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll { + *self.state.current_waker.lock().unwrap() = Some(context.waker().clone()); + let completed = self.state.completed.load(Ordering::Acquire); + if completed == self.workers { + return Poll::Ready(i32::try_from(completed).unwrap()); + } + if !self.started { + self.started = true; + for worker in 0..self.workers { + let state = self.state.clone(); + thread::Builder::new() + .name(format!("rust-concurrent-waker-{worker}")) + .spawn(move || { + thread::sleep(Duration::from_millis(4 * (worker as u64 + 1))); + state.completed.fetch_add(1, Ordering::AcqRel); + if let Some(waker) = state.current_waker.lock().unwrap().as_ref() { + waker.wake_by_ref(); + } + }) + .expect("could not spawn concurrent Rust waker"); + } + } + Poll::Pending + } +} + +struct Never; + +impl Future for Never { + type Output = (); + + fn poll(self: Pin<&mut Self>, _context: &mut Context<'_>) -> Poll { + Poll::Pending + } +} + +struct DropProbe; + +impl Drop for DropProbe { + fn drop(&mut self) { + DROP_COUNT.fetch_add(1, Ordering::SeqCst); + } +} + +#[repr(align(64))] +struct AlignedCapture { + bytes: [u8; 65], + value: i32, +} + +pub struct AsyncReport { + pub seed: i32, + pub stages: i32, + pub checksum: i32, + pub aligned: bool, +} + +pub enum AsyncOutcome { + Completed(AsyncReport), + Rejected(i32), +} + +pub async fn immediate(value: i32) -> i32 { + value + 1 +} + +pub async fn add_after_yield(left: i32, right: i32) -> i32 { + YieldOnce(false).await; + left + right +} + +pub async fn chained(value: i32) -> i32 { + let first = add_after_yield(value, 1).await; + add_after_yield(first, 1).await +} + +pub async fn unit_after_yield() { + YieldOnce(false).await; +} + +pub async fn synchronous_wake_storm(rounds: u32) -> i32 { + WakeStorm { + remaining: rounds, + polls: 0, + } + .await +} + +pub async fn wake_and_ready() -> i32 { + WakeAndReady(false).await +} + +pub async fn delayed_from_rust_thread(value: i32) -> i32 { + DelayedWake { + ready: Arc::new(AtomicUsize::new(0)), + spawned: false, + value, + } + .await +} + +pub async fn concurrent_rust_wakers(workers: u32) -> i32 { + ConcurrentWakes { + state: Arc::new(ConcurrentWakeState { + completed: AtomicUsize::new(0), + current_waker: Mutex::new(None), + }), + started: false, + workers: usize::try_from(workers).unwrap(), + } + .await +} + +pub async fn aligned_capture(value: i32) -> i32 { + let mut capture = AlignedCapture { + bytes: [0; 65], + value, + }; + capture.bytes[0] = 3; + capture.bytes[64] = 7; + WakeStorm { + remaining: 3, + polls: 0, + } + .await; + let address = core::ptr::addr_of!(capture) as usize; + assert_eq!(address % core::mem::align_of::(), 0); + capture.value + i32::from(capture.bytes[0]) + i32::from(capture.bytes[64]) +} + +pub async fn extensive_workflow(seed: i32, reject: bool) -> AsyncOutcome { + YieldOnce(false).await; + if reject { + return AsyncOutcome::Rejected(seed); + } + + let wake_storm_polls = WakeStorm { + remaining: 5, + polls: 0, + } + .await; + let delayed = delayed_from_rust_thread(seed * 3 + 1).await; + let concurrent = concurrent_rust_wakers(4).await; + let aligned_value = aligned_capture(seed).await; + AsyncOutcome::Completed(AsyncReport { + seed, + stages: 4, + checksum: seed + wake_storm_polls + delayed + concurrent + aligned_value, + aligned: true, + }) +} + +pub fn reset_drop_count() { + DROP_COUNT.store(0, Ordering::SeqCst); +} + +pub fn drop_count() -> usize { + DROP_COUNT.load(Ordering::SeqCst) +} + +pub async fn tracked_completion(value: i32) -> i32 { + let _probe = DropProbe; + YieldOnce(false).await; + value +} + +pub async fn abandoned_future() { + let _probe = DropProbe; + Never.await; +} + +pub fn silence_panics() { + std::panic::set_hook(Box::new(|_| {})); +} + +pub async fn panic_after_yield() -> i32 { + let _probe = DropProbe; + YieldOnce(false).await; + panic!("intentional Kotlin async bridge panic") +} diff --git a/tests/kotlin/install_kotlin.py b/tests/kotlin/install_kotlin.py new file mode 100644 index 0000000..c6da567 --- /dev/null +++ b/tests/kotlin/install_kotlin.py @@ -0,0 +1,69 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import hashlib +import os +import shutil +import urllib.request +import zipfile +from pathlib import Path + +VERSION = "2.4.10" +SHA256 = "473dd66c7a3ef4b182065b3da670466c1bf2773a9dbb0ed8b33a39fe9d4f876d" +URL = ( + "https://github.com/JetBrains/kotlin/releases/download/" + f"v{VERSION}/kotlin-compiler-{VERSION}.zip" +) +ROOT = Path(__file__).resolve().parents[2] +INSTALL_ROOT = ROOT / "target" / "tools" / f"kotlin-{VERSION}" + + +def extract_bundle(bundle: zipfile.ZipFile, destination: Path) -> None: + for member in bundle.infolist(): + extracted = Path(bundle.extract(member, destination)) + if os.name == "nt" or not extracted.is_file(): + continue + executable_bits = (member.external_attr >> 16) & 0o111 + if executable_bits: + extracted.chmod(extracted.stat().st_mode | executable_bits) + + +def main() -> int: + parser = argparse.ArgumentParser(description="Install the Kotlin compiler used by tests") + parser.add_argument( + "--github-path", + action="store_true", + help="Append kotlinc's bin directory to the GitHub Actions PATH file", + ) + args = parser.parse_args() + + executable = "kotlinc.bat" if os.name == "nt" else "kotlinc" + bin_directory = INSTALL_ROOT / "kotlinc" / "bin" + if not (bin_directory / executable).is_file(): + archive = INSTALL_ROOT.with_suffix(".zip") + archive.parent.mkdir(parents=True, exist_ok=True) + print(f"Downloading Kotlin compiler {VERSION}...") + with urllib.request.urlopen(URL) as response, archive.open("wb") as output: + shutil.copyfileobj(response, output) + actual = hashlib.sha256(archive.read_bytes()).hexdigest() + if actual != SHA256: + archive.unlink(missing_ok=True) + raise SystemExit( + f"Kotlin compiler checksum mismatch: expected {SHA256}, found {actual}" + ) + INSTALL_ROOT.mkdir(parents=True, exist_ok=True) + with zipfile.ZipFile(archive) as bundle: + extract_bundle(bundle, INSTALL_ROOT) + archive.unlink() + + if args.github_path: + github_path_file = Path(os.environ["GITHUB_PATH"]) + with github_path_file.open("a", encoding="utf-8") as github_path: + github_path.write(f"{bin_directory}\n") + print(bin_directory) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())