various slop before making the point of the situation

Author: slinkydeveloper

sdk-core/build.gradle.kts

@@ -103,6 +103,24 @@ sourceSets["test"].compileClasspath += java23.output
 
 sourceSets["test"].runtimeClasspath += java23.output
 
+// ---------------------------------------------------------------------------
+// JMH benchmarks (manual wiring; run with `./gradlew :sdk-core:jmh`). Mirrors the `test` setup so
+// benchmarks drive the FFM (java23) state machine directly and reuse the test `ProtoUtils` helpers
+// (and the native lib, which ships in `main`'s resources). Add new benchmarks under src/jmh/java.
+// ---------------------------------------------------------------------------
+sourceSets.create("jmh") { java.setSrcDirs(listOf("src/jmh/java")) }
+
+sourceSets["jmh"].compileClasspath +=
+    sourceSets["main"].output + java23.output + sourceSets["test"].output
+
+sourceSets["jmh"].runtimeClasspath +=
+    sourceSets["main"].output + java23.output + sourceSets["test"].output
+
+// Give the benchmarks everything the tests have (ProtoUtils' deps: protobuf, sdk-common, …).
+configurations["jmhImplementation"].extendsFrom(configurations["testImplementation"])
+
+configurations["jmhRuntimeOnly"].extendsFrom(configurations["testRuntimeOnly"])
+
 // jextract bindings are generated into the java23 (FFM) source set only — they
 // reference java.lang.foreign, which is not available at the Java 17 base level.
 jextract.libraries {
@@ -224,6 +242,11 @@ dependencies {
   testImplementation(libs.vertx.junit5)
   testImplementation(libs.vertx.kotlin.coroutines)
   testRuntimeOnly(libs.junit.platform.launcher)
+
+  // JMH benchmarks (src/jmh). Manual wiring — no champeau plugin (keeps us off its Gradle-9
+  // compat).
+  "jmhImplementation"("org.openjdk.jmh:jmh-core:1.37")
+  "jmhAnnotationProcessor"("org.openjdk.jmh:jmh-generator-annprocess:1.37")
 }
 
 // ---------------------------------------------------------------------------
@@ -327,3 +350,22 @@ ksp {
       )
   arg("dev.restate.codegen.disabledClasses", disabledClassesCodegen.joinToString(","))
 }
+
+// ---------------------------------------------------------------------------
+// JMH run task. `./gradlew :sdk-core:jmh` runs all benchmarks; pass JMH CLI args via
+// `-PjmhArgs="SysCall -f 1 -wi 3 -i 5 -prof gc"`. Runs on the JDK 25 toolchain (FFM enabled).
+// ---------------------------------------------------------------------------
+val jmh by
+    tasks.registering(JavaExec::class) {
+      group = "benchmark"
+      description = "Run JMH benchmarks (FfmStateMachine / shared-core boundary)"
+      dependsOn("jmhClasses", copyNativeLib)
+      javaLauncher.set(javaToolchains.launcherFor { languageVersion = JavaLanguageVersion.of(25) })
+      mainClass.set("org.openjdk.jmh.Main")
+      classpath = sourceSets["jmh"].runtimeClasspath
+      // JDK 25 enables FFM by default; silence the native-access warning.
+      jvmArgs("--enable-native-access=ALL-UNNAMED")
+      (project.findProperty("jmhArgs") as String?)?.let {
+        if (it.isNotBlank()) args(it.trim().split(" "))
+      }
+    }

sdk-core/src/jmh/java/dev/restate/sdk/core/benchmarks/SysCallBenchmark.java

