diff --git a/docs/operations/docker.md b/docs/operations/docker.md index 356c489..05188f6 100644 --- a/docs/operations/docker.md +++ b/docs/operations/docker.md @@ -216,7 +216,8 @@ The [`Dockerfile`](../../Dockerfile) has two stages: 1. **build**: `ghcr.io/graalvm/native-image-community:25`. Runs `./mvnw install -DskipTests` and then `./mvnw -Pnative package -pl server`, producing a native `hstore` executable linked against glibc (Serial GC, `-O2`, - `-march=compatibility` so the image runs on any CPU of the target architecture). A BuildKit cache mount + `-march=compatibility` so the image runs on any CPU of the target architecture, and shared arenas so segment maps + can be closed when a segment is deleted). A BuildKit cache mount keeps `~/.m2` between builds. 2. **runtime**: `debian:bookworm-slim` plus the binary, the entrypoint and the non-root user. diff --git a/docs/storage/pages.md b/docs/storage/pages.md index b93dc6a..a4372e2 100644 --- a/docs/storage/pages.md +++ b/docs/storage/pages.md @@ -193,14 +193,41 @@ Compared with fixed page slots, packing removes almost all internal fragmentatio ### Reading (`SegmentFile.read(position, maximum)`) -The length of an image is not known before its header is read, so a read takes at most two system calls: - -1. Allocate one buffer of `min(pageSize, 4096)` bytes and read it at `offset × 64`. This is the only allocation for most images. -2. If at least 80 bytes arrived, compute `length = 80 + payloadLength` from header offset 24: - * if `length` is already covered by the first read, or exceeds `pageSize` (a corrupt header), return the first buffer; `verify` then decides; - * otherwise, and only then, allocate a second buffer of exactly `length` bytes, copy the first read into it, and read the remainder. - -Most nodes fit in 4 KiB, so most reads are a single 4 KiB `pread`. A short read at end of file returns whatever arrived, and the header check then rejects it. +Each segment file has a read-only memory map. If the map covers the image, `read` copies exactly +`80 + payloadLength` bytes out of it into a new heap array and returns that. There is no system call, and the raw +bytes stay in the OS page cache rather than on the Java heap, which only holds the decoded nodes in `NodeCache`. + +The copy is deliberate. `ByteCursor` reads one byte at a time through `MemorySegment`, and the JIT only makes that +fast when every segment it sees is the same kind. Decoding straight from the map, while the WAL, catalog and +feed decode from heap arrays, made decoding two to five times slower everywhere in a microbenchmark. Copying a +page costs well under a microsecond and keeps every decoder on heap memory. + +If the image ends past the map, the file may need mapping again: + +* A sealed segment is remapped at its current size right away. The store seals a segment when it rolls to the + next one, and seals every segment except the active one when it opens. +* The active segment is remapped only after the file has grown 4 MiB past the map. Until then, reads near the end + of the file use `pread`. This keeps a segment that is still filling up from being remapped on every new page. + +Each map has its own `Arena.ofShared()`. When a segment is remapped, the old map is closed as soon as the new one is +in place, and deleting or closing a segment closes its map. A deleted segment's disk space therefore comes back +straight away instead of whenever the garbage collector gets to the map. A read that was copying from a map when it +was closed gets an `IllegalStateException`, sees that the map is no longer alive, and reads again from the current +one. Readers only touch a map for the length of one copy, and compaction only deletes segments that no reader can +reach, so this retry is rare. Closing a shared arena in a Native Image build needs `-H:+SharedArenaSupport`, which +GraalVM still lists as experimental; the `native` profile in `server/pom.xml` turns it on. Segment files are only +ever appended to or deleted, never truncated, so a map can't end up pointing past the end of its file. + +`pread` doesn't know how long an image is until it has the header, so it takes one or two system calls: + +1. Allocate a buffer of `min(pageSize, 4096)` bytes and read it at `offset × 64`. +2. If at least 80 bytes came back, compute `length = 80 + payloadLength` from header offset 24: + * if `length` is larger than `pageSize`, the header is corrupt; return the buffer and let `verify` fail it; + * if the first read already has `length` bytes, return those; + * otherwise allocate a buffer of exactly `length` bytes, copy in what was read, and read the rest. + +A short read at the end of the file returns what it got, and the header check rejects it. Both paths return +read-only memory, so nothing can write into a stored image through a read. ## Caching and I/O accounting diff --git a/engine/src/main/java/io/hstore/engine/page/PageHeader.java b/engine/src/main/java/io/hstore/engine/page/PageHeader.java index 4d014d3..0617e81 100644 --- a/engine/src/main/java/io/hstore/engine/page/PageHeader.java +++ b/engine/src/main/java/io/hstore/engine/page/PageHeader.java @@ -3,6 +3,8 @@ import io.hstore.engine.HStoreException; import java.lang.foreign.MemorySegment; +import java.lang.foreign.ValueLayout; +import java.nio.ByteOrder; public record PageHeader( PageType type, @@ -24,6 +26,7 @@ public record PageHeader( private static final int PAGE_ID_OFFSET = 8; private static final int PAYLOAD_LENGTH_OFFSET = 24; private static final int CHECKSUM_OFFSET = 72; + private static final ValueLayout.OfInt INT = ValueLayout.JAVA_INT_UNALIGNED.withOrder(ByteOrder.LITTLE_ENDIAN); public void writeTo(ByteCursor page) { page.position(0) @@ -55,6 +58,10 @@ public static int payloadLength(MemorySegment page) { return ByteCursor.over(page).position(PAYLOAD_LENGTH_OFFSET).getInt(); } + static int payloadLengthAt(MemorySegment file, long position) { + return file.get(INT, position + PAYLOAD_LENGTH_OFFSET); + } + public static void seal(MemorySegment page, int payloadLength) { int checksum = Checksums.crc32c(page.asSlice(0, CHECKSUM_OFFSET), page.asSlice(SIZE, payloadLength)); ByteCursor.over(page).position(CHECKSUM_OFFSET).putInt(checksum); diff --git a/engine/src/main/java/io/hstore/engine/page/PageStore.java b/engine/src/main/java/io/hstore/engine/page/PageStore.java index f58464f..fc866ee 100644 --- a/engine/src/main/java/io/hstore/engine/page/PageStore.java +++ b/engine/src/main/java/io/hstore/engine/page/PageStore.java @@ -67,6 +67,11 @@ private void restore(Collection known) { activeSegment = info.id(); nextUnit = Math.toIntExact(info.units()); }); + files.forEach((id, file) -> { + if (id != activeSegment) { + file.seal(); + } + }); } public int pageSize() { @@ -95,6 +100,10 @@ public long allocate(long epoch, int length) { private void rollSegment() { if (activeSegment != 0) { segments.computeIfPresent(activeSegment, (_, info) -> info.withState(SegmentState.SEALED, 0)); + SegmentFile previous = files.get(activeSegment); + if (previous != null) { + previous.seal(); + } } int id = segments.isEmpty() ? 1 : segments.lastKey() + 1; files.put(id, SegmentFile.open(id, SegmentFile.pathFor(directory, id))); @@ -120,6 +129,11 @@ public void write(long pageId, MemorySegment image) { bytesWritten.addAndGet(image.byteSize()); } + long mappedBytes(int segment) { + SegmentFile file = files.get(segment); + return file == null ? 0 : file.mappedBytes(); + } + public MemorySegment read(long pageId) { SegmentFile file = files.get(PageId.segmentOf(pageId)); if (file == null) { diff --git a/engine/src/main/java/io/hstore/engine/page/SegmentFile.java b/engine/src/main/java/io/hstore/engine/page/SegmentFile.java index d3b096d..b85a81e 100644 --- a/engine/src/main/java/io/hstore/engine/page/SegmentFile.java +++ b/engine/src/main/java/io/hstore/engine/page/SegmentFile.java @@ -4,22 +4,30 @@ import java.io.IOException; import java.io.UncheckedIOException; +import java.lang.foreign.Arena; import java.lang.foreign.MemorySegment; +import java.lang.foreign.ValueLayout; import java.nio.ByteBuffer; import java.nio.channels.FileChannel; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.StandardOpenOption; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.ReentrantLock; final class SegmentFile implements AutoCloseable { private static final int FIRST_READ = 4096; + private static final long REMAP_BYTES = 4L << 20; private final int id; private final Path path; private final FileChannel channel; private final AtomicBoolean dirty = new AtomicBoolean(); + private final ReentrantLock remapping = new ReentrantLock(); + private volatile MemorySegment mapped = MemorySegment.NULL; + private Arena arena; + private volatile boolean sealed; private SegmentFile(int id, Path path, FileChannel channel) { this.id = id; @@ -57,19 +65,102 @@ long unitsOnDisk() { } } + void seal() { + sealed = true; + } + MemorySegment read(long position, int maximum) { + while (true) { + MemorySegment view = covering(position + PageHeader.SIZE); + if (view == null) { + return copy(position, maximum); + } + try { + return read(view, position, maximum); + } catch (IllegalStateException closed) { + if (view.scope().isAlive()) { + throw closed; + } + } + } + } + + private MemorySegment read(MemorySegment view, long position, int maximum) { + long length = PageHeader.SIZE + Integer.toUnsignedLong(PageHeader.payloadLengthAt(view, position)); + if (length > maximum) { + return onHeap(view, position, Math.min(maximum, view.byteSize() - position)); + } + if (position + length > view.byteSize()) { + view = covering(position + length); + if (view == null) { + return copy(position, maximum); + } + } + return onHeap(view, position, length); + } + + long mappedBytes() { + return mapped.byteSize(); + } + + private static MemorySegment onHeap(MemorySegment view, long position, long length) { + byte[] image = new byte[Math.toIntExact(length)]; + MemorySegment.copy(view, ValueLayout.JAVA_BYTE, position, image, 0, image.length); + return MemorySegment.ofArray(image).asReadOnly(); + } + + private MemorySegment covering(long required) { + MemorySegment current = mapped; + if (required <= current.byteSize()) { + return current; + } + if (!sealed && required - current.byteSize() < REMAP_BYTES) { + return null; + } + remapping.lock(); + try { + current = mapped; + if (required <= current.byteSize()) { + return current; + } + long size = channel.size(); + if (size < required) { + return null; + } + Arena replacement = Arena.ofShared(); + try { + current = channel.map(FileChannel.MapMode.READ_ONLY, 0, size, replacement); + } catch (IOException | RuntimeException e) { + replacement.close(); + throw e; + } + mapped = current; + unmap(); + arena = replacement; + return current; + } catch (IOException e) { + throw HStoreException.io("cannot map segment " + id, e); + } finally { + remapping.unlock(); + } + } + + private MemorySegment copy(long position, int maximum) { ByteBuffer first = ByteBuffer.allocate(Math.min(maximum, FIRST_READ)); fill(first, position); if (first.position() < PageHeader.SIZE) { - return MemorySegment.ofBuffer(first.flip()); + return MemorySegment.ofBuffer(first.flip()).asReadOnly(); } int length = PageHeader.SIZE + PageHeader.payloadLength(MemorySegment.ofArray(first.array())); - if (length <= first.position() || length > maximum) { - return MemorySegment.ofBuffer(first.flip()); + if (length < PageHeader.SIZE || length > maximum) { + return MemorySegment.ofBuffer(first.flip()).asReadOnly(); + } + if (length <= first.position()) { + return MemorySegment.ofBuffer(first.flip()).asSlice(0, length).asReadOnly(); } ByteBuffer whole = ByteBuffer.allocate(length).put(first.flip()); fill(whole, position); - return MemorySegment.ofBuffer(whole.flip()); + return MemorySegment.ofBuffer(whole.flip()).asReadOnly(); } private void fill(ByteBuffer buffer, long position) { @@ -116,8 +207,22 @@ void delete() { } } + private void unmap() { + if (arena != null) { + arena.close(); + arena = null; + } + } + @Override public void close() { + remapping.lock(); + try { + mapped = MemorySegment.NULL; + unmap(); + } finally { + remapping.unlock(); + } try { channel.close(); } catch (IOException e) { diff --git a/engine/src/test/java/io/hstore/engine/page/PageStoreTest.java b/engine/src/test/java/io/hstore/engine/page/PageStoreTest.java new file mode 100644 index 0000000..cb5628b --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/page/PageStoreTest.java @@ -0,0 +1,156 @@ +package io.hstore.engine.page; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.lang.foreign.MemorySegment; +import java.lang.foreign.ValueLayout; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class PageStoreTest { + + private static final int PAGE_SIZE = 4096; + private static final int PAGES_PER_SEGMENT = 2048; + private static final int PAYLOAD_LENGTH_OFFSET = 24; + + @TempDir + Path directory; + + @Test + void pagesReadBackExactlyAsWrittenAcrossSegmentsAndReopens() { + Random random = new Random(7); + Map written = new LinkedHashMap<>(); + List ids = new ArrayList<>(); + List segments; + try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) { + for (int i = 0; i < 16_000; i++) { + byte[] image = image(random, PageHeader.SIZE + random.nextInt(PAGE_SIZE - PageHeader.SIZE + 1)); + long pageId = store.allocate(1, image.length); + store.write(pageId, MemorySegment.ofArray(image)); + written.put(pageId, image); + ids.add(pageId); + for (int probe = 0; probe < 3; probe++) { + long earlier = ids.get(random.nextInt(ids.size())); + assertArrayEquals(written.get(earlier), bytes(store.read(earlier)), "page " + PageId.unpack(earlier)); + } + } + assertTrue(store.segments().size() > 3, "the test should span several segments"); + written.forEach((pageId, image) -> assertArrayEquals(image, bytes(store.read(pageId)))); + segments = store.segments(); + } + try (PageStore reopened = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, segments)) { + written.forEach((pageId, image) -> assertArrayEquals(image, bytes(reopened.read(pageId)))); + long first = ids.getFirst(); + reopened.read(first); + assertTrue(reopened.mappedBytes(PageId.segmentOf(first)) > 0); + } + } + + @Test + void pagesCannotBeChangedThroughWhatReadReturns() { + try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) { + long first = write(store, PAGE_SIZE); + fillPastOneSegment(store); + long last = write(store, 200); + for (long pageId : new long[]{first, last}) { + MemorySegment page = store.read(pageId); + assertThrows(IllegalArgumentException.class, () -> page.set(ValueLayout.JAVA_BYTE, 0, (byte) 1)); + } + } + } + + @Test + void finishedSegmentsAreMappedAndTheGrowingOneIsNotUntilItGrows() { + try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) { + long finished = write(store, PAGE_SIZE); + fillPastOneSegment(store); + long growing = write(store, PAGE_SIZE); + store.read(finished); + store.read(growing); + assertTrue(store.mappedBytes(PageId.segmentOf(finished)) >= (long) PAGES_PER_SEGMENT * PAGE_SIZE - PAGE_SIZE); + assertEquals(0, store.mappedBytes(PageId.segmentOf(growing))); + for (int i = 0; i < 1100; i++) { + write(store, PAGE_SIZE); + } + long later = write(store, PAGE_SIZE); + store.read(later); + assertTrue(store.mappedBytes(PageId.segmentOf(later)) > 4L << 20); + } + } + + @Test + void readersKeepSeeingWholePagesWhileOldMapsAreClosed() throws Exception { + int pages = 6_000; + long[] ids = new long[pages]; + byte[][] images = new byte[pages][]; + AtomicInteger written = new AtomicInteger(); + try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of()); + ExecutorService readers = Executors.newFixedThreadPool(4)) { + List> checks = new ArrayList<>(); + for (int t = 0; t < 4; t++) { + Random random = new Random(t); + checks.add(readers.submit(() -> { + long reads = 0; + while (written.get() < pages) { + int known = written.get(); + if (known > 0) { + int i = random.nextInt(known); + assertArrayEquals(images[i], bytes(store.read(ids[i])), "page " + i); + reads++; + } + } + return reads; + })); + } + Random random = new Random(11); + for (int i = 0; i < pages; i++) { + images[i] = image(random, PAGE_SIZE - random.nextInt(64)); + ids[i] = store.allocate(1, images[i].length); + store.write(ids[i], MemorySegment.ofArray(images[i])); + written.incrementAndGet(); + } + for (Future check : checks) { + assertTrue(check.get() > 0); + } + } + } + + private static long write(PageStore store, int length) { + byte[] image = image(new Random(length), length); + long pageId = store.allocate(1, image.length); + store.write(pageId, MemorySegment.ofArray(image)); + return pageId; + } + + private static void fillPastOneSegment(PageStore store) { + int segment = store.activeSegment(); + while (store.activeSegment() == segment) { + write(store, PAGE_SIZE); + } + } + + private static byte[] image(Random random, int length) { + byte[] image = new byte[length]; + random.nextBytes(image); + ByteCursor.wrap(image).position(PAYLOAD_LENGTH_OFFSET).putInt(length - PageHeader.SIZE); + return image; + } + + private static byte[] bytes(MemorySegment page) { + return page.toArray(ValueLayout.JAVA_BYTE); + } +} diff --git a/server/pom.xml b/server/pom.xml index 2a70029..6634617 100644 --- a/server/pom.xml +++ b/server/pom.xml @@ -90,6 +90,8 @@ -O2 --gc=serial -H:+ReportExceptionStackTraces + -H:+UnlockExperimentalVMOptions + -H:+SharedArenaSupport --enable-url-protocols=http,https -march=compatibility