diff --git a/.github/workflows/publish-java.yml b/.github/workflows/publish-java.yml index 5307c735..1061971c 100644 --- a/.github/workflows/publish-java.yml +++ b/.github/workflows/publish-java.yml @@ -129,7 +129,9 @@ jobs: uses: actions/setup-java@v5 with: distribution: temurin - java-version: "21" + # 22+ so the jar task compiles the FFM overlay into META-INF/versions/22 + # (see build.gradle.kts ffmCapable); a 17/21 build ships JNI-only. + java-version: "25" - name: Build dashboard SPA uses: ./.github/actions/dashboard-build diff --git a/crates/taskito-java/src/ffi_c.rs b/crates/taskito-java/src/ffi_c.rs new file mode 100644 index 00000000..ac2463bf --- /dev/null +++ b/crates/taskito-java/src/ffi_c.rs @@ -0,0 +1,270 @@ +//! C-ABI fast path for the hot byte ops, parallel to the jni-rs surface. +//! +//! These `extern "C"` exports let the Java SDK call `enqueue`, `enqueueMany`, and +//! `getResult` over Project Panama (FFM) on JDK 22+, avoiding the JNI string and +//! `byte[]` marshalling. They are perf-only: the JNI entry points in [`crate::queue`] +//! remain the control plane and the fallback on JDKs without FFM. Both surfaces +//! share one cdylib and operate on the same opaque `QueueHandle` pointer. +//! +//! ## Contract +//! Each function returns an `i32` status — [`OK`], [`ERR`], or (results only) +//! [`ABSENT`] — and, when it produces bytes, writes a freshly heap-allocated +//! buffer's pointer and length through the `out_data`/`out_len` out-params. The +//! caller copies the bytes and must release the buffer with [`taskito_ffi_free`]. +//! On [`ERR`] the buffer holds a UTF-8 error message. Inputs are borrowed for the +//! duration of the call and never freed here. Every body is panic-guarded so a +//! Rust panic can never unwind across the FFI boundary. + +use std::panic::{catch_unwind, AssertUnwindSafe}; +use std::slice; + +use taskito_core::Storage; + +use crate::backend::QueueHandle; +use crate::convert::{build_new_job, EnqueueOptions}; + +/// The call succeeded; `out` holds the result bytes (job id, ids frame, or result). +pub const OK: i32 = 0; +/// The call failed; `out` holds a UTF-8 error message. +pub const ERR: i32 = 1; +/// `getResult` only: the job has no result yet; `out` is null/empty. +pub const ABSENT: i32 = 2; + +/// Borrow `len` bytes at `ptr` as an owned `Vec`; a null or empty input is empty. +/// +/// # Safety +/// `ptr` must be valid for `len` bytes for the duration of the call, or null. +unsafe fn read_bytes(ptr: *const u8, len: usize) -> Vec { + if ptr.is_null() || len == 0 { + Vec::new() + } else { + slice::from_raw_parts(ptr, len).to_vec() + } +} + +/// Borrow `len` bytes at `ptr` as a UTF-8 `String`. +/// +/// # Safety +/// See [`read_bytes`]. +unsafe fn read_string(ptr: *const u8, len: usize) -> Result { + String::from_utf8(read_bytes(ptr, len)).map_err(|e| format!("invalid UTF-8 argument: {e}")) +} + +/// Hand `bytes` to the caller: write its pointer and length to the out-params and +/// leak the allocation, to be reclaimed by [`taskito_ffi_free`]. +/// +/// # Safety +/// `out_data` and `out_len` must be valid, writable pointers. +unsafe fn emit(bytes: Vec, out_data: *mut *mut u8, out_len: *mut usize) { + let mut boxed = bytes.into_boxed_slice(); + *out_len = boxed.len(); + *out_data = boxed.as_mut_ptr(); + std::mem::forget(boxed); +} + +/// Resolve a producer result into a status code, emitting bytes or the error text. +/// +/// # Safety +/// `out_data`/`out_len` must be valid, writable pointers. +unsafe fn finish( + result: std::thread::Result, String>>, + out_data: *mut *mut u8, + out_len: *mut usize, +) -> i32 { + match result { + Ok(Ok(bytes)) => { + emit(bytes, out_data, out_len); + OK + } + Ok(Err(message)) => { + emit(message.into_bytes(), out_data, out_len); + ERR + } + Err(_) => { + emit(b"taskito native panic".to_vec(), out_data, out_len); + ERR + } + } +} + +/// Length-prefixed frame: `[count u32][len u32][bytes]...`, all little-endian. +/// Matches the Java side's framing so batch payloads and ids cross as one buffer. +fn frame(items: &[Vec]) -> Vec { + let total = 4 + items.iter().map(|item| 4 + item.len()).sum::(); + let mut out = Vec::with_capacity(total); + out.extend_from_slice(&(items.len() as u32).to_le_bytes()); + for item in items { + out.extend_from_slice(&(item.len() as u32).to_le_bytes()); + out.extend_from_slice(item); + } + out +} + +/// Decode a [`frame`]d buffer back into its elements. +fn unframe(buf: &[u8]) -> Result>, String> { + let mut pos = 0usize; + let take_u32 = |buf: &[u8], pos: &mut usize| -> Result { + let end = pos.checked_add(4).ok_or("frame length overflow")?; + let slice = buf.get(*pos..end).ok_or("truncated frame header")?; + *pos = end; + Ok(u32::from_le_bytes(slice.try_into().unwrap()) as usize) + }; + let count = take_u32(buf, &mut pos)?; + let mut items = Vec::with_capacity(count); + for _ in 0..count { + let len = take_u32(buf, &mut pos)?; + let end = pos.checked_add(len).ok_or("frame element overflow")?; + let element = buf.get(pos..end).ok_or("truncated frame element")?; + items.push(element.to_vec()); + pos = end; + } + Ok(items) +} + +/// `enqueue` — mirrors `NativeQueue.enqueue`. On success `out` is the job id. +/// +/// # Safety +/// `handle` must be a live `QueueHandle` pointer; the input pointers must be valid +/// for their lengths; `out_data`/`out_len` must be writable. +#[no_mangle] +pub unsafe extern "C" fn taskito_ffi_enqueue( + handle: i64, + task_ptr: *const u8, + task_len: usize, + payload_ptr: *const u8, + payload_len: usize, + options_ptr: *const u8, + options_len: usize, + out_data: *mut *mut u8, + out_len: *mut usize, +) -> i32 { + let work = AssertUnwindSafe(|| { + let queue = handle_ref(handle); + let task = read_string(task_ptr, task_len)?; + let payload = read_bytes(payload_ptr, payload_len); + let raw_options = read_string(options_ptr, options_len)?; + let options: EnqueueOptions = parse_options(&raw_options, "enqueue options")?; + let unique = options.unique_key.is_some(); + let new_job = build_new_job(task, payload, options, queue.namespace.as_deref()); + let job = if unique { + queue.storage.enqueue_unique(new_job) + } else { + queue.storage.enqueue(new_job) + } + .map_err(|e| e.to_string())?; + Ok::, String>(job.id.into_bytes()) + }); + finish(catch_unwind(work), out_data, out_len) +} + +/// `enqueueMany` — mirrors `NativeQueue.enqueueMany`. `payloads` is a [`frame`]d +/// buffer; `options` is a JSON array. On success `out` is a frame of job ids. +/// +/// # Safety +/// See [`taskito_ffi_enqueue`]. +#[no_mangle] +pub unsafe extern "C" fn taskito_ffi_enqueue_many( + handle: i64, + task_ptr: *const u8, + task_len: usize, + payloads_ptr: *const u8, + payloads_len: usize, + options_ptr: *const u8, + options_len: usize, + out_data: *mut *mut u8, + out_len: *mut usize, +) -> i32 { + let work = AssertUnwindSafe(|| { + let queue = handle_ref(handle); + let task = read_string(task_ptr, task_len)?; + let payloads = unframe(&read_bytes(payloads_ptr, payloads_len))?; + let raw_options = read_string(options_ptr, options_len)?; + let option_list: Vec = + parse_options(&raw_options, "enqueue batch options")?; + if option_list.len() != payloads.len() { + return Err("payloads and options must have the same length".to_string()); + } + let namespace = queue.namespace.as_deref(); + let new_jobs = payloads + .into_iter() + .zip(option_list) + .map(|(payload, options)| build_new_job(task.clone(), payload, options, namespace)) + .collect(); + let created = queue + .storage + .enqueue_batch(new_jobs) + .map_err(|e| e.to_string())?; + let ids: Vec> = created.into_iter().map(|job| job.id.into_bytes()).collect(); + Ok::, String>(frame(&ids)) + }); + finish(catch_unwind(work), out_data, out_len) +} + +/// `getResult` — mirrors `NativeQueue.getResult`. Returns [`ABSENT`] (with a null +/// `out`) when the job has no result, [`OK`] with the result bytes, or [`ERR`]. +/// +/// # Safety +/// See [`taskito_ffi_enqueue`]. +#[no_mangle] +pub unsafe extern "C" fn taskito_ffi_get_result( + handle: i64, + job_ptr: *const u8, + job_len: usize, + out_data: *mut *mut u8, + out_len: *mut usize, +) -> i32 { + let work = AssertUnwindSafe(|| { + let queue = handle_ref(handle); + let id = read_string(job_ptr, job_len)?; + let result = queue + .storage + .get_job(&id) + .map_err(|e| e.to_string())? + .and_then(|job| job.result); + Ok::>, String>(result) + }); + match catch_unwind(work) { + Ok(Ok(Some(bytes))) => { + emit(bytes, out_data, out_len); + OK + } + Ok(Ok(None)) => { + *out_data = std::ptr::null_mut(); + *out_len = 0; + ABSENT + } + Ok(Err(message)) => { + emit(message.into_bytes(), out_data, out_len); + ERR + } + Err(_) => { + emit(b"taskito native panic".to_vec(), out_data, out_len); + ERR + } + } +} + +/// Reclaim a buffer previously handed out by one of the producer functions. +/// +/// # Safety +/// `ptr`/`len` must be a pair produced by [`emit`] (i.e. returned through an +/// `out_data`/`out_len` pair), and freed at most once. +#[no_mangle] +pub unsafe extern "C" fn taskito_ffi_free(ptr: *mut u8, len: usize) { + if !ptr.is_null() && len != 0 { + drop(Box::from_raw(std::ptr::slice_from_raw_parts_mut(ptr, len))); + } +} + +/// Borrow the queue behind a handle. +/// +/// # Safety +/// `handle` must be a live `QueueHandle` pointer (see [`crate::handle`]). +unsafe fn handle_ref<'a>(handle: i64) -> &'a QueueHandle { + crate::handle::borrow::(handle) +} + +/// Parse JSON options without the JNI-flavoured `BindingError` (no env to throw on). +fn parse_options<'a, T: serde::Deserialize<'a>>(raw: &'a str, what: &str) -> Result { + serde_json::from_str(raw).map_err(|e| format!("invalid {what} JSON: {e}")) +} diff --git a/crates/taskito-java/src/lib.rs b/crates/taskito-java/src/lib.rs index 4d0fda19..759044e2 100644 --- a/crates/taskito-java/src/lib.rs +++ b/crates/taskito-java/src/lib.rs @@ -13,6 +13,7 @@ mod convert; mod dispatcher; mod error; mod ffi; +mod ffi_c; mod handle; mod jvm; mod queue; diff --git a/sdks/java/README.md b/sdks/java/README.md index fbfdd01c..f03be585 100644 --- a/sdks/java/README.md +++ b/sdks/java/README.md @@ -379,3 +379,10 @@ development against a freshly built library, point the loader at it directly: ```bash -Dtaskito.native.lib=/abs/path/to/target/release/libtaskito_java.so ``` + +On JDK 22+ the hot byte ops use a Project Panama (FFM) fast path; older JDKs (or +a jar built without the overlay) transparently fall back to JNI. FFM calls a +restricted native method, so a future JDK will deny it by default. The jar's +`Enable-Native-Access: ALL-UNNAMED` manifest attribute only covers `java -jar`; +apps that use the SDK as a classpath dependency should launch with +`--enable-native-access=ALL-UNNAMED` to grant access and silence the warning. diff --git a/sdks/java/build.gradle.kts b/sdks/java/build.gradle.kts index 9428d60a..96562437 100644 --- a/sdks/java/build.gradle.kts +++ b/sdks/java/build.gradle.kts @@ -136,6 +136,49 @@ tasks.named("processResources") { dependsOn(copyNative) } // dependency lazily — sourcesJar is registered after this script evaluates. tasks.withType().configureEach { dependsOn(copyNative) } +// --- FFM fast-path overlay (Multi-Release JAR) ---------------------------- +// Base classes target 17 (JNI transport + the fallback). On a build JDK >= 22 we +// also compile the Project Panama (FFM) transport at --release 22 and package it +// under META-INF/versions/22; the runtime selects it on JDK 22+ (see +// NativeTransport.create), else stays on JNI. Older build JDKs simply omit the +// overlay — same public API, faster impl where available (not feature divergence). +val ffmCapable = JavaVersion.current() >= JavaVersion.VERSION_22 + +if (ffmCapable) { + val java22 by sourceSets.creating { + java.srcDir("src/main/java22") + compileClasspath += sourceSets["main"].output + runtimeClasspath += sourceSets["main"].output + } + + tasks.named("compileJava22Java") { + options.release.set(22) + } + + tasks.named("jar") { + manifest { + attributes( + "Multi-Release" to "true", + // Only takes effect when the jar is run directly (java -jar); it does + // NOT cover consumers that depend on the SDK on their classpath — they + // must pass --enable-native-access=ALL-UNNAMED themselves (see README). + // Restricted FFM methods only warn today but a future JDK denies them. + "Enable-Native-Access" to "ALL-UNNAMED", + ) + } + into("META-INF/versions/22") { from(java22.output) } + } + + // Exercise the FFM transport in the test suite on this JDK 22+ build. Set on + // the test task directly: mutating the source set's runtimeClasspath here is + // too late (the java plugin has already captured the test task's classpath). + tasks.named("test") { + classpath += java22.output + // Silence (and forward-proof against) the restricted-native-access warning. + jvmArgs("--enable-native-access=ALL-UNNAMED") + } +} + tasks.test { useJUnitPlatform() } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java index d97ffacc..3491fb33 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java @@ -8,10 +8,15 @@ /** JNI-backed {@link QueueBackend} over a native queue handle. */ public final class JniQueueBackend implements QueueBackend { private final long handle; + + /** Hot byte ops (enqueue/enqueueMany/getResult) route through this; FFM when available, else JNI. */ + private final NativeTransport transport; + private volatile boolean closed; private JniQueueBackend(long handle) { this.handle = handle; + this.transport = NativeTransport.create(handle); } /** Open a native backend from its JSON options. */ @@ -21,12 +26,12 @@ public static JniQueueBackend open(String optionsJson) { @Override public String enqueue(String taskName, byte[] payload, String optionsJson) { - return NativeQueue.enqueue(handle, taskName, payload, optionsJson); + return transport.enqueue(taskName, payload, optionsJson); } @Override public String[] enqueueMany(String taskName, byte[][] payloads, String optionsJson) { - return NativeQueue.enqueueMany(handle, taskName, payloads, optionsJson); + return transport.enqueueMany(taskName, payloads, optionsJson); } @Override @@ -36,7 +41,7 @@ public Optional getJobJson(String jobId) { @Override public Optional getResult(String jobId) { - return Optional.ofNullable(NativeQueue.getResult(handle, jobId)); + return Optional.ofNullable(transport.getResult(jobId)); } @Override diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniTransport.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniTransport.java new file mode 100644 index 00000000..5585c724 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniTransport.java @@ -0,0 +1,25 @@ +package org.byteveda.taskito.internal; + +/** The default {@link NativeTransport}: the existing JNI entry points. */ +final class JniTransport implements NativeTransport { + private final long handle; + + JniTransport(long handle) { + this.handle = handle; + } + + @Override + public String enqueue(String taskName, byte[] payload, String optionsJson) { + return NativeQueue.enqueue(handle, taskName, payload, optionsJson); + } + + @Override + public String[] enqueueMany(String taskName, byte[][] payloads, String optionsJson) { + return NativeQueue.enqueueMany(handle, taskName, payloads, optionsJson); + } + + @Override + public byte[] getResult(String jobId) { + return NativeQueue.getResult(handle, jobId); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeTransport.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeTransport.java new file mode 100644 index 00000000..caa878a5 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeTransport.java @@ -0,0 +1,35 @@ +package org.byteveda.taskito.internal; + +/** + * The hot byte-transfer ops ({@code enqueue}, {@code enqueueMany}, + * {@code getResult}) behind a swappable transport. The default transport is JNI + * ({@link JniTransport}); on a JDK that supports Project Panama (FFM) the faster + * {@code FfmTransport} overlay is selected instead. Everything else on + * {@link NativeQueue} stays on JNI — this seam covers only the per-job hot path. + */ +public interface NativeTransport { + + /** Enqueue one job; returns its id. */ + String enqueue(String taskName, byte[] payload, String optionsJson); + + /** Enqueue a batch in one call; returns ids in input order. */ + String[] enqueueMany(String taskName, byte[][] payloads, String optionsJson); + + /** The job's serialized result, or {@code null} if absent/incomplete. */ + byte[] getResult(String jobId); + + /** + * The best transport for {@code handle}: the FFM fast path when its overlay + * class resolves (JDK 22+ via the multi-release jar) and its native symbols + * link, otherwise JNI. Any failure to initialize FFM falls back to JNI, so + * the seam never breaks the 17 floor. + */ + static NativeTransport create(long handle) { + try { + Class ffm = Class.forName("org.byteveda.taskito.internal.FfmTransport"); + return (NativeTransport) ffm.getMethod("create", long.class).invoke(null, handle); + } catch (ReflectiveOperationException | LinkageError ffmUnavailable) { + return new JniTransport(handle); + } + } +} diff --git a/sdks/java/src/main/java22/org/byteveda/taskito/internal/FfmTransport.java b/sdks/java/src/main/java22/org/byteveda/taskito/internal/FfmTransport.java new file mode 100644 index 00000000..4073b86a --- /dev/null +++ b/sdks/java/src/main/java22/org/byteveda/taskito/internal/FfmTransport.java @@ -0,0 +1,246 @@ +package org.byteveda.taskito.internal; + +import java.lang.foreign.Arena; +import java.lang.foreign.FunctionDescriptor; +import java.lang.foreign.Linker; +import java.lang.foreign.MemorySegment; +import java.lang.foreign.SymbolLookup; +import java.lang.foreign.ValueLayout; +import java.lang.invoke.MethodHandle; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.charset.StandardCharsets; +import org.byteveda.taskito.TaskitoException; + +/** + * Project Panama (FFM) fast path for the hot byte ops, used instead of JNI on + * JDK 22+. Lives only in the {@code META-INF/versions/22} overlay of the jar, so + * it is loaded only where {@code java.lang.foreign} is stable; {@link + * NativeTransport#create} resolves it reflectively and falls back to JNI when it + * (or its native symbols) are absent. + * + *

