fix(core): avoid ARM64 Java 8 native atomic crashes

.github/workflows/ci.yml

@@ -22,9 +22,23 @@ jobs:
   # longer committed, so this job compiles libquestdb.so from source (hence the
   # zstd submodule + cmake/nasm/build-essential toolchain) before the tests run.
   build-jdk8:
-    name: Build, test & javadoc (JDK 8)
-    runs-on: ubuntu-latest
+    name: ${{ matrix.name }}
+    runs-on: ${{ matrix.os }}
     timeout-minutes: 45
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - name: Build, test & javadoc (JDK 8)
+            os: ubuntu-latest
+            jdk_arch: x64
+            platform: linux-x86-64
+            cmake_args: ""
+          - name: Build, test & javadoc (JDK 8, ARM64)
+            os: ubuntu-22.04-arm
+            jdk_arch: aarch64
+            platform: linux-aarch64
+            cmake_args: "-DCMAKE_TOOLCHAIN_FILE=./src/main/c/toolchains/linux-arm64.cmake"
     steps:
       - name: Check out
         uses: actions/checkout@v4
@@ -33,22 +47,30 @@ jobs:
           submodules: recursive
 
       - name: Set up JDK 11 (MRJAR versioned classes)
-        # Exports JAVA_HOME_11_X64. The JDK 8 build compiles src/main/java11 with
-        # it into META-INF/versions/11 (see the antrun execution in core/pom.xml);
-        # JDK 8 javac cannot compile that source root.
+        # Keep its path before JDK 8 replaces JAVA_HOME. The JDK 8 build uses
+        # this compiler for META-INF/versions/11 on either architecture.
+        id: jdk11
         uses: actions/setup-java@v4
         with:
           distribution: temurin
           java-version: "11"
+          architecture: ${{ matrix.jdk_arch }}
 
       - name: Set up JDK 8
         # Last setup-java call wins JAVA_HOME: the build itself runs on JDK 8.
         uses: actions/setup-java@v4
         with:
           distribution: temurin
           java-version: "8"
+          architecture: ${{ matrix.jdk_arch }}
           cache: maven
 
+      - name: Show test runtime
+        run: |
+          java -version
+          mvn -version
+          uname -m
+
       - name: Install native build toolchain
         run: sudo apt-get update && sudo apt-get install -y cmake nasm build-essential
 
@@ -57,24 +79,38 @@ jobs:
         # Java 8 JNI headers -- the artifact's Java floor. Copy it into src
         # resources (not target/) so it survives the `mvn clean` in the next step
         # and gets packaged + loaded via the production bin/<platform> path.
-        # NOTE: this builds on ubuntu-latest for FUNCTIONAL testing only; the
+        # NOTE: this builds on the Ubuntu runners for FUNCTIONAL testing only; the
         # library's glibc runtime floor is validated separately by the
         # `glibc-floor` job, which rebuilds in the release low-glibc container.
         run: |
           cd core
-          cmake -DCMAKE_BUILD_TYPE=Release -B cmake-build-release -S.
+          cmake ${{ matrix.cmake_args }} -DCMAKE_BUILD_TYPE=Release -B cmake-build-release -S.
           cmake --build cmake-build-release --config Release
           test -f target/classes/io/questdb/client/bin-local/libquestdb.so
-          mkdir -p src/main/resources/io/questdb/client/bin/linux-x86-64
+          mkdir -p "src/main/resources/io/questdb/client/bin/${{ matrix.platform }}"
           cp target/classes/io/questdb/client/bin-local/libquestdb.so \
-             src/main/resources/io/questdb/client/bin/linux-x86-64/libquestdb.so
+             "src/main/resources/io/questdb/client/bin/${{ matrix.platform }}/libquestdb.so"
 
       - name: Compile, test, and build javadoc
         env:
-          JAVA11_HOME: ${{ env.JAVA_HOME_11_X64 }}
+          JAVA11_HOME: ${{ steps.jdk11.outputs.path }}
         run: mvn -B -ntp -P javadoc clean install
 
