diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md index dd3793f1..d689920a 100644 --- a/docs/architecture/overview.md +++ b/docs/architecture/overview.md @@ -83,7 +83,7 @@ The engine has no dependencies outside `java.base`. All three modules build into | Package | Purpose | |---|---| | `io.hstore.engine` | `StorageEngine` (open/close, maintenance thread, checkpoint/compact entry points), `EngineOptions`, `EngineStats`, `Dictionary` (interned role names and role sets), `HStoreException` (error codes, retryable conflicts). | -| `engine.page` | Byte-level storage. `PageStore` and `SegmentFile` hold segment files and allocation. `PageHeader` is the 80-byte header and crc32c seal. `PageId` is a page's permanent number, `PageAddress` its place in a segment, and `PageDirectory` the memory-mapped `pages.dir` that maps one to the other. `ReferenceCounts` is `pages.ref`, the count of references to each page, with its checkpoint journal. `ByteCursor` provides little-endian and varint codecs over `MemorySegment`. `IoTrace` counts page visits per query through a `ScopedValue`. | +| `engine.page` | Byte-level storage. `PageStore` and `SegmentFile` hold segment files and allocation, and `FilePool` bounds how many segment files are open at once. `PageHeader` is the 80-byte header and crc32c seal. `PageId` is a page's permanent number, `PageAddress` its place in a segment, and `PageDirectory` the memory-mapped `pages.dir` that maps one to the other. `ReferenceCounts` is `pages.ref`, the count of references to each page, with its checkpoint journal. `ByteCursor` provides little-endian and varint codecs over `MemorySegment`. `IoTrace` counts page visits per query through a `ScopedValue`. | | `engine.tree` | The copy-on-write counted B+tree. `Tree` is the immutable handle and `TreeWriter` does path-copy updates. Nodes are `Leaf` and `Branch`, referenced by `Ref` (`Empty`, `Pending`, `Stored`, or `Inline` for a small leaf carried inside an edge record). Also here: `Summary` (monoid annotations), `TreeCursor`, `TreeAlgebra` (leapfrog, set algebra), `TreeDiff`, `BulkBuilder`, `NodeCodec`, `NodeCache`, `Materializer` (writes pending nodes to pages at checkpoints), `Epoch` (one checkpoint attempt, for the stamps on written nodes), `TreeWalker` and `TreeVerifier`. | | `engine.topology` | Hyperedge representation: `Hyperedge` (members tree plus order index), `Incidence` (one member's role set, weight, validity, payload and qualifier), `IncidenceCodec` (columnar leaf encoding), `MemberChange`, `Weight` (fixed point) and `Validity`. | | `engine.index` | `PostingIndex`, a two-level index from a key to a postings set. Small sets are inlined and large ones are promoted to their own tree. Used by the reverse incidence, type, canonical-key and symbol indexes. | diff --git a/docs/operations/configuration.md b/docs/operations/configuration.md index 1b4fadda..0dde46d0 100644 --- a/docs/operations/configuration.md +++ b/docs/operations/configuration.md @@ -92,6 +92,7 @@ max_connections = 200 | `durability` | `HSTORE_DURABILITY` | `sync` | `sync`: a commit is acknowledged after the group-commit barrier has `fsync`ed it. `async`: acknowledged once appended, so a crash can lose a suffix of acknowledged commits but never corrupts the database. | | `wal_mode` | `HSTORE_WAL_MODE` | `references` | Applies to the pages a commit writes itself: spilled pages of bulk loads and trees that changed in more than 4,096 places. `references`: the WAL logs `PageRef(pageId, address, length, crc32c)` records and those pages are synced before the WAL. `images`: the WAL logs their full images. Ordinary commits log only the keys they changed, and pages written at checkpoints are not logged in either mode. See [WAL and recovery](../transactions/wal-and-recovery.md). | | `node_cache_mb` | `HSTORE_NODE_CACHE_MB` | `256` | Megabytes of Java heap for decoded tree nodes, each counted at an estimate of the memory it takes. | +| `max_open_segment_files` | `HSTORE_MAX_OPEN_SEGMENT_FILES` | `256` | Segment files kept open at once. A segment file is opened when a page in it is written or read without its memory map, and once the limit is reached the least recently used file that no thread is using is closed to make room, after its unsynced writes are forced to disk. Must be between 1 and 2,147,483,647. | | `checkpoint_wal_mb` | `HSTORE_CHECKPOINT_WAL_MB` | `256` | WAL volume, or volume of pages written, since the last checkpoint that triggers a background checkpoint. | | `write_buffer_mb` | `HSTORE_WRITE_BUFFER_MB` | `64` | Megabytes of Java heap held by committed tree nodes that no checkpoint has written yet. A background checkpoint starts once more than half of it is in use; a commit that finds it full waits for a checkpoint, for at most 30 seconds. Must be between 1 and 16,777,216. | | `history_limit` | `HSTORE_HISTORY_LIMIT` | `64` | Committed generations kept addressable for `AT GENERATION`, `AS OF`, `HISTORY` and the Studio time slider. History is kept in memory only: after a restart it holds the checkpointed generation and the generations replayed from the WAL after it. | @@ -157,6 +158,13 @@ The value must be between 1 and 16,777,216 (16 TiB). Two older names are refused `cache_mb`, the name this setting had before it was released, and `cache_nodes`, a node count. Both messages name `node_cache_mb`. +**`max_open_segment_files`.** Each open segment file holds one file descriptor. Reads from sealed segments go +through a read-only memory map, and a map stays valid after its file is closed, so in steady state only the +segment being written and the ones written or remapped recently are open. Closing a file costs an `fsync` if it +has unsynced writes, and opening it again costs one `open` call. The write-ahead log, the change feed, +`pages.dir`, `pages.ref` and client connections need descriptors too, so keep the process limit (`ulimit -n`) +well above this value. + **`checkpoint_wal_mb`.** A checkpoint writes the tree nodes that commits left in memory, publishes the catalog image of the generation it wrote, and truncates WAL segments before the checkpoint LSN. Lower values shorten recovery, since recovery replays WAL after the last checkpoint, at the cost of more frequent catalog writes. Recovery time scales roughly linearly with WAL volume since the diff --git a/docs/storage/pages.md b/docs/storage/pages.md index dedcdd99..8095db59 100644 --- a/docs/storage/pages.md +++ b/docs/storage/pages.md @@ -281,7 +281,7 @@ sequenceDiagram * Allocation is serialized by a `ReentrantLock`, and only the counter and directory updates run under it. * Rolling a segment marks the previous one `SEALED`, creates the next id, forces the `data/segments` directory so the new file's entry is durable before any page in it is, and resets `nextUnit` to 0. * `PageStore.write(pageId, image)` looks the number up in the directory and performs a positional `FileChannel.write` at `offset × 64`, then marks the file dirty. It refuses an image longer than the units allocated for it. A checkpoint writes each image as soon as it has allocated it. For the pages a commit writes, the write happens later, inside `TransactionManager.append`, after the WAL records have been appended (see [wal-and-recovery.md](../transactions/wal-and-recovery.md)). -* `PageStore.sync()` calls `FileChannel.force(false)` on every dirty segment. The group committer calls it once per batch in `PAGE_REFERENCES` WAL mode, before the WAL sync, if a commit in the batch wrote pages. `PageStore.checkpoint()` does the same and then forces the page directory; only `Checkpointer` calls it. +* `PageStore.sync()` calls `FileChannel.force(false)` on every segment written since the last sync, and leaves every other segment alone, open or not. The group committer calls it once per batch in `PAGE_REFERENCES` WAL mode, before the WAL sync, if a commit in the batch wrote pages. `PageStore.checkpoint()` does the same and then forces the page directory; only `Checkpointer` calls it. Compared with fixed page slots, packing removes almost all internal fragmentation for the many small nodes a hypergraph produces. A member tree stored as a page is still an image of at least 128 bytes (two units), even for two members. That is why small member trees are not stored as pages at all, but inline in their edge record ([persistent-tree.md](persistent-tree.md#inline-trees)). @@ -326,6 +326,26 @@ ever appended to or deleted, never truncated, so a map can't end up pointing pas 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. +### Open files (`FilePool`) + +A store does not keep a file open per segment. `PageStore` owns a `FilePool` of at most `max_open_segment_files` +(`EngineOptions.maxOpenSegmentFiles`, default 256) open `FileHandle`s, and every write, `pread`, map and sync on a +segment borrows its handle from the pool for the length of that one call: + +* Opening the store only lists the segment files and reads their sizes with `stat`. A file is opened the first + time a call needs it, and rolling to a new segment opens (and creates) the new file straight away. +* A handle in use by a call is never closed. When the pool is full, the handle that has gone unused the longest is + closed to make room. If every handle is in use, the caller waits until one is returned. +* A file closed to make room is forced first if it has writes no sync has covered yet. A failed force puts the file + back in the unsynced set and fails the call that needed the room. +* Each write marks its file unsynced. `sync()` forces only the unsynced files and takes them out of the set. +* Deleting or closing a segment closes its handle once no call is using it. A deleted segment's entry refuses to + open again, so a late sync can't recreate the file. + +A memory map does not need its file to stay open, so a sealed segment that is mapped is read without a handle at +all. Its handle is closed like any other once it is the least recently used. The map is only replaced when the +segment grows past it, which takes a handle again for the `map` call. + ### Interrupted threads A `FileChannel` closes itself, for every thread using it, when a thread is interrupted during a read, write, sync or diff --git a/engine/src/main/java/io/hstore/engine/EngineOptions.java b/engine/src/main/java/io/hstore/engine/EngineOptions.java index 02443ce8..c309de64 100644 --- a/engine/src/main/java/io/hstore/engine/EngineOptions.java +++ b/engine/src/main/java/io/hstore/engine/EngineOptions.java @@ -15,6 +15,7 @@ public record EngineOptions( int pageSize, int pagesPerSegment, + int maxOpenSegmentFiles, long walSegmentBytes, long nodeCacheBytes, Durability durability, @@ -33,6 +34,9 @@ public record EngineOptions( if (Integer.bitCount(pageSize) != 1 || pageSize < 1024 || pageSize > 1 << 20) { throw new IllegalArgumentException("page size must be a power of two between 1 KiB and 1 MiB: " + pageSize); } + if (maxOpenSegmentFiles < 1) { + throw new IllegalArgumentException("at least one segment file must be allowed open: " + maxOpenSegmentFiles); + } if (compactionLiveRatio <= 0 || compactionLiveRatio >= 1) { throw new IllegalArgumentException("compaction live ratio must be between 0 and 1: " + compactionLiveRatio); } @@ -53,77 +57,82 @@ public record EngineOptions( } public static EngineOptions defaults() { - return new EngineOptions(16 * 1024, 16 * 1024, 64L << 20, 256L << 20, Durability.SYNC, WalMode.PAGE_REFERENCES, 64, + return new EngineOptions(16 * 1024, 16 * 1024, 256, 64L << 20, 256L << 20, Durability.SYNC, WalMode.PAGE_REFERENCES, 64, 256L << 20, 64L << 20, 0.5, 100_000, Duration.ofMillis(500), List.of(), List.of(), CrashPoint.Injector.NONE); } public EngineOptions withPageSize(int bytes) { - return new EngineOptions(bytes, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, + return new EngineOptions(bytes, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults); } public EngineOptions withWalSegmentBytes(long bytes) { - return new EngineOptions(pageSize, pagesPerSegment, bytes, nodeCacheBytes, durability, walMode, historyLimit, + return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, bytes, nodeCacheBytes, durability, walMode, historyLimit, checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults); } public EngineOptions withPagesPerSegment(int pages) { - return new EngineOptions(pageSize, pages, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, + return new EngineOptions(pageSize, pages, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, + checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults); + } + + public EngineOptions withMaxOpenSegmentFiles(int files) { + return new EngineOptions(pageSize, pagesPerSegment, files, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults); } public EngineOptions withDurability(Durability mode) { - return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, mode, walMode, historyLimit, + return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, mode, walMode, historyLimit, checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults); } public EngineOptions withWalMode(WalMode mode) { - return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, mode, historyLimit, + return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, mode, historyLimit, checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults); } public EngineOptions withHistoryLimit(int generations) { - return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, generations, + return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, generations, checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults); } public EngineOptions withCheckpointWalBytes(long bytes) { - return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, + return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, bytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults); } public EngineOptions withWriteBufferBytes(long bytes) { - return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, + return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, checkpointWalBytes, bytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults); } public EngineOptions withMaintenanceInterval(Duration interval) { - return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, + return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, interval, slots, derivations, faults); } public EngineOptions withNodeCacheBytes(long bytes) { - return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, bytes, durability, walMode, historyLimit, + return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, bytes, durability, walMode, historyLimit, checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, faults); } public EngineOptions withCompactionLiveRatio(double ratio) { - return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, + return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, checkpointWalBytes, writeBufferBytes, ratio, feedRetention, maintenanceInterval, slots, derivations, faults); } public EngineOptions withFeedRetention(long generations) { - return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, + return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, checkpointWalBytes, writeBufferBytes, compactionLiveRatio, generations, maintenanceInterval, slots, derivations, faults); } public EngineOptions withExtensions(List> extensionSlots, List extensionDerivations) { - return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, + return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, extensionSlots, extensionDerivations, faults); } public EngineOptions withFaults(CrashPoint.Injector injector) { - return new EngineOptions(pageSize, pagesPerSegment, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, + return new EngineOptions(pageSize, pagesPerSegment, maxOpenSegmentFiles, walSegmentBytes, nodeCacheBytes, durability, walMode, historyLimit, checkpointWalBytes, writeBufferBytes, compactionLiveRatio, feedRetention, maintenanceInterval, slots, derivations, injector); } } diff --git a/engine/src/main/java/io/hstore/engine/StorageEngine.java b/engine/src/main/java/io/hstore/engine/StorageEngine.java index 7554ca19..ea0120d6 100644 --- a/engine/src/main/java/io/hstore/engine/StorageEngine.java +++ b/engine/src/main/java/io/hstore/engine/StorageEngine.java @@ -82,7 +82,7 @@ private StorageEngine(Path directory, EngineOptions options, FileChannel lockCha CatalogStore catalog = CatalogStore.open(directory.resolve("catalog")); CatalogImage checkpoint = Recovery.discover(catalog); this.pages = PageStore.open(directory.resolve("data").resolve("segments"), options.pageSize(), options.pagesPerSegment(), - checkpoint.segments(), checkpoint.checkpointLsn()); + options.maxOpenSegmentFiles(), checkpoint.segments(), checkpoint.checkpointLsn()); this.wal = WriteAheadLog.open(directory.resolve("wal").resolve("segments"), options.walSegmentBytes()); this.slots = new SlotRegistry(options.slots()); FeedCodec codec = new FeedCodec(slots); diff --git a/engine/src/main/java/io/hstore/engine/page/FilePool.java b/engine/src/main/java/io/hstore/engine/page/FilePool.java new file mode 100644 index 00000000..beb56ed3 --- /dev/null +++ b/engine/src/main/java/io/hstore/engine/page/FilePool.java @@ -0,0 +1,255 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine.page; + +import io.hstore.engine.HStoreException; + +import java.io.IOException; +import java.nio.channels.ClosedChannelException; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Iterator; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.ReentrantLock; + +final class FilePool { + + private final int capacity; + private final ReentrantLock lock = new ReentrantLock(); + private final Condition released = lock.newCondition(); + private final LinkedHashSet idle = new LinkedHashSet<>(); + private final Set unsynced = ConcurrentHashMap.newKeySet(); + private final AtomicLong forces = new AtomicLong(); + private int open; + + FilePool(int capacity) { + if (capacity < 1) { + throw new IllegalArgumentException("the file pool must allow at least one open file: " + capacity); + } + this.capacity = capacity; + } + + Entry entry(Path path) { + return new Entry(path); + } + + int open() { + lock.lock(); + try { + return open; + } finally { + lock.unlock(); + } + } + + int capacity() { + return capacity; + } + + long forces() { + return forces.get(); + } + + int unsynced() { + return unsynced.size(); + } + + void sync() { + for (Entry entry : List.copyOf(unsynced)) { + try { + entry.force(); + } catch (IOException e) { + throw HStoreException.io("sync failed for " + entry.path(), e); + } + } + } + + private FileHandle acquire(Entry entry) throws IOException { + lock.lock(); + try { + while (true) { + if (entry.closed) { + throw new ClosedChannelException(); + } + if (entry.handle != null) { + if (entry.users++ == 0) { + idle.remove(entry); + } + return entry.handle; + } + if (open < capacity) { + entry.handle = openCreating(entry.path); + entry.users = 1; + open++; + return entry.handle; + } + Iterator oldest = idle.iterator(); + if (!oldest.hasNext()) { + released.awaitUninterruptibly(); + continue; + } + Entry victim = oldest.next(); + oldest.remove(); + FileHandle handle = victim.handle; + victim.handle = null; + lock.unlock(); + try { + victim.retire(handle); + } finally { + lock.lock(); + open--; + released.signalAll(); + } + } + } finally { + lock.unlock(); + } + } + + private static FileHandle openCreating(Path path) throws IOException { + boolean created = !Files.exists(path); + FileHandle handle = FileHandle.open(path); + if (created) { + FileHandle.syncDirectory(path.getParent()); + } + return handle; + } + + private void release(Entry entry) { + lock.lock(); + try { + if (--entry.users == 0) { + if (!entry.closed) { + idle.add(entry); + } + released.signalAll(); + } + } finally { + lock.unlock(); + } + } + + private void close(Entry entry) throws IOException { + FileHandle handle; + lock.lock(); + try { + entry.closed = true; + unsynced.remove(entry); + while (entry.users > 0) { + released.awaitUninterruptibly(); + } + idle.remove(entry); + handle = entry.handle; + entry.handle = null; + } finally { + lock.unlock(); + } + if (handle == null) { + return; + } + try { + handle.close(); + } finally { + lock.lock(); + try { + open--; + released.signalAll(); + } finally { + lock.unlock(); + } + } + } + + final class Entry { + + private final Path path; + private final AtomicLong written = new AtomicLong(); + private final AtomicLong durable = new AtomicLong(); + private FileHandle handle; + private int users; + private volatile boolean closed; + + private Entry(Path path) { + this.path = path; + } + + Path path() { + return path; + } + + T call(FileHandle.Io io) throws IOException { + FileHandle leased = acquire(this); + try { + return leased.call(io); + } finally { + release(this); + } + } + + void run(FileHandle.Action action) throws IOException { + call(channel -> { + action.run(channel); + return null; + }); + } + + void write(FileHandle.Action action) throws IOException { + call(channel -> { + action.run(channel); + written.incrementAndGet(); + unsynced.add(this); + return null; + }); + } + + void force() throws IOException { + unsynced.remove(this); + long target = written.get(); + if (durable.get() >= target) { + return; + } + try { + run(channel -> force(channel, target)); + } catch (IOException e) { + if (closed) { + return; + } + unsynced.add(this); + throw e; + } + } + + void close() throws IOException { + FilePool.this.close(this); + } + + private void force(FileChannel channel, long target) throws IOException { + if (durable.get() >= target) { + return; + } + channel.force(false); + forces.incrementAndGet(); + durable.accumulateAndGet(target, Math::max); + } + + private void retire(FileHandle handle) throws IOException { + long target = written.get(); + try { + handle.run(channel -> force(channel, target)); + } finally { + handle.close(); + } + unsynced.remove(this); + if (durable.get() < written.get()) { + unsynced.add(this); + } + } + } +} 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 1ea4cf1a..a1526d36 100644 --- a/engine/src/main/java/io/hstore/engine/page/PageStore.java +++ b/engine/src/main/java/io/hstore/engine/page/PageStore.java @@ -32,6 +32,7 @@ public record Scan(long[] dead, long[] unreferenced, long unplaced) { private final Path directory; private final int pageSize; private final int segmentUnits; + private final FilePool pool; private final Map files = new ConcurrentHashMap<>(); private final ConcurrentSkipListMap segments = new ConcurrentSkipListMap<>(); private final Map live = new ConcurrentHashMap<>(); @@ -48,18 +49,20 @@ public record Scan(long[] dead, long[] unreferenced, long unplaced) { private int activeSegment; private int nextUnit; - private PageStore(Path directory, int pageSize, int pagesPerSegment) { + private PageStore(Path directory, int pageSize, int pagesPerSegment, int openFiles) { this.directory = directory; this.pageSize = pageSize; this.segmentUnits = (int) Math.min(PageAddress.MAX_OFFSET + 1L, (long) pagesPerSegment * pageSize / PageId.UNIT_BYTES); + this.pool = new FilePool(openFiles); } - public static PageStore open(Path directory, int pageSize, int pagesPerSegment, Collection known) { - return open(directory, pageSize, pagesPerSegment, known, -1); + public static PageStore open(Path directory, int pageSize, int pagesPerSegment, int openFiles, Collection known) { + return open(directory, pageSize, pagesPerSegment, openFiles, known, -1); } - public static PageStore open(Path directory, int pageSize, int pagesPerSegment, Collection known, long checkpointLsn) { - PageStore store = new PageStore(directory, pageSize, pagesPerSegment); + public static PageStore open(Path directory, int pageSize, int pagesPerSegment, int openFiles, Collection known, + long checkpointLsn) { + PageStore store = new PageStore(directory, pageSize, pagesPerSegment, openFiles); store.restore(known, checkpointLsn); return store; } @@ -80,7 +83,7 @@ private void restore(Collection known, long checkpointLsn) { try (Stream listing = Files.list(directory)) { for (Path file : listing.filter(p -> p.getFileName().toString().endsWith(".seg")).toList()) { int id = SegmentFile.idOf(file); - SegmentFile segment = SegmentFile.open(id, file); + SegmentFile segment = SegmentFile.open(id, file, pool); files.put(id, segment); long units = segment.unitsOnDisk(); SegmentInfo info = byId.getOrDefault(id, new SegmentInfo(id, SegmentState.SEALED, units, units, 0)); @@ -251,7 +254,7 @@ private void rollSegment() { } } int id = segments.isEmpty() ? 1 : segments.lastKey() + 1; - files.put(id, SegmentFile.open(id, SegmentFile.pathFor(directory, id))); + files.put(id, SegmentFile.create(id, SegmentFile.pathFor(directory, id), pool)); segments.put(id, new SegmentInfo(id, SegmentState.ACTIVE, 0, 0, 0)); activeSegment = id; nextUnit = 0; @@ -286,7 +289,7 @@ public void writeUnsettled(long pageId, MemorySegment image) { int offset = PageAddress.offsetOf(address); SegmentFile file = files.computeIfAbsent(segment, id -> { segments.putIfAbsent(id, new SegmentInfo(id, SegmentState.SEALED, 0, 0, 0)); - return SegmentFile.open(id, SegmentFile.pathFor(directory, id)); + return SegmentFile.open(id, SegmentFile.pathFor(directory, id), pool); }); file.write((long) offset * PageId.UNIT_BYTES, image); long end = offset + (long) PageId.unitsFor(image.byteSize()); @@ -413,7 +416,19 @@ public Map liveBytes() { } public void sync() { - files.values().forEach(SegmentFile::force); + pool.sync(); + } + + int openFiles() { + return pool.open(); + } + + int unsyncedFiles() { + return pool.unsynced(); + } + + long segmentForces() { + return pool.forces(); } public void checkpoint() { 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 c37be92c..ea9a073e 100644 --- a/engine/src/main/java/io/hstore/engine/page/SegmentFile.java +++ b/engine/src/main/java/io/hstore/engine/page/SegmentFile.java @@ -14,7 +14,6 @@ import java.nio.channels.FileChannel; import java.nio.file.Files; import java.nio.file.Path; -import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.ReentrantLock; final class SegmentFile implements AutoCloseable { @@ -24,30 +23,30 @@ final class SegmentFile implements AutoCloseable { private final int id; private final Path path; - private final FileHandle file; - private final AtomicBoolean dirty = new AtomicBoolean(); + private final FilePool.Entry file; 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, FileHandle file) { + private SegmentFile(int id, Path path, FilePool.Entry file) { this.id = id; this.path = path; this.file = file; } - static SegmentFile open(int id, Path path) { + static SegmentFile open(int id, Path path, FilePool pool) { + return new SegmentFile(id, path, pool.entry(path)); + } + + static SegmentFile create(int id, Path path, FilePool pool) { + SegmentFile segment = open(id, path, pool); try { - boolean created = !Files.exists(path); - SegmentFile segment = new SegmentFile(id, path, FileHandle.open(path)); - if (created) { - FileHandle.syncDirectory(path.getParent()); - } - return segment; + segment.file.run(_ -> { }); } catch (IOException e) { - throw new UncheckedIOException("cannot open segment " + path, e); + throw new UncheckedIOException("cannot create segment " + path, e); } + return segment; } static Path pathFor(Path directory, int id) { @@ -65,7 +64,7 @@ int id() { long unitsOnDisk() { try { - return PageId.unitsFor(file.size()); + return PageId.unitsFor(Files.size(path)); } catch (IOException e) { throw HStoreException.io("cannot size segment " + path, e); } @@ -107,7 +106,7 @@ private MemorySegment read(MemorySegment view, long position, int maximum) { MemorySegment view() { try { - MemorySegment whole = covering(file.size()); + MemorySegment whole = covering(Files.size(path)); return whole == null ? MemorySegment.NULL : whole; } catch (IOException e) { throw HStoreException.io("cannot size segment " + path, e); @@ -138,7 +137,7 @@ private MemorySegment covering(long required) { if (required <= current.byteSize()) { return current; } - long size = file.size(); + long size = Files.size(path); if (size < required) { return null; } @@ -195,28 +194,16 @@ private void fill(ByteBuffer buffer, long position) { void write(long position, MemorySegment image) { ByteBuffer buffer = image.asByteBuffer(); try { - file.run(channel -> { + file.write(channel -> { while (buffer.hasRemaining()) { channel.write(buffer, position + buffer.position()); } }); - dirty.set(true); } catch (IOException e) { throw HStoreException.io("write failed in segment " + id + " at byte " + position, e); } } - void force() { - if (dirty.getAndSet(false)) { - try { - file.run(channel -> channel.force(false)); - } catch (IOException e) { - dirty.set(true); - throw HStoreException.io("sync failed for segment " + id, e); - } - } - } - void delete() { close(); try { diff --git a/engine/src/test/java/io/hstore/engine/page/FilePoolTest.java b/engine/src/test/java/io/hstore/engine/page/FilePoolTest.java new file mode 100644 index 00000000..52d1b981 --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/page/FilePoolTest.java @@ -0,0 +1,156 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine.page; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.ByteBuffer; +import java.nio.channels.ClosedChannelException; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +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.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class FilePoolTest { + + @TempDir + Path directory; + + @Test + void filesOpenOnDemandAndTheLeastRecentlyUsedIsClosedFirst() throws Exception { + FilePool pool = new FilePool(2); + FilePool.Entry a = pool.entry(directory.resolve("a")); + FilePool.Entry b = pool.entry(directory.resolve("b")); + FilePool.Entry c = pool.entry(directory.resolve("c")); + assertEquals(0, pool.open()); + List seen = new ArrayList<>(); + a.run(seen::add); + b.run(seen::add); + a.run(channel -> assertSame(seen.getFirst(), channel)); + assertEquals(2, pool.open()); + c.run(seen::add); + assertEquals(2, pool.open()); + assertTrue(seen.get(0).isOpen(), "a was used more recently than b"); + assertFalse(seen.get(1).isOpen(), "b was the least recently used"); + b.run(seen::add); + assertFalse(seen.get(0).isOpen(), "a is now the least recently used"); + assertTrue(seen.get(2).isOpen()); + assertEquals(2, pool.open()); + } + + @Test + void aFileIsForcedBeforeItIsClosedToMakeRoom() throws Exception { + FilePool pool = new FilePool(1); + FilePool.Entry a = pool.entry(directory.resolve("a")); + FilePool.Entry b = pool.entry(directory.resolve("b")); + a.write(channel -> channel.write(ByteBuffer.wrap(new byte[]{1, 2, 3}), 0)); + assertEquals(1, pool.unsynced()); + assertEquals(0, pool.forces()); + b.run(channel -> channel.read(ByteBuffer.allocate(1), 0)); + assertEquals(1, pool.forces()); + assertEquals(0, pool.unsynced()); + pool.sync(); + assertEquals(1, pool.forces()); + assertEquals(3, Files.size(directory.resolve("a"))); + } + + @Test + void syncForcesOnlyTheFilesWrittenSinceTheLastSync() throws Exception { + FilePool pool = new FilePool(64); + List entries = new ArrayList<>(); + for (int i = 0; i < 20; i++) { + FilePool.Entry entry = pool.entry(directory.resolve("f" + i)); + entry.write(channel -> channel.write(ByteBuffer.wrap(new byte[]{7}), 0)); + entries.add(entry); + } + pool.sync(); + assertEquals(20, pool.forces()); + pool.sync(); + assertEquals(20, pool.forces()); + entries.get(3).write(channel -> channel.write(ByteBuffer.wrap(new byte[]{8}), 1)); + entries.get(3).run(channel -> channel.read(ByteBuffer.allocate(2), 0)); + entries.get(9).run(channel -> channel.read(ByteBuffer.allocate(1), 0)); + pool.sync(); + assertEquals(21, pool.forces()); + } + + @Test + void aFileInUseIsNeverClosedAndTheBoundHoldsUnderContention() throws Exception { + int capacity = 3; + FilePool pool = new FilePool(capacity); + List entries = new ArrayList<>(); + for (int i = 0; i < 16; i++) { + FilePool.Entry entry = pool.entry(directory.resolve("f" + i)); + int value = i; + entry.write(channel -> channel.write(ByteBuffer.allocate(4096).putInt(0, value), 0)); + entries.add(entry); + } + AtomicInteger closedUnderneath = new AtomicInteger(); + AtomicInteger overBound = new AtomicInteger(); + try (ExecutorService threads = Executors.newFixedThreadPool(8)) { + List> results = new ArrayList<>(); + for (int t = 0; t < 8; t++) { + Random random = new Random(t); + results.add(threads.submit(() -> { + int calls = 0; + for (int i = 0; i < 2_000; i++) { + int pick = random.nextInt(entries.size()); + FilePool.Entry entry = entries.get(pick); + int read = entry.call(channel -> { + ByteBuffer buffer = ByteBuffer.allocate(4); + channel.read(buffer, 0); + Thread.yield(); + if (!channel.isOpen()) { + closedUnderneath.incrementAndGet(); + } + if (pool.open() > capacity) { + overBound.incrementAndGet(); + } + if (random.nextInt(4) == 0) { + channel.write(ByteBuffer.allocate(4).putInt(0, pick), 0); + } + return buffer.getInt(0); + }); + assertEquals(pick, read); + calls++; + } + return calls; + })); + } + for (Future result : results) { + assertEquals(2_000, result.get()); + } + } + assertEquals(0, closedUnderneath.get()); + assertEquals(0, overBound.get()); + assertTrue(pool.open() <= capacity); + } + + @Test + void aClosedEntryIsNotReopenedOrRecreated() throws Exception { + FilePool pool = new FilePool(4); + Path path = directory.resolve("gone"); + FilePool.Entry entry = pool.entry(path); + entry.write(channel -> channel.write(ByteBuffer.wrap(new byte[]{1}), 0)); + entry.close(); + Files.delete(path); + assertEquals(0, pool.open()); + assertThrows(ClosedChannelException.class, () -> entry.run(channel -> channel.read(ByteBuffer.allocate(1), 0))); + pool.sync(); + assertFalse(Files.exists(path)); + } +} diff --git a/engine/src/test/java/io/hstore/engine/page/PageStoreTest.java b/engine/src/test/java/io/hstore/engine/page/PageStoreTest.java index 15c5105b..36131c33 100644 --- a/engine/src/test/java/io/hstore/engine/page/PageStoreTest.java +++ b/engine/src/test/java/io/hstore/engine/page/PageStoreTest.java @@ -8,8 +8,10 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import java.io.IOException; import java.lang.foreign.MemorySegment; import java.lang.foreign.ValueLayout; +import java.nio.file.Files; import java.nio.file.Path; import java.util.ArrayList; import java.util.LinkedHashMap; @@ -19,7 +21,9 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Stream; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -27,11 +31,14 @@ import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; class PageStoreTest { private static final int PAGE_SIZE = 4096; private static final int PAGES_PER_SEGMENT = 2048; + private static final int OPEN_FILES = 256; + private static final Path DESCRIPTORS = Path.of("/dev/fd"); private static final int PAYLOAD_LENGTH_OFFSET = 24; @TempDir @@ -43,7 +50,7 @@ void pagesReadBackExactlyAsWrittenAcrossSegmentsAndReopens() { Map written = new LinkedHashMap<>(); List ids = new ArrayList<>(); List segments; - try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) { + try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, OPEN_FILES, 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(image.length); @@ -59,7 +66,7 @@ void pagesReadBackExactlyAsWrittenAcrossSegmentsAndReopens() { 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)) { + try (PageStore reopened = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, OPEN_FILES, segments)) { written.forEach((pageId, image) -> assertArrayEquals(image, bytes(reopened.read(pageId)))); long first = ids.getFirst(); reopened.read(first); @@ -69,7 +76,7 @@ void pagesReadBackExactlyAsWrittenAcrossSegmentsAndReopens() { @Test void pagesCannotBeChangedThroughWhatReadReturns() { - try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) { + try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, OPEN_FILES, List.of())) { long first = write(store, PAGE_SIZE); fillPastOneSegment(store); long last = write(store, 200); @@ -82,7 +89,7 @@ void pagesCannotBeChangedThroughWhatReadReturns() { @Test void finishedSegmentsAreMappedAndTheGrowingOneIsNotUntilItGrows() { - try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) { + try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, OPEN_FILES, List.of())) { long finished = write(store, PAGE_SIZE); fillPastOneSegment(store); long growing = write(store, PAGE_SIZE); @@ -101,7 +108,7 @@ void finishedSegmentsAreMappedAndTheGrowingOneIsNotUntilItGrows() { @Test void anInterruptedThreadReadsAndWritesPagesWithoutClosingTheSegment() throws Exception { - try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) { + try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, OPEN_FILES, List.of())) { long before = write(store, PAGE_SIZE); byte[] expected = bytes(store.read(before)); long[] written = new long[1]; @@ -128,7 +135,7 @@ void anInterruptedThreadReadsAndWritesPagesWithoutClosingTheSegment() throws Exc @Test void pagesAreNumberedInOrderWhicheverSegmentHoldsThem() { - try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) { + try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, OPEN_FILES, List.of())) { List ids = new ArrayList<>(); while (store.segments().size() < 3) { ids.add(write(store, PAGE_SIZE / 2)); @@ -142,7 +149,7 @@ void pagesAreNumberedInOrderWhicheverSegmentHoldsThem() { @Test void aNumberWithoutAnAddressIsReportedAsCorrupt() { - try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) { + try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, OPEN_FILES, List.of())) { write(store, 200); assertThrows(HStoreException.CorruptPage.class, () -> store.read(PageId.pack(99, 0))); } @@ -150,7 +157,7 @@ void aNumberWithoutAnAddressIsReportedAsCorrupt() { @Test void aPageCannotBeWrittenPastWhatWasAllocatedForIt() { - try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) { + try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, OPEN_FILES, List.of())) { long pageId = store.allocate(100); assertThrows(IllegalArgumentException.class, () -> store.write(pageId, MemorySegment.ofArray(image(new Random(1), 200)))); } @@ -158,7 +165,7 @@ void aPageCannotBeWrittenPastWhatWasAllocatedForIt() { @Test void aSegmentIsNotSettledWhileAPageAllocatedInItIsUnwritten() { - try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) { + try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, OPEN_FILES, List.of())) { long pending = store.allocate(PAGE_SIZE); int segment = store.segmentOf(pending); fillPastOneSegment(store); @@ -170,7 +177,7 @@ void aSegmentIsNotSettledWhileAPageAllocatedInItIsUnwritten() { @Test void evacuatingASegmentMovesOnlyThePagesTheDirectoryStillPointsAt() { - try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) { + try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, OPEN_FILES, List.of())) { Map kept = new LinkedHashMap<>(); List released = new ArrayList<>(); int segment = store.activeSegment() == 0 ? 1 : store.activeSegment(); @@ -218,7 +225,7 @@ void readersKeepSeeingWholePagesWhileOldMapsAreClosed() throws Exception { 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()); + try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, OPEN_FILES, List.of()); ExecutorService readers = Executors.newFixedThreadPool(4)) { List> checks = new ArrayList<>(); for (int t = 0; t < 4; t++) { @@ -249,6 +256,117 @@ void readersKeepSeeingWholePagesWhileOldMapsAreClosed() throws Exception { } } + @Test + void segmentFilesOpenOnDemandAndStayWithinThePool() throws Exception { + assumeTrue(Files.isDirectory(DESCRIPTORS)); + int pool = 4; + Random random = new Random(5); + Map written = new LinkedHashMap<>(); + List ids = new ArrayList<>(); + List segments; + long before = openDescriptors(); + try (PageStore store = PageStore.open(directory, PAGE_SIZE, 16, pool, List.of())) { + for (int i = 0; i < 2_000; i++) { + byte[] image = image(random, PAGE_SIZE - random.nextInt(64)); + long pageId = store.allocate(image.length); + store.write(pageId, MemorySegment.ofArray(image)); + written.put(pageId, image); + ids.add(pageId); + long earlier = ids.get(random.nextInt(ids.size())); + assertArrayEquals(written.get(earlier), bytes(store.read(earlier))); + assertTrue(store.openFiles() <= pool, store.openFiles() + " segment files open"); + } + store.sync(); + segments = store.segments(); + assertTrue(segments.size() > 100, segments.size() + " segments"); + assertTrue(openDescriptors() - before <= pool + 2, (openDescriptors() - before) + " more descriptors than before"); + } + before = openDescriptors(); + try (PageStore reopened = PageStore.open(directory, PAGE_SIZE, 16, pool, segments)) { + assertEquals(0, reopened.openFiles()); + written.forEach((pageId, image) -> assertArrayEquals(image, bytes(reopened.read(pageId)))); + assertTrue(reopened.openFiles() <= pool); + assertTrue(openDescriptors() - before <= pool + 2, (openDescriptors() - before) + " more descriptors than before"); + } + } + + @Test + void aSegmentClosedToMakeRoomIsForcedFirstAndSyncSkipsIt() { + try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, 1, List.of())) { + long first = write(store, PAGE_SIZE); + assertEquals(1, store.unsyncedFiles()); + long pending = store.allocate(PAGE_SIZE); + fillPastOneSegment(store); + store.sync(); + long forced = store.segmentForces(); + store.write(pending, MemorySegment.ofArray(image(new Random(PAGE_SIZE), PAGE_SIZE))); + assertEquals(store.segmentOf(first), store.segmentOf(pending)); + long next = write(store, PAGE_SIZE); + assertNotEquals(store.segmentOf(pending), store.segmentOf(next)); + assertEquals(forced + 1, store.segmentForces(), "the earlier segment was forced when the active one took its place"); + assertEquals(1, store.unsyncedFiles()); + store.sync(); + assertEquals(forced + 2, store.segmentForces()); + store.sync(); + assertEquals(forced + 2, store.segmentForces()); + assertArrayEquals(bytes(store.read(first)), bytes(store.read(pending))); + } + } + + @Test + void readersAndWritersShareASmallPoolWithoutSeeingAClosedFile() throws Exception { + int pages = 3_000; + long[] ids = new long[pages]; + byte[][] images = new byte[pages][]; + AtomicInteger written = new AtomicInteger(); + AtomicBoolean done = new AtomicBoolean(); + try (PageStore store = PageStore.open(directory, PAGE_SIZE, 64, 2, 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 (!done.get()) { + int known = written.get(); + if (known > 0) { + int i = known - 1 - random.nextInt(Math.min(known, 32)); + assertArrayEquals(images[i], bytes(store.read(ids[i])), "page " + i); + if (++reads % 64 == 0) { + store.sync(); + } + } + } + return reads; + })); + } + Random random = new Random(11); + try { + for (int i = 0; i < pages; i++) { + images[i] = image(random, PAGE_SIZE - random.nextInt(64)); + ids[i] = store.allocate(images[i].length); + store.write(ids[i], MemorySegment.ofArray(images[i])); + written.incrementAndGet(); + assertTrue(store.openFiles() <= 2, store.openFiles() + " segment files open"); + } + } finally { + done.set(true); + } + for (Future check : checks) { + assertTrue(check.get() > 0); + } + for (int i = 0; i < pages; i++) { + assertArrayEquals(images[i], bytes(store.read(ids[i]))); + } + } + } + + private static long openDescriptors() throws IOException { + try (Stream listing = Files.list(DESCRIPTORS)) { + return listing.count(); + } + } + private static long write(PageStore store, int length) { byte[] image = image(new Random(length), length); long pageId = store.allocate(image.length); diff --git a/server/src/main/java/io/hstore/server/ServerConfig.java b/server/src/main/java/io/hstore/server/ServerConfig.java index 2014a77a..66b65ada 100644 --- a/server/src/main/java/io/hstore/server/ServerConfig.java +++ b/server/src/main/java/io/hstore/server/ServerConfig.java @@ -107,6 +107,14 @@ long megabytes(Setting setting) { return megabytes << 20; } + int positive(Setting setting) { + long value = number(setting); + if (value < 1 || value > Integer.MAX_VALUE) { + throw new IllegalArgumentException("%s must be between 1 and %,d, not %d".formatted(setting.key(), Integer.MAX_VALUE, value)); + } + return (int) value; + } + boolean flag(Setting setting) { return switch (values.get(setting).toLowerCase()) { case "on", "true", "yes", "1" -> true; @@ -125,6 +133,7 @@ EngineOptions engineOptions() { .withDurability(Durability.valueOf(string(Setting.DURABILITY).toUpperCase())) .withWalMode(string(Setting.WAL_MODE).equalsIgnoreCase("images") ? WalMode.PAGE_IMAGES : WalMode.PAGE_REFERENCES) .withNodeCacheBytes(megabytes(Setting.NODE_CACHE_MB)) + .withMaxOpenSegmentFiles(positive(Setting.MAX_OPEN_SEGMENT_FILES)) .withCheckpointWalBytes(number(Setting.CHECKPOINT_WAL_MB) << 20) .withWriteBufferBytes(megabytes(Setting.WRITE_BUFFER_MB)) .withHistoryLimit(integer(Setting.HISTORY_LIMIT)) diff --git a/server/src/main/java/io/hstore/server/Setting.java b/server/src/main/java/io/hstore/server/Setting.java index 0dd84230..0104cf63 100644 --- a/server/src/main/java/io/hstore/server/Setting.java +++ b/server/src/main/java/io/hstore/server/Setting.java @@ -25,6 +25,7 @@ enum Setting { DURABILITY("durability", "sync", "sync waits for fsync before acknowledging commits; async does not"), WAL_MODE("wal_mode", "references", "references logs page checksums and syncs data first; images logs full page images"), NODE_CACHE_MB("node_cache_mb", "256", "megabytes of heap for decoded tree nodes, each counted at an estimate of its size"), + MAX_OPEN_SEGMENT_FILES("max_open_segment_files", "256", "segment files kept open at once; opening another closes the least recently used"), CHECKPOINT_WAL_MB("checkpoint_wal_mb", "256", "write-ahead log volume that triggers a background checkpoint"), WRITE_BUFFER_MB("write_buffer_mb", "64", "megabytes of committed tree nodes kept in memory until a checkpoint writes them"), HISTORY_LIMIT("history_limit", "64", "committed generations retained for time travel"), diff --git a/server/src/test/java/io/hstore/server/ServerTest.java b/server/src/test/java/io/hstore/server/ServerTest.java index a2df9480..c6b8f1f8 100644 --- a/server/src/test/java/io/hstore/server/ServerTest.java +++ b/server/src/test/java/io/hstore/server/ServerTest.java @@ -164,6 +164,22 @@ void nodeCacheIsSizedInMegabytesAndOldNamesAreRejected(@TempDir Path data) throw failures.forEach(failure -> assertTrue(failure.getMessage().contains("node_cache_mb"), failure.getMessage())); } + @Test + void openSegmentFilesComeFromTheFileTheEnvironmentOrAFlag(@TempDir Path data) throws Exception { + assertEquals(256, ServerConfig.load(Optional.empty(), Map.of(), Map.of()).engineOptions().maxOpenSegmentFiles()); + Files.writeString(data.resolve(ServerConfig.FILE), "max_open_segment_files = 64\n"); + assertEquals(64, ServerConfig.load(Optional.of(data), Map.of(), Map.of()).engineOptions().maxOpenSegmentFiles()); + assertEquals(32, ServerConfig.load(Optional.of(data), Map.of("HSTORE_MAX_OPEN_SEGMENT_FILES", "32"), Map.of()) + .engineOptions().maxOpenSegmentFiles()); + assertEquals(16, ServerConfig.load(Optional.of(data), Map.of("HSTORE_MAX_OPEN_SEGMENT_FILES", "32"), + Map.of("max-open-segment-files", "16")).engineOptions().maxOpenSegmentFiles()); + for (String outOfRange : List.of("0", "-3", "4294967296")) { + IllegalArgumentException failure = assertThrows(IllegalArgumentException.class, + () -> ServerConfig.load(Optional.empty(), Map.of("HSTORE_MAX_OPEN_SEGMENT_FILES", outOfRange), Map.of()).engineOptions()); + assertTrue(failure.getMessage().contains("max_open_segment_files"), failure.getMessage()); + } + } + @Test void tlsWithoutACertificateIsRejectedAtStartup() { ServerConfig config = ServerConfig.load(Optional.empty(), Map.of(), Map.of("tls", "on"));