It calls the {@code taskito_ffi_*} C-ABI exports in the same cdylib the JNI + * surface uses — the library is already in the process via {@link + * NativeLoader#load()}, so symbols resolve through {@link + * SymbolLookup#loaderLookup()}. Each call confines its native arguments to a + * short-lived {@link Arena}; results are copied out and freed via {@code + * taskito_ffi_free}. The calls perform blocking storage I/O, so they are ordinary + * (non-{@code critical}) downcalls — never hold the heap for the JVM. + */ +public final class FfmTransport implements NativeTransport { + + /** + * Little-endian framing matching the Rust side's {@code u32::to_le_bytes}. + * Unaligned because the length prefixes sit at arbitrary offsets after each + * variable-length payload in the packed frame. + */ + private static final ValueLayout.OfInt FRAME_INT = + ValueLayout.JAVA_INT_UNALIGNED.withOrder(ByteOrder.LITTLE_ENDIAN); + + private static final int STATUS_OK = 0; + private static final int STATUS_ERR = 1; + private static final int STATUS_ABSENT = 2; + + private static final Linker LINKER = Linker.nativeLinker(); + private static final MethodHandle ENQUEUE; + private static final MethodHandle ENQUEUE_MANY; + private static final MethodHandle GET_RESULT; + private static final MethodHandle FREE; + + static { + NativeLoader.load(); // ensure the cdylib is loaded (System.load) before lookup + SymbolLookup lookup = SymbolLookup.loaderLookup(); + FunctionDescriptor producer = FunctionDescriptor.of( + ValueLayout.JAVA_INT, // status + ValueLayout.JAVA_LONG, // handle + ValueLayout.ADDRESS, + ValueLayout.JAVA_LONG, // task ptr,len + ValueLayout.ADDRESS, + ValueLayout.JAVA_LONG, // payload/frame ptr,len + ValueLayout.ADDRESS, + ValueLayout.JAVA_LONG, // options ptr,len + ValueLayout.ADDRESS, + ValueLayout.ADDRESS); // out_data, out_len + ENQUEUE = downcall(lookup, "taskito_ffi_enqueue", producer); + ENQUEUE_MANY = downcall(lookup, "taskito_ffi_enqueue_many", producer); + GET_RESULT = downcall( + lookup, + "taskito_ffi_get_result", + FunctionDescriptor.of( + ValueLayout.JAVA_INT, + ValueLayout.JAVA_LONG, + ValueLayout.ADDRESS, + ValueLayout.JAVA_LONG, // job-id ptr,len + ValueLayout.ADDRESS, + ValueLayout.ADDRESS)); // out_data, out_len + FREE = downcall( + lookup, "taskito_ffi_free", FunctionDescriptor.ofVoid(ValueLayout.ADDRESS, ValueLayout.JAVA_LONG)); + } + + private final long handle; + + private FfmTransport(long handle) { + this.handle = handle; + } + + /** Factory invoked reflectively by {@link NativeTransport#create} on JDK 22+. */ + public static NativeTransport create(long handle) { + return new FfmTransport(handle); + } + + @Override + public String enqueue(String taskName, byte[] payload, String optionsJson) { + try (Arena arena = Arena.ofConfined()) { + MemorySegment task = utf8(arena, taskName); + MemorySegment data = bytes(arena, payload); + MemorySegment options = utf8(arena, optionsJson); + MemorySegment outData = arena.allocate(ValueLayout.ADDRESS); + MemorySegment outLen = arena.allocate(ValueLayout.JAVA_LONG); + int status = (int) ENQUEUE.invokeExact( + handle, task, task.byteSize(), data, data.byteSize(), options, options.byteSize(), outData, outLen); + byte[] out = take(outData, outLen); + if (status == STATUS_OK) { + return new String(out, StandardCharsets.UTF_8); + } + throw error(out); + } catch (Throwable t) { + throw rethrow(t); + } + } + + @Override + public String[] enqueueMany(String taskName, byte[][] payloads, String optionsJson) { + try (Arena arena = Arena.ofConfined()) { + MemorySegment task = utf8(arena, taskName); + MemorySegment framed = frame(arena, payloads); + MemorySegment options = utf8(arena, optionsJson); + MemorySegment outData = arena.allocate(ValueLayout.ADDRESS); + MemorySegment outLen = arena.allocate(ValueLayout.JAVA_LONG); + int status = (int) ENQUEUE_MANY.invokeExact( + handle, + task, + task.byteSize(), + framed, + framed.byteSize(), + options, + options.byteSize(), + outData, + outLen); + byte[] out = take(outData, outLen); + if (status == STATUS_OK) { + return unframe(out); + } + throw error(out); + } catch (Throwable t) { + throw rethrow(t); + } + } + + @Override + public byte[] getResult(String jobId) { + try (Arena arena = Arena.ofConfined()) { + MemorySegment id = utf8(arena, jobId); + MemorySegment outData = arena.allocate(ValueLayout.ADDRESS); + MemorySegment outLen = arena.allocate(ValueLayout.JAVA_LONG); + int status = (int) GET_RESULT.invokeExact(handle, id, id.byteSize(), outData, outLen); + if (status == STATUS_ABSENT) { + return null; + } + byte[] out = take(outData, outLen); + if (status == STATUS_OK) { + return out; + } + throw error(out); + } catch (Throwable t) { + throw rethrow(t); + } + } + + private static MethodHandle downcall(SymbolLookup lookup, String symbol, FunctionDescriptor descriptor) { + MemorySegment address = + lookup.find(symbol).orElseThrow(() -> new UnsatisfiedLinkError("missing C-ABI symbol " + symbol)); + return LINKER.downcallHandle(address, descriptor); + } + + /** Allocate {@code bytes} off-heap for the call; an empty array maps to a null pointer. */ + private static MemorySegment bytes(Arena arena, byte[] value) { + if (value == null || value.length == 0) { + return MemorySegment.NULL; + } + MemorySegment segment = arena.allocate(value.length); + MemorySegment.copy(value, 0, segment, ValueLayout.JAVA_BYTE, 0, value.length); + return segment; + } + + private static MemorySegment utf8(Arena arena, String value) { + return bytes(arena, value.getBytes(StandardCharsets.UTF_8)); + } + + /** Encode {@code [count][len][bytes]...} (LE) for the batch payloads. */ + private static MemorySegment frame(Arena arena, byte[][] items) { + long size = Integer.BYTES; + for (byte[] item : items) { + size += Integer.BYTES + item.length; + } + MemorySegment segment = arena.allocate(size); + long offset = 0; + segment.set(FRAME_INT, offset, items.length); + offset += Integer.BYTES; + for (byte[] item : items) { + segment.set(FRAME_INT, offset, item.length); + offset += Integer.BYTES; + MemorySegment.copy(item, 0, segment, ValueLayout.JAVA_BYTE, offset, item.length); + offset += item.length; + } + return segment; + } + + /** Decode a {@code [count][len][bytes]...} (LE) frame of job ids. */ + private static String[] unframe(byte[] buffer) { + ByteBuffer reader = ByteBuffer.wrap(buffer).order(ByteOrder.LITTLE_ENDIAN); + String[] ids = new String[reader.getInt()]; + for (int i = 0; i < ids.length; i++) { + byte[] id = new byte[reader.getInt()]; + reader.get(id); + ids[i] = new String(id, StandardCharsets.UTF_8); + } + return ids; + } + + /** Copy out the bytes the native side allocated, then free them. */ + private static byte[] take(MemorySegment outData, MemorySegment outLen) { + MemorySegment pointer = outData.get(ValueLayout.ADDRESS, 0); + long length = outLen.get(ValueLayout.JAVA_LONG, 0); + if (pointer.address() == 0 || length == 0) { + return new byte[0]; + } + // Free in finally so a failed copy never leaks the Rust allocation. + try { + return pointer.reinterpret(length).toArray(ValueLayout.JAVA_BYTE); + } finally { + free(pointer, length); + } + } + + private static void free(MemorySegment pointer, long length) { + try { + FREE.invokeExact(pointer, length); + } catch (Throwable t) { + throw rethrow(t); + } + } + + private static TaskitoException error(byte[] message) { + return new TaskitoException(new String(message, StandardCharsets.UTF_8)); + } + + private static RuntimeException rethrow(Throwable t) { + if (t instanceof RuntimeException runtime) { + return runtime; + } + if (t instanceof Error error) { + throw error; + } + return new TaskitoException("FFM transport failure: " + t.getMessage(), t); + } +} diff --git a/sdks/java/src/test/java/org/byteveda/taskito/FfmRoundTripTest.java b/sdks/java/src/test/java/org/byteveda/taskito/FfmRoundTripTest.java new file mode 100644 index 00000000..884c6967 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/FfmRoundTripTest.java @@ -0,0 +1,62 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.List; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +/** + * End-to-end payload fidelity through the active transport (FFM on JDK 22+, else + * JNI): tricky payloads must survive enqueue → store → worker → getResult + * unchanged, including empty, full-Unicode, and large payloads that stress the + * byte transfer. Also covers the batch (enqueueMany) path. + */ +class FfmRoundTripTest { + + private static final Task ECHO = Task.of("ffm.echo", String.class); + + @Test + @Timeout(60) + void payloadsRoundTrip(@TempDir Path dir) throws Exception { + String unicode = "héllo 🌍 \t\n quotes \" backslash \\ end"; + String large = "x".repeat(256 * 1024); + List payloads = List.of("", unicode, large); + + try (Taskito queue = + Taskito.builder().url(dir.resolve("ffm.db").toString()).open()) { + try (Worker worker = queue.worker().handle(ECHO, p -> p).start()) { + for (String payload : payloads) { + String id = queue.enqueue(ECHO, payload); + queue.awaitJob(id, Duration.ofSeconds(30)); + assertEquals(payload, queue.getResult(id, String.class).orElseThrow()); + } + } + } + } + + @Test + @Timeout(60) + void batchRoundTrips(@TempDir Path dir) throws Exception { + List payloads = List.of("alpha", "", "gamma 🌍"); + + try (Taskito queue = + Taskito.builder().url(dir.resolve("ffm-batch.db").toString()).open()) { + try (Worker worker = queue.worker().handle(ECHO, p -> p).start()) { + List ids = queue.enqueueMany(ECHO, payloads); + assertEquals(payloads.size(), ids.size()); + for (int i = 0; i < ids.size(); i++) { + queue.awaitJob(ids.get(i), Duration.ofSeconds(30)); + assertEquals( + payloads.get(i), + queue.getResult(ids.get(i), String.class).orElseThrow()); + } + } + } + } +} diff --git a/sdks/java/src/test/java/org/byteveda/taskito/internal/TransportParityTest.java b/sdks/java/src/test/java/org/byteveda/taskito/internal/TransportParityTest.java new file mode 100644 index 00000000..642b8e2b --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/internal/TransportParityTest.java @@ -0,0 +1,85 @@ +package org.byteveda.taskito.internal; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assumptions.assumeFalse; + +import java.nio.file.Path; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * Asserts the FFM transport is selected on JDK 22+ and agrees with JNI on the hot + * ops. Worker-free: it compares the enqueue/getResult marshalling directly over + * one shared native handle (present-result fidelity is covered end-to-end by + * {@code FfmRoundTripTest}). On JDKs without the FFM overlay the FFM-specific + * assertions are skipped. + */ +class TransportParityTest { + + @Test + void ffmSelectedOnJava22() { + NativeTransport transport = NativeTransport.create(0L); + assumeFalse(transport instanceof JniTransport, "FFM transport not active on this JDK"); + assertEquals("FfmTransport", transport.getClass().getSimpleName()); + } + + @Test + void jniAndFfmAgreeOnHotOps(@TempDir Path dir) { + long handle = NativeQueue.open(sqliteOptions(dir.resolve("parity.db"))); + try { + NativeTransport ffm = NativeTransport.create(handle); + assumeFalse(ffm instanceof JniTransport, "FFM transport not active on this JDK"); + NativeTransport jni = new JniTransport(handle); + + byte[] payload = binaryPayload(); + + // enqueue: both marshal task/payload/options and return a valid id. + String jniId = jni.enqueue("parity.task", payload, "{}"); + String ffmId = ffm.enqueue("parity.task", payload, "{}"); + assertValidId(jniId); + assertValidId(ffmId); + + // getResult: a freshly enqueued job has none — both report absent (null). + assertNull(jni.getResult(jniId)); + assertNull(ffm.getResult(ffmId)); + assertNull(jni.getResult("does-not-exist")); + assertNull(ffm.getResult("does-not-exist")); + + // enqueueMany: identical batch shape through both transports. + byte[][] batch = {new byte[0], payload, "third".getBytes()}; + String options = "[{},{},{}]"; + String[] jniIds = jni.enqueueMany("parity.batch", batch, options); + String[] ffmIds = ffm.enqueueMany("parity.batch", batch, options); + assertEquals(batch.length, jniIds.length); + assertEquals(jniIds.length, ffmIds.length); + for (int i = 0; i < ffmIds.length; i++) { + assertValidId(jniIds[i]); + assertValidId(ffmIds[i]); + } + } finally { + NativeQueue.close(handle); + } + } + + private static void assertValidId(String id) { + assertNotNull(id); + assertFalse(id.isBlank()); + } + + /** A payload spanning every byte value, to catch any sign/encoding mishandling. */ + private static byte[] binaryPayload() { + byte[] bytes = new byte[256]; + for (int i = 0; i < bytes.length; i++) { + bytes[i] = (byte) i; + } + return bytes; + } + + private static String sqliteOptions(Path db) { + String dsn = db.toString().replace("\\", "\\\\"); + return "{\"backend\":\"sqlite\",\"dsn\":\"" + dsn + "\"}"; + } +}