@@ -0,0 +1,100 @@
+// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH
+//
+// This file is part of the Restate Java SDK,
+// which is released under the MIT license.
+//
+// You can find a copy of the license in file LICENSE in the root
+// directory of this repository or package, or at
+// https://github.com/restatedev/sdk-java/blob/main/LICENSE
+package dev.restate.sdk.core.benchmarks;
+
+import dev.restate.common.Slice;
+import dev.restate.common.Target;
+import dev.restate.sdk.core.statemachine.JavaStateMachine;
+import dev.restate.sdk.core.statemachine.ProtoUtils;
+import dev.restate.sdk.core.statemachine.StateMachine;
+import dev.restate.sdk.core.statemachine.ffm.FfmStateMachine;
+import dev.restate.sdk.endpoint.HeadersAccessor;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.*;
+import org.openjdk.jmh.infra.Blackhole;
+
+/**
+ * Benchmarks the cost of a single {@code sys_call} — argument marshalling (the FFM {@code
+ * CallArguments} build), the native downcall, the Rust-side decode + {@code VM::sys_call}, and the
+ * result decode. Parameterized over both state machines ({@code ffmStateMachine}) so the FFM
+ * boundary can be read against the pure-Java baseline, and over {@code payloadSize} to see how the
+ * "allocate the payload in Rust (ownership transfer) vs. copy" cost scales with payload size.
+ *
+ * <p>{@code sys_call} is stateful — each call appends a command + notification handles to the VM's
+ * journal — so we prime a <b>fresh</b> VM per measured op ({@link Level#Invocation}, not measured)
+ * and do a single {@code sys_call} (draining its buffered output via {@code takeOutput}) in the
+ * measured body; the VM is discarded each op, bounding journal growth.
+ *
+ * <p>Run with {@code ./gradlew :sdk-core:jmh -PjmhArgs="SysCall"} (JDK 25 toolchain, FFM active).
+ */
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@Warmup(iterations = 5, time = 1)
+@Measurement(iterations = 5, time = 1)
+@Fork(1)
+@State(Scope.Thread)
+public class SysCallBenchmark {
+
+  // Immutable inputs, computed once.
+  private Slice startMessage;
+  private Slice inputMessage;
+  private Target target;
+  private Slice payload;
+  private HeadersAccessor headersAccessor;
+
+  // The primed VM, recreated per measured op.
+  private StateMachine sm;
+
+  @Param({"true", "false"})
+  boolean ffmStateMachine;
+
+  // Payload size in bytes — exercises how much the copy-vs-transfer cost scales with payload size.
+  @Param({"16", "1024", "65536"})
+  int payloadSize;
+
+  @Setup(Level.Trial)
+  public void prepareInputs() {
+    startMessage = ProtoUtils.encodeMessageToSlice(ProtoUtils.startMessage(1).build());
+    inputMessage = ProtoUtils.encodeMessageToSlice(ProtoUtils.inputCmd("benchmark-input"));
+    target = Target.service("BenchmarkService", "benchmarkHandler");
+    // Random payload of the parameterized length; fixed seed so runs are reproducible.
+    byte[] payloadBytes = new byte[payloadSize];
+    new Random().nextBytes(payloadBytes);
+    payload = Slice.wrap(payloadBytes);
+    headersAccessor =
+        HeadersAccessor.wrap(
+            Map.of("content-type", ProtoUtils.serviceProtocolContentTypeHeader(true)));
+  }
+
+  @Setup(Level.Invocation)
+  public void primeVm() {
+    sm =
+        ffmStateMachine
+            ? new FfmStateMachine(headersAccessor)
+            : new JavaStateMachine(headersAccessor);
+    sm.notifyInput(startMessage);
+    sm.notifyInput(inputMessage);
+    sm.notifyInputClosed();
+    sm.isReadyToExecute();
+    sm.input(); // consume the input command -> Processing, ready for sys_call
+  }
+
+  @TearDown(Level.Invocation)
+  public void freeVm() {
+    sm.close();
+  }
+
+  @Benchmark
+  public void sysCall(Blackhole bh) {
+    bh.consume(sm.call(target, payload, null, null, null, null));
+    bh.consume(sm.takeOutput());
+  }
+}

sdk-core/src/main/java23/dev/restate/sdk/core/statemachine/ffm/FfmStateMachine.java

@@ -158,12 +158,15 @@ public Slice takeOutput() {
       return Slice.EMPTY;
     }
     try (Arena arena = Arena.ofConfined()) {
-      // take_output never fails: it writes a bare Slice (not a BufferResult). Hand the buffer over
-      // zero-copy (GC-tied): it flows straight into the transport (Netty wraps
-      // asReadOnlyByteBuffer()) without a copy-out.
+      // take_output never fails: it writes a bare Slice (not a BufferResult). Copy the buffer out
+      // into a heap byte[] and free the native buffer immediately (takeSliceBytes). This is the
+      // safe
+      // default while we chase a GC-tied-lifetime flake in the bidi path: the zero-copy
+      // wrapOwnedSlice variant handed the buffer to the async output stream and could be collected
+      // before the stream consumed it, intermittently dropping a flushed chunk.
       MemorySegment out = FfmEncoding.allocateSliceStruct(arena);
       SharedCoreNative.vm_take_output(vmHandle, out);
-      return FfmEncoding.wrapOwnedSlice(out);
+      return Slice.wrap(FfmEncoding.takeSliceBytes(out));
     }
   }
 

sdk-core/src/main/rust/src/lib.rs

@@ -160,7 +160,10 @@ pub struct Slice {
 }
 
 impl Slice {
-    const EMPTY: Slice = Slice { ptr: std::ptr::null(), len: 0 };
+    const EMPTY: Slice = Slice {
+        ptr: std::ptr::null(),
+        len: 0,
+    };
 
     /// Leak an owned buffer into a `Slice` the caller frees via `free_buffer`.
     #[inline]
@@ -169,7 +172,7 @@ impl Slice {
             return Self::EMPTY;
         }
         let len = v.len();
-        let ptr =    Box::into_raw(v.into_boxed_slice()) as *mut u8;
+        let ptr = Box::into_raw(v.into_boxed_slice()) as *mut u8;
         Slice { ptr, len }
     }
 
