Update FFMJavaTask and build.py for FFM hot-path changes in aws-crt-java

FFMJavaTask.java:
- Override onResponseBody(MemorySegment,...) for zero-copy downloads
  (discards data as the benchmark only measures throughput)
- Implement sendRequestBody(long address, long length) -> int in
  UploadFromRamStream for direct-write uploads via MemorySegment.copy()
- Set withUseFFM(true) on S3MetaRequestOptions to activate FFM callbacks
- Add resetPosition() and getLength() to UploadFromRamStream

scripts/utils/build.py:
- Add --enable-native-access=ALL-UNNAMED to the java runner launch
  command (required for MemorySegment.ofAddress() on Java 22+)

Author: sbSteveK

runners/s3-benchrunner-java/src/main/java/com/example/s3benchrunner/ffmjava/FFMJavaTask.java

@@ -8,7 +8,9 @@
 import software.amazon.awssdk.crt.http.HttpRequestBodyStream;
 import software.amazon.awssdk.crt.s3.*;
 
-import java.nio.ByteBuffer;
+import java.lang.foreign.Arena;
+import java.lang.foreign.MemorySegment;
+import java.lang.foreign.ValueLayout;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
 import java.util.ArrayList;
@@ -17,13 +19,18 @@
 
 /**
  * A single transfer task (upload or download) executed via the FFM-backed
- * aws-crt-java S3Client. The logic is identical to CRTJavaTask; the FFM
- * internals are transparent at this level since aws-crt-java exposes the same
- * public API regardless of whether JNI or FFM is used under the hood.
- *
- * If the FFM branch of aws-crt-java changes the HttpRequestBodyStream
- * sendRequestBody() signature (e.g. ByteBuffer -> MemorySegment), update
- * UploadFromRamStream below accordingly.
+ * aws-crt-java S3Client.
+ * <p>
+ * This task uses {@link S3MetaRequestOptions#withUseFFM(boolean) useFFM=true}
+ * so that:
+ * <ul>
+ *   <li><b>Downloads:</b> response body chunks are delivered as
+ *       {@link MemorySegment} views of native memory (zero-copy, no
+ *       {@code byte[]} allocation).</li>
+ *   <li><b>Uploads:</b> the upload stream writes directly into the native
+ *       buffer via {@link MemorySegment} (no {@code DirectByteBuffer} wrapper
+ *       object allocation).</li>
+ * </ul>
  */
 class FFMJavaTask implements S3MetaRequestResponseHandler {
 
@@ -43,6 +50,9 @@ class FFMJavaTask implements S3MetaRequestResponseHandler {
 
         options.withResponseHandler(this);
 
+        // Enable FFM mode: zero-copy downloads, direct-write uploads
+        options.withUseFFM(true);
+
         String httpMethod;
         String httpPath = "/" + config.key;
         HttpRequestBodyStream requestUploadStream = null;
@@ -100,11 +110,21 @@ void waitUntilDone() {
         }
     }
 
+    /**
+     * FFM download path: body chunk delivered as a zero-copy {@link MemorySegment}
+     * view of native memory. The benchmark discards the data, so we just return 0.
+     */
+    @Override
+    public int onResponseBody(MemorySegment bodyBytesIn, long objectRangeStart, long objectRangeEnd) {
+        // Benchmark discards downloaded data — no backpressure increment needed.
+        return 0;
+    }
+
     @Override
     public void onFinished(S3FinishedResponseContext context) {
         if (context.getErrorCode() != 0) {
             // FFMJavaTask failed. Report error and kill program...
-            System.err.printf("FFMJavaTask[%d] failed. actions:%s key:%s error_code:%s/n",
+            System.err.printf("FFMJavaTask[%d] failed. action:%s key:%s error_code:%s%n",
                     taskI, config.action, config.key, CRT.awsErrorName(context.getErrorCode()));
 
             if (context.getResponseStatus() != 0) {
@@ -128,6 +148,14 @@ public void onFinished(S3FinishedResponseContext context) {
         }
     }
 
+    /**
+     * FFM upload stream: writes random data directly into the native buffer via
+     * {@link MemorySegment}, avoiding the {@code DirectByteBuffer} wrapper object
+     * that the JNI path allocates on every call.
+     * <p>
+     * Returns the number of bytes written. Returning {@code 0} signals
+     * end-of-stream to the native layer.
+     */
     static class UploadFromRamStream implements HttpRequestBodyStream {
         final long size;
         long bytesWritten;
@@ -138,21 +166,45 @@ static class UploadFromRamStream implements HttpRequestBodyStream {
             this.size = size;
         }
 
+        /**
+         * FFM path: write directly into the native buffer at {@code address}.
+         * Returns bytes written; 0 signals end-of-stream.
+         */
         @Override
-        public boolean sendRequestBody(ByteBuffer dstBuf) {
-            /*
-             * `randomData` is just a buffer of random data whose length may not equal
-             * `size`. We'll send its contents repeatedly until size bytes have been
-             * uploaded. We do this, so we can upload huge objects without actually
-             * allocating a huge buffer.
-             */
-            while (bytesWritten < size && dstBuf.remaining() > 0) {
-                int amtToTransfer = (int) Math.min(size - bytesWritten, dstBuf.remaining());
-                amtToTransfer = Math.min(amtToTransfer, randomData.length);
-                dstBuf.put(randomData, 0, amtToTransfer);
-                bytesWritten += amtToTransfer;
+        public int sendRequestBody(long address, long length) {
+            long remaining = size - bytesWritten;
+            if (remaining <= 0) {
+                return 0; // end-of-stream
+            }
+
+            long toWrite = Math.min(remaining, length);
+
+            // Wrap the native destination buffer as a MemorySegment (zero-copy).
+            MemorySegment dest = MemorySegment.ofAddress(address)
+                    .reinterpret(toWrite, Arena.ofAuto(), null);
+
+            // Copy from randomData (looping) into the native segment.
+            long written = 0;
+            while (written < toWrite) {
+                int chunk = (int) Math.min(toWrite - written, randomData.length);
+                MemorySegment.copy(MemorySegment.ofArray(randomData), ValueLayout.JAVA_BYTE, 0,
+                        dest, ValueLayout.JAVA_BYTE, written, chunk);
+                written += chunk;
             }
-            return bytesWritten == size;
+
+            bytesWritten += written;
+            return (int) written;
+        }
+
+        @Override
+        public boolean resetPosition() {
+            bytesWritten = 0;
+            return true;
+        }
+
+        @Override
+        public long getLength() {
+            return size;
         }
     }
 }

scripts/utils/build.py

@@ -213,9 +213,11 @@ def _build_java(work_dir: Path, branch: Optional[str]) -> list[str]:
          '-Dawscrt.version=1.0.0-SNAPSHOT',
          ])
 
-    # return command for running the jar
+    # return command for running the jar.
+    # --enable-native-access=ALL-UNNAMED is required for Java 22+ FFM API
+    # (MemorySegment.ofAddress() on arbitrary native pointers).
     jar_path = runner_src/'target/s3-benchrunner-java-1.0-SNAPSHOT.jar'
-    return ['java', '-jar', str(jar_path)]
+    return ['java', '--enable-native-access=ALL-UNNAMED', '-jar', str(jar_path)]
 
 
 def _build_rust(work_dir: Path, branch: Optional[str]) -> list[str]: