From 03568574812f61dff8d7120dd778dd6c0c6d6ec5 Mon Sep 17 00:00:00 2001 From: victor Date: Tue, 3 Nov 2026 20:32:33 -0800 Subject: [PATCH 1/6] fix(qwp): publish local durable ack progress before counting it --- .../sf/cursor/CursorWebSocketSendLoop.java | 28 +++++---- ...CursorWebSocketSendLoopDurableAckTest.java | 60 +++++++++++++++++++ 2 files changed, 78 insertions(+), 10 deletions(-) diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java index 21d171ed9..003068be1 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java @@ -311,6 +311,7 @@ public final class CursorWebSocketSendLoop implements QuietCloseable { // when BOTH tiers are requested. Progress observability only -- the trim // path never reads it (the replicated ack drives the trim). In local-only // mode local acks feed durableTableWatermarks directly instead. + // Guarded by the map's monitor because progress readers run off the I/O thread. private final CharSequenceLongHashMap localDurableTableWatermarks = new CharSequenceLongHashMap(); // Pre-converted to nanos. Consulted only by the orphan terminal policy. Zero disables // the dwell entirely (count-only escalation at MAX_CATCHUP_CAP_GAP_ATTEMPTS); the @@ -1747,7 +1748,9 @@ public long getTotalLocalDurableAcks() { * this map stays empty. */ public long getLocalDurableTableWatermark(CharSequence tableName) { - return localDurableTableWatermarks.get(tableName); + synchronized (localDurableTableWatermarks) { + return localDurableTableWatermarks.get(tableName); + } } /** @@ -2047,13 +2050,15 @@ private void applyDurableAck() { * observability only: the trim path is driven by the replicated ack. */ private void applyLocalDurableAckProgress() { - int n = response.getTableEntryCount(); - for (int i = 0; i < n; i++) { - String name = response.getTableName(i); - long seqTxn = response.getTableSeqTxn(i); - long current = localDurableTableWatermarks.get(name); - if (seqTxn > current) { - localDurableTableWatermarks.put(name, seqTxn); + synchronized (localDurableTableWatermarks) { + int n = response.getTableEntryCount(); + for (int i = 0; i < n; i++) { + String name = response.getTableName(i); + long seqTxn = response.getTableSeqTxn(i); + long current = localDurableTableWatermarks.get(name); + if (seqTxn > current) { + localDurableTableWatermarks.put(name, seqTxn); + } } } } @@ -2106,7 +2111,9 @@ private void clearDurableAckTracking() { releasePendingEntry(pendingDurable.pollFirst()); } durableTableWatermarks.clear(); - localDurableTableWatermarks.clear(); + synchronized (localDurableTableWatermarks) { + localDurableTableWatermarks.clear(); + } } private static boolean isCatchUpCapGap(Throwable t) { @@ -4405,7 +4412,6 @@ public void onBinaryMessage(long payloadPtr, int payloadLen) { LOG.warn("received STATUS_LOCAL_DURABLE_ACK frame without opt-in -- ignoring"); return; } - totalLocalDurableAcks.incrementAndGet(); if (isLocalAckTrimming) { // Local is the strongest requested tier, so its ack is // this connection's trim trigger -- same watermark and @@ -4416,6 +4422,8 @@ public void onBinaryMessage(long payloadPtr, int payloadLen) { // local ack is an early progress signal only. applyLocalDurableAckProgress(); } + // Publish the count only after this frame's watermarks are visible. + totalLocalDurableAcks.incrementAndGet(); return; } // Application-layer rejection by the server. Classify by status diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopDurableAckTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopDurableAckTest.java index dcf4eb40f..96707faff 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopDurableAckTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopDurableAckTest.java @@ -42,8 +42,13 @@ import java.lang.reflect.Method; import java.nio.charset.StandardCharsets; import java.nio.file.Paths; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -323,6 +328,61 @@ public void testBothTiersLocalAckIsProgressOnly() throws Exception { }); } + @Test(timeout = 30_000L) + public void testLocalAckCountPublishesWatermarks() throws Exception { + TestUtils.assertMemoryLeak(() -> { + try (CursorSendEngine engine = newEngine()) { + CursorWebSocketSendLoop loop = newBothTiersLoop(engine); + CountDownLatch applyingAck = new CountDownLatch(1); + CountDownLatch releaseAck = new CountDownLatch(1); + AtomicReference failure = new AtomicReference<>(); + setField(loop, "response", new WebSocketResponse() { + @Override + public long getTableSeqTxn(int index) { + if (index == 0) { + applyingAck.countDown(); + try { + assertTrue("ack was not released", releaseAck.await(5, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } + return super.getTableSeqTxn(index); + } + }); + Thread writer = new Thread(() -> { + try { + deliverLocalDurableAck(loop, names("trades", "orders"), txns(7L, 9L)); + } catch (Throwable t) { + failure.set(t); + } + }, "local-ack-writer"); + writer.start(); + try { + assertTrue("ack processing did not start", applyingAck.await(5, TimeUnit.SECONDS)); + assertEquals("ack count must not advance before its watermarks", 0L, + loop.getTotalLocalDurableAcks()); + releaseAck.countDown(); + + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (loop.getTotalLocalDurableAcks() == 0 && System.nanoTime() < deadline) { + Thread.sleep(1); + } + assertEquals(1L, loop.getTotalLocalDurableAcks()); + assertEquals(7L, loop.getLocalDurableTableWatermark("trades")); + assertEquals(9L, loop.getLocalDurableTableWatermark("orders")); + assertEquals(0L, loop.getTotalDurableTrimAdvances()); + } finally { + releaseAck.countDown(); + writer.join(5_000L); + assertFalse("ack writer did not exit", writer.isAlive()); + } + assertNull(failure.get()); + } + }); + } + @Test public void testBothTiersLocalWatermarkIsMonotonic() throws Exception { // A delayed/duplicate local ack naming a smaller seqTxn must not move From 291bcc80a0baf8deb72d85bedc9097d2fafe1bab Mon Sep 17 00:00:00 2001 From: victor Date: Wed, 4 Nov 2026 20:13:45 -0800 Subject: [PATCH 2/6] Test the Java 8 client on ARM64 Run the Java 8 build and full test suite on a native Linux ARM64 runner alongside x64, without disabling JIT compilation. Select matching JDKs and native resource paths per architecture. Preserve the x64 smoke-test artifact and upload JVM crash reports and Maven test diagnostics when either test job fails. --- .github/workflows/ci.yml | 56 +++++++++++++++++++++++++++++++++------- 1 file changed, 46 insertions(+), 10 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 79565db95..38b50b8ef 100644 --- a/.github/workflows/ci.yml +++ b/.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,13 +47,14 @@ 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. @@ -47,8 +62,15 @@ jobs: 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/ 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 From 1f832763c96f41fefad0eb51c1b1a827bae08cc5 Mon Sep 17 00:00:00 2001 From: victor Date: Wed, 4 Nov 2026 21:19:43 -0800 Subject: [PATCH 3/6] Keep durable ACK synchronization in tests Wait for a follow-up OK before checking the preceding local ACK watermark and trim state. Preserve the assertions and release the replicated ACK in finally for cleanup. Remove the production locking and counter-publication changes, along with the test that required the new publication contract. The production sources now match the PR baseline. --- .../sf/cursor/CursorWebSocketSendLoop.java | 28 ++++----- .../client/sf/DurableAckIntegrationTest.java | 51 +++++++--------- ...CursorWebSocketSendLoopDurableAckTest.java | 60 ------------------- 3 files changed, 30 insertions(+), 109 deletions(-) diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java index 003068be1..21d171ed9 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java @@ -311,7 +311,6 @@ public final class CursorWebSocketSendLoop implements QuietCloseable { // when BOTH tiers are requested. Progress observability only -- the trim // path never reads it (the replicated ack drives the trim). In local-only // mode local acks feed durableTableWatermarks directly instead. - // Guarded by the map's monitor because progress readers run off the I/O thread. private final CharSequenceLongHashMap localDurableTableWatermarks = new CharSequenceLongHashMap(); // Pre-converted to nanos. Consulted only by the orphan terminal policy. Zero disables // the dwell entirely (count-only escalation at MAX_CATCHUP_CAP_GAP_ATTEMPTS); the @@ -1748,9 +1747,7 @@ public long getTotalLocalDurableAcks() { * this map stays empty. */ public long getLocalDurableTableWatermark(CharSequence tableName) { - synchronized (localDurableTableWatermarks) { - return localDurableTableWatermarks.get(tableName); - } + return localDurableTableWatermarks.get(tableName); } /** @@ -2050,15 +2047,13 @@ private void applyDurableAck() { * observability only: the trim path is driven by the replicated ack. */ private void applyLocalDurableAckProgress() { - synchronized (localDurableTableWatermarks) { - int n = response.getTableEntryCount(); - for (int i = 0; i < n; i++) { - String name = response.getTableName(i); - long seqTxn = response.getTableSeqTxn(i); - long current = localDurableTableWatermarks.get(name); - if (seqTxn > current) { - localDurableTableWatermarks.put(name, seqTxn); - } + int n = response.getTableEntryCount(); + for (int i = 0; i < n; i++) { + String name = response.getTableName(i); + long seqTxn = response.getTableSeqTxn(i); + long current = localDurableTableWatermarks.get(name); + if (seqTxn > current) { + localDurableTableWatermarks.put(name, seqTxn); } } } @@ -2111,9 +2106,7 @@ private void clearDurableAckTracking() { releasePendingEntry(pendingDurable.pollFirst()); } durableTableWatermarks.clear(); - synchronized (localDurableTableWatermarks) { - localDurableTableWatermarks.clear(); - } + localDurableTableWatermarks.clear(); } private static boolean isCatchUpCapGap(Throwable t) { @@ -4412,6 +4405,7 @@ public void onBinaryMessage(long payloadPtr, int payloadLen) { LOG.warn("received STATUS_LOCAL_DURABLE_ACK frame without opt-in -- ignoring"); return; } + totalLocalDurableAcks.incrementAndGet(); if (isLocalAckTrimming) { // Local is the strongest requested tier, so its ack is // this connection's trim trigger -- same watermark and @@ -4422,8 +4416,6 @@ public void onBinaryMessage(long payloadPtr, int payloadLen) { // local ack is an early progress signal only. applyLocalDurableAckProgress(); } - // Publish the count only after this frame's watermarks are visible. - totalLocalDurableAcks.incrementAndGet(); return; } // Application-layer rejection by the server. Classify by status diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/DurableAckIntegrationTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/DurableAckIntegrationTest.java index 2fd2b1e9a..7eb6ec5c9 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/DurableAckIntegrationTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/DurableAckIntegrationTest.java @@ -279,10 +279,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)) { @@ -293,42 +289,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(); } } }); diff --git a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopDurableAckTest.java b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopDurableAckTest.java index 96707faff..dcf4eb40f 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopDurableAckTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoopDurableAckTest.java @@ -42,13 +42,8 @@ import java.lang.reflect.Method; import java.nio.charset.StandardCharsets; import java.nio.file.Paths; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReference; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -328,61 +323,6 @@ public void testBothTiersLocalAckIsProgressOnly() throws Exception { }); } - @Test(timeout = 30_000L) - public void testLocalAckCountPublishesWatermarks() throws Exception { - TestUtils.assertMemoryLeak(() -> { - try (CursorSendEngine engine = newEngine()) { - CursorWebSocketSendLoop loop = newBothTiersLoop(engine); - CountDownLatch applyingAck = new CountDownLatch(1); - CountDownLatch releaseAck = new CountDownLatch(1); - AtomicReference failure = new AtomicReference<>(); - setField(loop, "response", new WebSocketResponse() { - @Override - public long getTableSeqTxn(int index) { - if (index == 0) { - applyingAck.countDown(); - try { - assertTrue("ack was not released", releaseAck.await(5, TimeUnit.SECONDS)); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new AssertionError(e); - } - } - return super.getTableSeqTxn(index); - } - }); - Thread writer = new Thread(() -> { - try { - deliverLocalDurableAck(loop, names("trades", "orders"), txns(7L, 9L)); - } catch (Throwable t) { - failure.set(t); - } - }, "local-ack-writer"); - writer.start(); - try { - assertTrue("ack processing did not start", applyingAck.await(5, TimeUnit.SECONDS)); - assertEquals("ack count must not advance before its watermarks", 0L, - loop.getTotalLocalDurableAcks()); - releaseAck.countDown(); - - long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); - while (loop.getTotalLocalDurableAcks() == 0 && System.nanoTime() < deadline) { - Thread.sleep(1); - } - assertEquals(1L, loop.getTotalLocalDurableAcks()); - assertEquals(7L, loop.getLocalDurableTableWatermark("trades")); - assertEquals(9L, loop.getLocalDurableTableWatermark("orders")); - assertEquals(0L, loop.getTotalDurableTrimAdvances()); - } finally { - releaseAck.countDown(); - writer.join(5_000L); - assertFalse("ack writer did not exit", writer.isAlive()); - } - assertNull(failure.get()); - } - }); - } - @Test public void testBothTiersLocalWatermarkIsMonotonic() throws Exception { // A delayed/duplicate local ack naming a smaller seqTxn must not move From d26385b28a7e475227652c54043896aa5f777d2a Mon Sep 17 00:00:00 2001 From: victor Date: Thu, 5 Nov 2026 07:38:11 -0800 Subject: [PATCH 4/6] Avoid native atomic crashes on ARM64 Java 8 Use a CAS fallback for native counter updates on Java 8 ARM64 HotSpot. Keep the original intrinsic directly at unaffected call sites, with a constant-false Java 11+ shim to avoid wrapper costs. Cover allocation, reallocation, signed accounting adjustments and concurrent counter updates. Check the packaged modern fast path and run the counter regression under C1 on ARM64 CI. --- .github/workflows/ci.yml | 8 + .../java/io/questdb/client/std/Unsafe.java | 27 ++- .../questdb/client/std/NativeAtomicLong.java | 36 ++++ .../questdb/client/std/NativeAtomicLong.java | 49 +++++ .../client/test/std/JarPackagingIT.java | 13 ++ .../test/std/UnsafeMemoryAccountingTest.java | 178 ++++++++++++++++++ 6 files changed, 306 insertions(+), 5 deletions(-) create mode 100644 core/src/main/java11/io/questdb/client/std/NativeAtomicLong.java create mode 100644 core/src/main/java8/io/questdb/client/std/NativeAtomicLong.java create mode 100644 core/src/test/java/io/questdb/client/test/std/UnsafeMemoryAccountingTest.java diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 38b50b8ef..c96d60db6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -91,6 +91,14 @@ jobs: cp target/classes/io/questdb/client/bin-local/libquestdb.so \ "src/main/resources/io/questdb/client/bin/${{ matrix.platform }}/libquestdb.so" + - name: Test native memory accounting with C1 + if: matrix.jdk_arch == 'aarch64' + run: >- + mvn -B -ntp -pl core -Dtest=UnsafeMemoryAccountingTest + '-DargLine=-ea -Dfile.encoding=UTF-8 -XX:+UseParallelGC + -Dslf4j.provider=ch.qos.logback.classic.spi.LogbackServiceProvider + -XX:+TieredCompilation -XX:TieredStopAtLevel=1' test + - name: Compile, test, and build javadoc env: JAVA11_HOME: ${{ steps.jdk11.outputs.path }} diff --git a/core/src/main/java/io/questdb/client/std/Unsafe.java b/core/src/main/java/io/questdb/client/std/Unsafe.java index 1382a1d90..7fa0f5360 100644 --- a/core/src/main/java/io/questdb/client/std/Unsafe.java +++ b/core/src/main/java/io/questdb/client/std/Unsafe.java @@ -120,16 +120,29 @@ public static sun.misc.Unsafe getUnsafe() { return UNSAFE; } + // Keep unaffected calls direct: newer C1 versions may not inline an atomic wrapper. public static void incrFreeCount() { - UNSAFE.getAndAddLong(null, FREE_COUNT_ADDR, 1); + if (NativeAtomicLong.IS_CAS_REQUIRED) { + NativeAtomicLong.getAndAdd(UNSAFE, FREE_COUNT_ADDR, 1); + } else { + UNSAFE.getAndAddLong(null, FREE_COUNT_ADDR, 1); + } } public static void incrMallocCount() { - UNSAFE.getAndAddLong(null, MALLOC_COUNT_ADDR, 1); + if (NativeAtomicLong.IS_CAS_REQUIRED) { + NativeAtomicLong.getAndAdd(UNSAFE, MALLOC_COUNT_ADDR, 1); + } else { + UNSAFE.getAndAddLong(null, MALLOC_COUNT_ADDR, 1); + } } public static void incrReallocCount() { - UNSAFE.getAndAddLong(null, REALLOC_COUNT_ADDR, 1); + if (NativeAtomicLong.IS_CAS_REQUIRED) { + NativeAtomicLong.getAndAdd(UNSAFE, REALLOC_COUNT_ADDR, 1); + } else { + UNSAFE.getAndAddLong(null, REALLOC_COUNT_ADDR, 1); + } } /** @@ -190,10 +203,14 @@ 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 = (NativeAtomicLong.IS_CAS_REQUIRED + ? NativeAtomicLong.getAndAdd(UNSAFE, RSS_MEM_USED_ADDR, size) + : UNSAFE.getAndAddLong(null, RSS_MEM_USED_ADDR, 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 = (NativeAtomicLong.IS_CAS_REQUIRED + ? NativeAtomicLong.getAndAdd(UNSAFE, NON_RSS_MEM_USED_ADDR, size) + : UNSAFE.getAndAddLong(null, NON_RSS_MEM_USED_ADDR, size)) + size; assert mem >= 0 : "unexpected non-RSS mem: " + mem + ", size: " + size + ", memoryTag:" + memoryTag; } } diff --git a/core/src/main/java11/io/questdb/client/std/NativeAtomicLong.java b/core/src/main/java11/io/questdb/client/std/NativeAtomicLong.java new file mode 100644 index 000000000..b1085e191 --- /dev/null +++ b/core/src/main/java11/io/questdb/client/std/NativeAtomicLong.java @@ -0,0 +1,36 @@ +/*+***************************************************************************** + * ___ _ ____ ____ + * / _ \ _ _ ___ ___| |_| _ \| __ ) + * | | | | | | |/ _ \/ __| __| | | | _ \ + * | |_| | |_| | __/\__ \ |_| |_| | |_) | + * \__\_\\__,_|\___||___/\__|____/|____/ + * + * Copyright (c) 2014-2019 Appsicle + * Copyright (c) 2019-2026 QuestDB + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + ******************************************************************************/ + +package io.questdb.client.std; + +final class NativeAtomicLong { + static final boolean IS_CAS_REQUIRED = false; + + private NativeAtomicLong() { + } + + static long getAndAdd(sun.misc.Unsafe unsafe, long address, long delta) { + return unsafe.getAndAddLong(null, address, delta); + } +} diff --git a/core/src/main/java8/io/questdb/client/std/NativeAtomicLong.java b/core/src/main/java8/io/questdb/client/std/NativeAtomicLong.java new file mode 100644 index 000000000..54400f06a --- /dev/null +++ b/core/src/main/java8/io/questdb/client/std/NativeAtomicLong.java @@ -0,0 +1,49 @@ +/*+***************************************************************************** + * ___ _ ____ ____ + * / _ \ _ _ ___ ___| |_| _ \| __ ) + * | | | | | | |/ _ \/ __| __| | | | _ \ + * | |_| | |_| | __/\__ \ |_| |_| | |_) | + * \__\_\\__,_|\___||___/\__|____/|____/ + * + * Copyright (c) 2014-2019 Appsicle + * Copyright (c) 2019-2026 QuestDB + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + ******************************************************************************/ + +package io.questdb.client.std; + +final class NativeAtomicLong { + static final boolean IS_CAS_REQUIRED; + + private NativeAtomicLong() { + } + + static long getAndAdd(sun.misc.Unsafe unsafe, long address, long delta) { + // Java 8 AArch64 C1 truncates constant native offsets in getAndAddLong. + long previous; + do { + previous = unsafe.getLongVolatile(null, address); + } while (!unsafe.compareAndSwapLong(null, address, previous, previous + delta)); + return previous; + } + + static { + String arch = System.getProperty("os.arch"); + String vmName = System.getProperty("java.vm.name", ""); + IS_CAS_REQUIRED = ("aarch64".equals(arch) || "arm64".equals(arch)) + && (vmName.startsWith("OpenJDK 64-Bit Server VM") + || vmName.startsWith("Java HotSpot(TM) 64-Bit Server VM")); + } +} diff --git a/core/src/test/java/io/questdb/client/test/std/JarPackagingIT.java b/core/src/test/java/io/questdb/client/test/std/JarPackagingIT.java index 8ddaf8a83..7c674dec4 100644 --- a/core/src/test/java/io/questdb/client/test/std/JarPackagingIT.java +++ b/core/src/test/java/io/questdb/client/test/std/JarPackagingIT.java @@ -50,7 +50,9 @@ */ public class JarPackagingIT { private static final String FD_BIG_ENTRY = "io/questdb/client/std/FdBig.class"; + private static final String NATIVE_ATOMIC_ENTRY = "io/questdb/client/std/NativeAtomicLong.class"; private static final String VERSIONED_FD_BIG_ENTRY = "META-INF/versions/11/" + FD_BIG_ENTRY; + private static final String VERSIONED_NATIVE_ATOMIC_ENTRY = "META-INF/versions/11/" + NATIVE_ATOMIC_ENTRY; @Test public void testDoubleFormattingAgainstPackagedJar() throws Exception { @@ -124,6 +126,9 @@ public void testJarLayout() throws Exception { "META-INF/versions/11 must carry the java11 Compat shim", jar.getEntry("META-INF/versions/11/io/questdb/client/std/Compat.class") ); + Assert.assertTrue("Java 8 must carry the native atomic fallback", + classReferences(jar, NATIVE_ATOMIC_ENTRY, "compareAndSwapLong")); + assertNativeAtomicFastPath(jar, VERSIONED_NATIVE_ATOMIC_ENTRY); // every source in src/main/java11 must ship in versions/11 -- a file added // to the source root but missed by the packaging step would recreate the // NoClassDefFoundError class of bug on Java 9+ @@ -141,10 +146,18 @@ public void testJarLayout() throws Exception { classReferences(jar, FD_BIG_ENTRY, "jdk.internal.math.FDBigInteger") ); Assert.assertNotNull("module-info.class missing", jar.getEntry("module-info.class")); + assertNativeAtomicFastPath(jar, NATIVE_ATOMIC_ENTRY); } } } + private static void assertNativeAtomicFastPath(JarFile jar, String entry) throws IOException { + Assert.assertTrue("Java 11+ must retain the atomic intrinsic", + classReferences(jar, entry, "getAndAddLong")); + Assert.assertFalse("Java 11+ must not use the CAS fallback", + classReferences(jar, entry, "compareAndSwapLong")); + } + private static java.util.List collectJavaSources(File dir, String prefix) { java.util.List result = new java.util.ArrayList(); File[] files = dir.listFiles(); diff --git a/core/src/test/java/io/questdb/client/test/std/UnsafeMemoryAccountingTest.java b/core/src/test/java/io/questdb/client/test/std/UnsafeMemoryAccountingTest.java new file mode 100644 index 000000000..bdbed7002 --- /dev/null +++ b/core/src/test/java/io/questdb/client/test/std/UnsafeMemoryAccountingTest.java @@ -0,0 +1,178 @@ +/*+***************************************************************************** + * ___ _ ____ ____ + * / _ \ _ _ ___ ___| |_| _ \| __ ) + * | | | | | | |/ _ \/ __| __| | | | _ \ + * | |_| | |_| | __/\__ \ |_| |_| | |_) | + * \__\_\\__,_|\___||___/\__|____/|____/ + * + * Copyright (c) 2014-2019 Appsicle + * Copyright (c) 2019-2026 QuestDB + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + ******************************************************************************/ + +package io.questdb.client.test.std; + +import io.questdb.client.std.MemoryTag; +import io.questdb.client.std.Unsafe; +import io.questdb.client.test.tools.TestUtils; +import org.junit.Assert; +import org.junit.Test; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +public class UnsafeMemoryAccountingTest { + private static final int ITERATIONS = 100_000; + + @Test + public void testAllocationAccounting() throws Exception { + TestUtils.assertMemoryLeak(() -> { + assertAllocationAccounting(MemoryTag.NATIVE_DEFAULT, true); + assertAllocationAccounting(MemoryTag.NATIVE_PATH, false); + }); + } + + @Test + public void testConcurrentCounters() throws Exception { + TestUtils.assertMemoryLeak(() -> { + long mallocs = Unsafe.getMallocCount(); + long reallocs = Unsafe.getReallocCount(); + long frees = Unsafe.getFreeCount(); + int threadCount = 4; + CountDownLatch start = new CountDownLatch(1); + ExecutorService executor = Executors.newFixedThreadPool(threadCount, runnable -> { + Thread thread = new Thread(runnable, "native-counter-test"); + thread.setDaemon(true); + return thread; + }); + try { + Future[] tasks = new Future[threadCount]; + for (int i = 0; i < threadCount; i++) { + tasks[i] = executor.submit(() -> { + start.await(); + for (int j = 0; j < ITERATIONS && !Thread.currentThread().isInterrupted(); j++) { + allocationCycle((j & 1) == 0 ? MemoryTag.NATIVE_DEFAULT : MemoryTag.NATIVE_PATH); + } + return null; + }); + } + start.countDown(); + for (Future task : tasks) { + task.get(30, TimeUnit.SECONDS); + } + } finally { + start.countDown(); + executor.shutdownNow(); + Assert.assertTrue("counter workers did not stop", executor.awaitTermination(30, TimeUnit.SECONDS)); + } + assertCounts(mallocs, reallocs, frees, (long) threadCount * ITERATIONS); + }); + } + + @Test + public void testLargeAdjustments() throws Exception { + TestUtils.assertMemoryLeak(() -> { + assertAdjustment(MemoryTag.NATIVE_DEFAULT, true); + assertAdjustment(MemoryTag.MMAP_DEFAULT, false); + }); + } + + @Test + public void testRepeatedCounters() throws Exception { + TestUtils.assertMemoryLeak(() -> { + long mallocs = Unsafe.getMallocCount(); + long reallocs = Unsafe.getReallocCount(); + long frees = Unsafe.getFreeCount(); + for (int i = 0; i < ITERATIONS; i++) { + allocationCycle((i & 1) == 0 ? MemoryTag.NATIVE_DEFAULT : MemoryTag.NATIVE_PATH); + } + assertCounts(mallocs, reallocs, frees, ITERATIONS); + }); + } + + private static void allocationCycle(int memoryTag) { + long size = 8; + long ptr = Unsafe.malloc(size, memoryTag); + try { + ptr = Unsafe.realloc(ptr, size, 32, memoryTag); + size = 32; + } finally { + Unsafe.free(ptr, size, memoryTag); + } + } + + private static void assertAdjustment(int memoryTag, boolean isRss) { + long bytes = 1L << 33; + long total = Unsafe.getMemUsed(); + long rss = Unsafe.getRssMemUsed(); + long tagged = Unsafe.getMemUsedByTag(memoryTag); + Unsafe.recordMemAlloc(bytes, memoryTag); + try { + Assert.assertEquals(total + bytes, Unsafe.getMemUsed()); + Assert.assertEquals(rss + (isRss ? bytes : 0), Unsafe.getRssMemUsed()); + Assert.assertEquals(tagged + bytes, Unsafe.getMemUsedByTag(memoryTag)); + Unsafe.recordMemAlloc(0, memoryTag); + Assert.assertEquals(total + bytes, Unsafe.getMemUsed()); + } finally { + Unsafe.recordMemAlloc(-bytes, memoryTag); + } + Assert.assertEquals(total, Unsafe.getMemUsed()); + Assert.assertEquals(rss, Unsafe.getRssMemUsed()); + Assert.assertEquals(tagged, Unsafe.getMemUsedByTag(memoryTag)); + } + + private static void assertAllocationAccounting(int memoryTag, boolean isRss) { + long mallocs = Unsafe.getMallocCount(); + long reallocs = Unsafe.getReallocCount(); + long frees = Unsafe.getFreeCount(); + long total = Unsafe.getMemUsed(); + long rss = Unsafe.getRssMemUsed(); + long tagged = Unsafe.getMemUsedByTag(memoryTag); + long size = 8; + long ptr = Unsafe.malloc(size, memoryTag); + try { + assertMemory(total, rss, tagged, memoryTag, isRss, size); + ptr = Unsafe.realloc(ptr, size, 32, memoryTag); + size = 32; + assertMemory(total, rss, tagged, memoryTag, isRss, size); + ptr = Unsafe.realloc(ptr, size, size, memoryTag); + assertMemory(total, rss, tagged, memoryTag, isRss, size); + ptr = Unsafe.realloc(ptr, size, 4, memoryTag); + size = 4; + assertMemory(total, rss, tagged, memoryTag, isRss, size); + } finally { + Unsafe.free(ptr, size, memoryTag); + } + assertMemory(total, rss, tagged, memoryTag, isRss, 0); + Assert.assertEquals(mallocs + 1, Unsafe.getMallocCount()); + Assert.assertEquals(reallocs + 3, Unsafe.getReallocCount()); + Assert.assertEquals(frees + 1, Unsafe.getFreeCount()); + } + + private static void assertCounts(long mallocs, long reallocs, long frees, long count) { + Assert.assertEquals(mallocs + count, Unsafe.getMallocCount()); + Assert.assertEquals(reallocs + count, Unsafe.getReallocCount()); + Assert.assertEquals(frees + count, Unsafe.getFreeCount()); + } + + private static void assertMemory(long total, long rss, long tagged, int memoryTag, boolean isRss, long size) { + Assert.assertEquals(total + size, Unsafe.getMemUsed()); + Assert.assertEquals(rss + (isRss ? size : 0), Unsafe.getRssMemUsed()); + Assert.assertEquals(tagged + size, Unsafe.getMemUsedByTag(memoryTag)); + } +} From de7cb00bcf9b764bb9529215a688e717febd99a8 Mon Sep 17 00:00:00 2001 From: victor Date: Thu, 5 Nov 2026 17:56:45 -0800 Subject: [PATCH 5/6] Use object-based atomic memory counters Keep the counters in a fixed long array and use the existing atomic intrinsics with object-relative offsets. These offsets stay small on every JDK, avoiding the Java 8 ARM64 C1 native-address bug without version checks, a CAS fallback, or per-update allocation. Remove the unused native per-tag counters, version shims and their packaging checks. Preserve eager native-library initialization. Keep the normal ARM64 JDK 8 test matrix, without an extra C1 run. --- .github/workflows/ci.yml | 8 -- .../java/io/questdb/client/std/Unsafe.java | 86 ++++++------------- .../questdb/client/std/NativeAtomicLong.java | 36 -------- .../questdb/client/std/NativeAtomicLong.java | 49 ----------- .../client/test/std/JarPackagingIT.java | 13 --- 5 files changed, 28 insertions(+), 164 deletions(-) delete mode 100644 core/src/main/java11/io/questdb/client/std/NativeAtomicLong.java delete mode 100644 core/src/main/java8/io/questdb/client/std/NativeAtomicLong.java diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c96d60db6..38b50b8ef 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -91,14 +91,6 @@ jobs: cp target/classes/io/questdb/client/bin-local/libquestdb.so \ "src/main/resources/io/questdb/client/bin/${{ matrix.platform }}/libquestdb.so" - - name: Test native memory accounting with C1 - if: matrix.jdk_arch == 'aarch64' - run: >- - mvn -B -ntp -pl core -Dtest=UnsafeMemoryAccountingTest - '-DargLine=-ea -Dfile.encoding=UTF-8 -XX:+UseParallelGC - -Dslf4j.provider=ch.qos.logback.classic.spi.LogbackServiceProvider - -XX:+TieredCompilation -XX:TieredStopAtLevel=1' test - - name: Compile, test, and build javadoc env: JAVA11_HOME: ${{ steps.jdk11.outputs.path }} diff --git a/core/src/main/java/io/questdb/client/std/Unsafe.java b/core/src/main/java/io/questdb/client/std/Unsafe.java index 7fa0f5360..87c495007 100644 --- a/core/src/main/java/io/questdb/client/std/Unsafe.java +++ b/core/src/main/java/io/questdb/client/std/Unsafe.java @@ -31,18 +31,16 @@ 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; + // An object base keeps atomic offsets small, including on Java 8 AArch64 C1. + 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,11 +85,11 @@ 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); } /** @@ -99,50 +97,37 @@ public static long getMallocCount() { * 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; } - // Keep unaffected calls direct: newer C1 versions may not inline an atomic wrapper. public static void incrFreeCount() { - if (NativeAtomicLong.IS_CAS_REQUIRED) { - NativeAtomicLong.getAndAdd(UNSAFE, FREE_COUNT_ADDR, 1); - } else { - UNSAFE.getAndAddLong(null, FREE_COUNT_ADDR, 1); - } + UNSAFE.getAndAddLong(GLOBAL_COUNTERS, FREE_COUNT_OFFSET, 1); } public static void incrMallocCount() { - if (NativeAtomicLong.IS_CAS_REQUIRED) { - NativeAtomicLong.getAndAdd(UNSAFE, MALLOC_COUNT_ADDR, 1); - } else { - UNSAFE.getAndAddLong(null, MALLOC_COUNT_ADDR, 1); - } + UNSAFE.getAndAddLong(GLOBAL_COUNTERS, MALLOC_COUNT_OFFSET, 1); } public static void incrReallocCount() { - if (NativeAtomicLong.IS_CAS_REQUIRED) { - NativeAtomicLong.getAndAdd(UNSAFE, REALLOC_COUNT_ADDR, 1); - } else { - UNSAFE.getAndAddLong(null, REALLOC_COUNT_ADDR, 1); - } + UNSAFE.getAndAddLong(GLOBAL_COUNTERS, REALLOC_COUNT_OFFSET, 1); } /** @@ -203,14 +188,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 = (NativeAtomicLong.IS_CAS_REQUIRED - ? NativeAtomicLong.getAndAdd(UNSAFE, RSS_MEM_USED_ADDR, size) - : 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 = (NativeAtomicLong.IS_CAS_REQUIRED - ? NativeAtomicLong.getAndAdd(UNSAFE, NON_RSS_MEM_USED_ADDR, size) - : 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; } } @@ -260,28 +241,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 native-library initialization before any allocation can succeed. + Os.init(); } } diff --git a/core/src/main/java11/io/questdb/client/std/NativeAtomicLong.java b/core/src/main/java11/io/questdb/client/std/NativeAtomicLong.java deleted file mode 100644 index b1085e191..000000000 --- a/core/src/main/java11/io/questdb/client/std/NativeAtomicLong.java +++ /dev/null @@ -1,36 +0,0 @@ -/*+***************************************************************************** - * ___ _ ____ ____ - * / _ \ _ _ ___ ___| |_| _ \| __ ) - * | | | | | | |/ _ \/ __| __| | | | _ \ - * | |_| | |_| | __/\__ \ |_| |_| | |_) | - * \__\_\\__,_|\___||___/\__|____/|____/ - * - * Copyright (c) 2014-2019 Appsicle - * Copyright (c) 2019-2026 QuestDB - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - ******************************************************************************/ - -package io.questdb.client.std; - -final class NativeAtomicLong { - static final boolean IS_CAS_REQUIRED = false; - - private NativeAtomicLong() { - } - - static long getAndAdd(sun.misc.Unsafe unsafe, long address, long delta) { - return unsafe.getAndAddLong(null, address, delta); - } -} diff --git a/core/src/main/java8/io/questdb/client/std/NativeAtomicLong.java b/core/src/main/java8/io/questdb/client/std/NativeAtomicLong.java deleted file mode 100644 index 54400f06a..000000000 --- a/core/src/main/java8/io/questdb/client/std/NativeAtomicLong.java +++ /dev/null @@ -1,49 +0,0 @@ -/*+***************************************************************************** - * ___ _ ____ ____ - * / _ \ _ _ ___ ___| |_| _ \| __ ) - * | | | | | | |/ _ \/ __| __| | | | _ \ - * | |_| | |_| | __/\__ \ |_| |_| | |_) | - * \__\_\\__,_|\___||___/\__|____/|____/ - * - * Copyright (c) 2014-2019 Appsicle - * Copyright (c) 2019-2026 QuestDB - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - ******************************************************************************/ - -package io.questdb.client.std; - -final class NativeAtomicLong { - static final boolean IS_CAS_REQUIRED; - - private NativeAtomicLong() { - } - - static long getAndAdd(sun.misc.Unsafe unsafe, long address, long delta) { - // Java 8 AArch64 C1 truncates constant native offsets in getAndAddLong. - long previous; - do { - previous = unsafe.getLongVolatile(null, address); - } while (!unsafe.compareAndSwapLong(null, address, previous, previous + delta)); - return previous; - } - - static { - String arch = System.getProperty("os.arch"); - String vmName = System.getProperty("java.vm.name", ""); - IS_CAS_REQUIRED = ("aarch64".equals(arch) || "arm64".equals(arch)) - && (vmName.startsWith("OpenJDK 64-Bit Server VM") - || vmName.startsWith("Java HotSpot(TM) 64-Bit Server VM")); - } -} diff --git a/core/src/test/java/io/questdb/client/test/std/JarPackagingIT.java b/core/src/test/java/io/questdb/client/test/std/JarPackagingIT.java index 7c674dec4..8ddaf8a83 100644 --- a/core/src/test/java/io/questdb/client/test/std/JarPackagingIT.java +++ b/core/src/test/java/io/questdb/client/test/std/JarPackagingIT.java @@ -50,9 +50,7 @@ */ public class JarPackagingIT { private static final String FD_BIG_ENTRY = "io/questdb/client/std/FdBig.class"; - private static final String NATIVE_ATOMIC_ENTRY = "io/questdb/client/std/NativeAtomicLong.class"; private static final String VERSIONED_FD_BIG_ENTRY = "META-INF/versions/11/" + FD_BIG_ENTRY; - private static final String VERSIONED_NATIVE_ATOMIC_ENTRY = "META-INF/versions/11/" + NATIVE_ATOMIC_ENTRY; @Test public void testDoubleFormattingAgainstPackagedJar() throws Exception { @@ -126,9 +124,6 @@ public void testJarLayout() throws Exception { "META-INF/versions/11 must carry the java11 Compat shim", jar.getEntry("META-INF/versions/11/io/questdb/client/std/Compat.class") ); - Assert.assertTrue("Java 8 must carry the native atomic fallback", - classReferences(jar, NATIVE_ATOMIC_ENTRY, "compareAndSwapLong")); - assertNativeAtomicFastPath(jar, VERSIONED_NATIVE_ATOMIC_ENTRY); // every source in src/main/java11 must ship in versions/11 -- a file added // to the source root but missed by the packaging step would recreate the // NoClassDefFoundError class of bug on Java 9+ @@ -146,18 +141,10 @@ public void testJarLayout() throws Exception { classReferences(jar, FD_BIG_ENTRY, "jdk.internal.math.FDBigInteger") ); Assert.assertNotNull("module-info.class missing", jar.getEntry("module-info.class")); - assertNativeAtomicFastPath(jar, NATIVE_ATOMIC_ENTRY); } } } - private static void assertNativeAtomicFastPath(JarFile jar, String entry) throws IOException { - Assert.assertTrue("Java 11+ must retain the atomic intrinsic", - classReferences(jar, entry, "getAndAddLong")); - Assert.assertFalse("Java 11+ must not use the CAS fallback", - classReferences(jar, entry, "compareAndSwapLong")); - } - private static java.util.List collectJavaSources(File dir, String prefix) { java.util.List result = new java.util.ArrayList(); File[] files = dir.listFiles(); From f86eecc179909b623e95f1f5e397d39d3ed73912 Mon Sep 17 00:00:00 2001 From: victor Date: Thu, 5 Nov 2026 19:25:01 -0800 Subject: [PATCH 6/6] Clarify native-library initialization comment --- core/src/main/java/io/questdb/client/std/Unsafe.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/core/src/main/java/io/questdb/client/std/Unsafe.java b/core/src/main/java/io/questdb/client/std/Unsafe.java index 87c495007..0f0f9a219 100644 --- a/core/src/main/java/io/questdb/client/std/Unsafe.java +++ b/core/src/main/java/io/questdb/client/std/Unsafe.java @@ -34,7 +34,6 @@ public final class Unsafe { public static final long BYTE_OFFSET; private static final LongAdder[] COUNTERS = new LongAdder[MemoryTag.SIZE]; private static final long FREE_COUNT_OFFSET; - // An object base keeps atomic offsets small, including on Java 8 AArch64 C1. 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; @@ -251,7 +250,7 @@ private static boolean isJava8Or11() { for (int i = 0; i < COUNTERS.length; i++) { COUNTERS[i] = new LongAdder(); } - // Preserve native-library initialization before any allocation can succeed. + // Preserve the eager native-library load previously triggered by Vect.memset. Os.init(); } }