+      - name: Upload test failure diagnostics
+        if: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name: jdk8-diagnostics-${{ matrix.platform }}
+          path: |
+            core/hs_err_pid*.log
+            core/target/surefire-reports/
+            core/target/failsafe-reports/
+          if-no-files-found: warn
+          retention-days: 7
+
       - name: Upload client jar for the cross-JDK smoke job
+        # The downstream smoke job runs on x64; upload only its matching jar.
+        if: matrix.jdk_arch == 'x64'
         uses: actions/upload-artifact@v4
         with:
           name: questdb-client-jar

core/src/main/java/io/questdb/client/std/Unsafe.java

@@ -31,18 +31,15 @@
 import java.util.concurrent.atomic.LongAdder;
 
 public final class Unsafe {
-    // The various _ADDR fields are `long` in Java, but they are `* mut usize` in Rust, or `size_t*` in C.
-    // These are off-heap allocated atomic counters for memory usage tracking.
-
     public static final long BYTE_OFFSET;
     private static final LongAdder[] COUNTERS = new LongAdder[MemoryTag.SIZE];
-    private static final long FREE_COUNT_ADDR;
-    private static final long MALLOC_COUNT_ADDR;
-    private static final long[] NATIVE_MEM_COUNTER_ADDRS = new long[MemoryTag.SIZE];
-    private static final long NON_RSS_MEM_USED_ADDR;
+    private static final long FREE_COUNT_OFFSET;
+    private static final long[] GLOBAL_COUNTERS = new long[5];
+    private static final long MALLOC_COUNT_OFFSET;
+    private static final long NON_RSS_MEM_USED_OFFSET;
     private static final long OVERRIDE;
-    private static final long REALLOC_COUNT_ADDR;
-    private static final long RSS_MEM_USED_ADDR;
+    private static final long REALLOC_COUNT_OFFSET;
+    private static final long RSS_MEM_USED_OFFSET;
     private static final sun.misc.Unsafe UNSAFE;
 
     private Unsafe() {
@@ -87,49 +84,49 @@ public static long getFieldOffset(Class<?> clazz, String name) {
     }
 
     public static long getFreeCount() {
-        return UNSAFE.getLongVolatile(null, FREE_COUNT_ADDR);
+        return UNSAFE.getLongVolatile(GLOBAL_COUNTERS, FREE_COUNT_OFFSET);
     }
 
     public static long getMallocCount() {
-        return UNSAFE.getLongVolatile(null, MALLOC_COUNT_ADDR);
+        return UNSAFE.getLongVolatile(GLOBAL_COUNTERS, MALLOC_COUNT_OFFSET);
     }
 
     /**
      * Get the total memory used by the process, this includes both resident memory
      * and that assigned to memory mapped files.
      */
     public static long getMemUsed() {
-        return UNSAFE.getLongVolatile(null, NON_RSS_MEM_USED_ADDR) +
-                UNSAFE.getLongVolatile(null, RSS_MEM_USED_ADDR);
+        return UNSAFE.getLongVolatile(GLOBAL_COUNTERS, NON_RSS_MEM_USED_OFFSET) +
+                UNSAFE.getLongVolatile(GLOBAL_COUNTERS, RSS_MEM_USED_OFFSET);
     }
 
     public static long getMemUsedByTag(int memoryTag) {
         assert memoryTag >= 0 && memoryTag < MemoryTag.SIZE;
-        return COUNTERS[memoryTag].sum() + UNSAFE.getLongVolatile(null, NATIVE_MEM_COUNTER_ADDRS[memoryTag]);
+        return COUNTERS[memoryTag].sum();
     }
 
     public static long getReallocCount() {
-        return UNSAFE.getLongVolatile(null, REALLOC_COUNT_ADDR);
+        return UNSAFE.getLongVolatile(GLOBAL_COUNTERS, REALLOC_COUNT_OFFSET);
     }
 
     public static long getRssMemUsed() {
-        return UNSAFE.getLongVolatile(null, RSS_MEM_USED_ADDR);
+        return UNSAFE.getLongVolatile(GLOBAL_COUNTERS, RSS_MEM_USED_OFFSET);
     }
 
     public static sun.misc.Unsafe getUnsafe() {
         return UNSAFE;
     }
 
     public static void incrFreeCount() {
-        UNSAFE.getAndAddLong(null, FREE_COUNT_ADDR, 1);
+        UNSAFE.getAndAddLong(GLOBAL_COUNTERS, FREE_COUNT_OFFSET, 1);
     }
 
     public static void incrMallocCount() {
-        UNSAFE.getAndAddLong(null, MALLOC_COUNT_ADDR, 1);
+        UNSAFE.getAndAddLong(GLOBAL_COUNTERS, MALLOC_COUNT_OFFSET, 1);
     }
 
     public static void incrReallocCount() {
-        UNSAFE.getAndAddLong(null, REALLOC_COUNT_ADDR, 1);
+        UNSAFE.getAndAddLong(GLOBAL_COUNTERS, REALLOC_COUNT_OFFSET, 1);
     }
 
     /**
@@ -190,10 +187,10 @@ public static void recordMemAlloc(long size, int memoryTag) {
         assert memoryTag >= 0 && memoryTag < MemoryTag.SIZE;
         COUNTERS[memoryTag].add(size);
         if (memoryTag >= MemoryTag.NATIVE_DEFAULT) {
-            final long mem = UNSAFE.getAndAddLong(null, RSS_MEM_USED_ADDR, size) + size;
+            final long mem = UNSAFE.getAndAddLong(GLOBAL_COUNTERS, RSS_MEM_USED_OFFSET, size) + size;
             assert mem >= 0 : "unexpected RSS mem: " + mem + ", size: " + size + ", memoryTag:" + memoryTag;
         } else {
-            final long mem = UNSAFE.getAndAddLong(null, NON_RSS_MEM_USED_ADDR, size) + size;
+            final long mem = UNSAFE.getAndAddLong(GLOBAL_COUNTERS, NON_RSS_MEM_USED_OFFSET, size) + size;
             assert mem >= 0 : "unexpected non-RSS mem: " + mem + ", size: " + size + ", memoryTag:" + memoryTag;
         }
     }
@@ -243,28 +240,17 @@ private static boolean isJava8Or11() {
             throw new ExceptionInInitializerError(e);
         }
 
-        // A single allocation for all the off-heap native memory counters.
-        // Might help with locality, given they're often incremented together.
-        // All initial values set to 0.
-        final long nativeMemCountersArraySize = (5 + COUNTERS.length) * 8;
-        final long nativeMemCountersArray = UNSAFE.allocateMemory(nativeMemCountersArraySize);
-        long ptr = nativeMemCountersArray;
-        Vect.memset(nativeMemCountersArray, nativeMemCountersArraySize, 0);
-
-        RSS_MEM_USED_ADDR = ptr;
-        ptr += 8;
-        MALLOC_COUNT_ADDR = ptr;
-        ptr += 8;
-        REALLOC_COUNT_ADDR = ptr;
-        ptr += 8;
-        FREE_COUNT_ADDR = ptr;
-        ptr += 8;
-        NON_RSS_MEM_USED_ADDR = ptr;
-        ptr += 8;
+        long base = UNSAFE.arrayBaseOffset(long[].class);
+        long scale = UNSAFE.arrayIndexScale(long[].class);
+        RSS_MEM_USED_OFFSET = base;
+        MALLOC_COUNT_OFFSET = base + scale;
+        REALLOC_COUNT_OFFSET = base + 2 * scale;
+        FREE_COUNT_OFFSET = base + 3 * scale;
+        NON_RSS_MEM_USED_OFFSET = base + 4 * scale;
         for (int i = 0; i < COUNTERS.length; i++) {
             COUNTERS[i] = new LongAdder();
-            NATIVE_MEM_COUNTER_ADDRS[i] = ptr;
-            ptr += 8;
         }
+        // Preserve the eager native-library load previously triggered by Vect.memset.
+        Os.init();
     }
 }

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/DurableAckIntegrationTest.java

@@ -284,10 +284,6 @@ public void testRequestHeaderCarriesConfiguredTierSet() throws Exception {
 
     @Test
     public void testEndToEndBothTiersLocalAckIsProgressOnly() throws Exception {
-        // request_durable_ack=local,replicated: the replicated ack is the trim
-        // trigger; local acks surface early per-table progress without popping
-        // anything. The loop's counters and the per-table local watermark are
-        // the observable surface for that split.
         TestUtils.assertMemoryLeak(() -> {
             DurableAckCapableHandler handler = new DurableAckCapableHandler();
             try (TestWebSocketServer server = new TestWebSocketServer(handler, true)) {
@@ -298,42 +294,35 @@ public void testEndToEndBothTiersLocalAckIsProgressOnly() throws Exception {
                 String config = "ws::addr=localhost:" + port + ";sf_dir=" + sfDir
                         + ";request_durable_ack=local,replicated;close_flush_timeout_millis=5000;";
                 try (Sender sender = Sender.fromConfig(config)) {
-                    for (int i = 0; i < 10; i++) {
-                        sender.table("trades").longColumn("v", i).atNow();
-                    }
-                    sender.flush(); // one batch -> one OK
+                    sender.table("trades").longColumn("v", 1L).atNow();
+                    sender.flush();
                     handler.awaitOkBatches(1);
-                    long batches = 1;
 
                     CursorWebSocketSendLoop loop =
                             ((QwpWebSocketSender) sender).cursorSendLoopForTest();
                     Assert.assertNotNull(loop);
 
-                    // Release local acks covering everything OK'd, then nudge
-                    // the connection with extra rows until the I/O thread has
-                    // observed one -- the local ack alone must not trim.
                     handler.emitLocalDurableAckForAll();
-                    long deadline = System.currentTimeMillis() + 5000;
-                    while (loop.getTotalLocalDurableAcks() == 0
-                            && System.currentTimeMillis() < deadline) {
-                        sender.table("trades").longColumn("v", -1L).atNow();
+                    try {
+                        // The follow-up OK publishes the preceding local ACK's updates.
+                        // No more local ACKs are sent before reading the watermark.
+                        sender.table("trades").longColumn("v", 2L).atNow();
                         sender.flush();
-                        batches++;
-                        Thread.sleep(10);
+                        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+                        while (loop.getTotalAcks() < 2 && System.nanoTime() < deadline) {
+                            Thread.sleep(10);
+                        }
+                        Assert.assertEquals("follow-up OK not observed", 2L, loop.getTotalAcks());
+                        Assert.assertEquals("local ack not observed", 1L, loop.getTotalLocalDurableAcks());
+                        Assert.assertEquals("local watermark tracks the first batch's fsync frontier",
+                                0L, loop.getLocalDurableTableWatermark("trades"));
+                        Assert.assertEquals(
+                                "local acks must not advance trim when replicated is requested",
+                                0L, loop.getTotalDurableTrimAdvances());
+                        Assert.assertEquals("neither batch is replicated yet", -1L, sender.getAckedFsn());
+                    } finally {
+                        handler.emitDurableAckForAll();
                     }
-                    Assert.assertTrue("local ack never observed",
-                            loop.getTotalLocalDurableAcks() > 0);
-                    Assert.assertTrue("local watermark tracks the fsync frontier",
-                            loop.getLocalDurableTableWatermark("trades") >= 0);
-                    Assert.assertEquals(
-                            "local acks must not advance trim when replicated is requested",
-                            0L, loop.getTotalDurableTrimAdvances());
-
-                    // Cover every batch sent (nudges included) with a
-                    // replicated ack so close() drains on the actual trim
-                    // trigger.
-                    handler.awaitOkBatches(batches);
-                    handler.emitDurableAckForAll();
                 }
             }
         });