diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 79565db9..38b50b8e 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 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 1382a1d9..0f0f9a21 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,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,11 +84,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,21 +96,21 @@ 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() { @@ -121,15 +118,15 @@ public static sun.misc.Unsafe getUnsafe() { } 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(); } } 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 85b7b9b6..6836e919 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 @@ -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(); } } }); 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 00000000..bdbed700 --- /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)); + } +}