@@ -188,7 +191,8 @@ impl Slice {
     /// Re-take ownership of a transferred-in payload (Java filled an `alloc_buffer` buffer) as
     /// `Bytes`, zero-copy. The caller must not touch the buffer after the call.
     #[inline]
-    unsafe fn take(self) -> Bytes { let ptr = self.ptr as *mut u8;
+    unsafe fn take(self) -> Bytes {
+        let ptr = self.ptr as *mut u8;
         let len = self.len;
         if ptr.is_null() || len == 0 {
             Bytes::new()
@@ -227,21 +231,20 @@ impl ForeignSlice {
     /// Borrow the foreign bytes for the duration of the call (Java keeps ownership and frees them).
     #[inline]
     unsafe fn as_slice<'a>(self) -> &'a [u8] {
-            if self.ptr.is_null() || self.len == 0 {
-                &[]
-            } else {
-                slice::from_raw_parts(self.ptr, self.len)
-            }
+        if self.ptr.is_null() || self.len == 0 {
+            &[]
+        } else {
+            slice::from_raw_parts(self.ptr, self.len)
         }
+    }
 
     /// Borrow the foreign bytes as str for the duration of the call (Java keeps ownership and frees them).
     #[inline]
     unsafe fn as_str<'a>(self) -> &'a str {
         if self.ptr.is_null() || self.len == 0 {
             ""
         } else {
-            str::from_utf8_unchecked(
-            slice::from_raw_parts(self.ptr, self.len))
+            str::from_utf8_unchecked(slice::from_raw_parts(self.ptr, self.len))
         }
     }
 }
@@ -454,7 +457,12 @@ fn vm_take_output(vm: &mut CoreVM) -> Slice {
 #[export_name = "vm_get_response_content_type"]
 pub unsafe extern "C" fn _vm_get_response_content_type(handle: *mut VmHandle, out: *mut Slice) {
     let h = vm_mut(handle);
-    write_out(out, vm_get_response_content_type(&mut h.vm).map(Slice::from_string).unwrap_or( Slice::EMPTY));
+    write_out(
+        out,
+        vm_get_response_content_type(&mut h.vm)
+            .map(Slice::from_string)
+            .unwrap_or(Slice::EMPTY),
+    );
 }
 
 #[inline]
@@ -727,10 +735,7 @@ pub unsafe extern "C" fn _vm_sys_state_set(
     out: *mut EmptyResult,
 ) {
     let h = vm_mut(handle);
-    write_out(
-        out,
-        vm_sys_state_set(&mut h.vm, key.as_str(), value.take()),
-    );
+    write_out(out, vm_sys_state_set(&mut h.vm, key.as_str(), value.take()));
 }
 
 #[inline]
@@ -872,7 +877,11 @@ fn complete_awakeable(vm: &mut CoreVM, id: &str, value: NonEmptyValue) -> EmptyR
 // =========================================================================
 
 #[export_name = "vm_sys_call"]
-pub unsafe extern "C" fn _vm_sys_call(handle: *mut VmHandle, args: CallArguments, out: *mut CallResult) {
+pub unsafe extern "C" fn _vm_sys_call(
+    handle: *mut VmHandle,
+    args: CallArguments,
+    out: *mut CallResult,
+) {
     let h = vm_mut(handle);
     let target = args.borrow();
     let input = args.input.take();
@@ -1368,10 +1377,12 @@ impl CallArguments {
     /// actual `Target` is built by the safe `BorrowedTarget::into_core`.
     #[inline]
     unsafe fn borrow<'a>(&self) -> BorrowedTarget<'a> {
-        let opt = |fs: ForeignSlice| (!fs.ptr.is_null()).then(|| fs.as_slice());
+        // Strings cross as already-valid UTF-8 (Java encoded them), so borrow them unchecked as
+        // `&str`; `headers` is the encoded header-list blob and stays raw bytes.
+        let opt = |fs: ForeignSlice| (!fs.ptr.is_null()).then(|| fs.as_str());
         BorrowedTarget {
-            service: self.service.as_slice(),
-            handler: self.handler.as_slice(),
+            service: self.service.as_str(),
+            handler: self.handler.as_str(),
             key: opt(self.key),
             idempotency_key: opt(self.idempotency_key),
             scope: opt(self.scope),
@@ -1391,19 +1402,19 @@ struct BorrowedTarget<'a> {
     idempotency_key: Option<&'a str>,
     scope: Option<&'a str>,
     limit_key: Option<&'a str>,
-    headers: &'a str,
+    headers: &'a [u8],
 }
 
 impl BorrowedTarget<'_> {
     #[inline]
     fn into_core(self) -> Target {
         Target {
-            service: utf8(self.service).to_owned(),
-            handler: utf8(self.handler).to_owned(),
-            key: self.key.map(|k| utf8(k).to_owned()),
-            idempotency_key: self.idempotency_key.map(|k| utf8(k).to_owned()),
-            scope: self.scope.map(|s| utf8(s).to_owned()),
-            limit_key: self.limit_key.map(|s| utf8(s).to_owned()),
+            service: self.service.to_owned(),
+            handler: self.handler.to_owned(),
+            key: self.key.map(str::to_owned),
+            idempotency_key: self.idempotency_key.map(str::to_owned),
+            scope: self.scope.map(str::to_owned),
+            limit_key: self.limit_key.map(str::to_owned),
             headers: decode_header_list(self.headers)
                 .into_iter()
                 .map(|(k, v)| Header {