diff --git a/docs/storage/pages.md b/docs/storage/pages.md index 0aa2dc1..b239157 100644 --- a/docs/storage/pages.md +++ b/docs/storage/pages.md @@ -317,6 +317,25 @@ 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. +### Interrupted threads + +A `FileChannel` closes itself, for every thread using it, when a thread is interrupted during a read, write, sync or +map on it. Commits, reads and feed subscribers run on the caller's threads, and callers do interrupt them, for +example when a request times out. Before this was handled, one interrupted commit closed the write-ahead log and +every later commit failed until the process restarted. + +Segment files, the active write-ahead log segment, the change feed's segments, `pages.dir` and `pages.ref` therefore +go through `page/FileHandle`: + +* It clears the caller's interrupt flag for the length of the call and sets it again afterwards, so the caller still + sees its interrupt. +* If the channel was closed anyway (an interrupt that arrived in the middle of the call, or another thread's + interrupt), it reopens the file and runs the call again. Every read and write is positional, so running it again + reads or writes the same bytes. + +A checkpoint and `close()` also run with the flag cleared, because they write short-lived files such as the catalog +and the reference journal. + ## Caching and I/O accounting `PagedNodeSource.load(pageId, schema)` (`tree/PagedNodeSource.java`): diff --git a/engine/src/main/java/io/hstore/engine/StorageEngine.java b/engine/src/main/java/io/hstore/engine/StorageEngine.java index 261d443..964fdcc 100644 --- a/engine/src/main/java/io/hstore/engine/StorageEngine.java +++ b/engine/src/main/java/io/hstore/engine/StorageEngine.java @@ -396,24 +396,31 @@ public void halt() { } private void shutdown() { - closed = true; - transactions.shutdown(); - references.close(); - maintainer.interrupt(); + boolean interrupted = Thread.interrupted(); try { - maintainer.join(Duration.ofSeconds(2)); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - feed.close(); - wal.close(); - pages.close(); - cache.clear(); - try { - lock.release(); - lockChannel.close(); - } catch (IOException e) { - throw new UncheckedIOException(e); + closed = true; + transactions.shutdown(); + references.close(); + maintainer.interrupt(); + try { + maintainer.join(Duration.ofSeconds(2)); + } catch (InterruptedException e) { + interrupted = true; + } + feed.close(); + wal.close(); + pages.close(); + cache.clear(); + try { + lock.release(); + lockChannel.close(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } finally { + if (interrupted) { + Thread.currentThread().interrupt(); + } } } } diff --git a/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java b/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java index fd7f30c..0782bb2 100644 --- a/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java +++ b/engine/src/main/java/io/hstore/engine/feed/ChangeFeed.java @@ -6,6 +6,7 @@ import io.hstore.engine.HStoreException; import io.hstore.engine.page.ByteCursor; import io.hstore.engine.page.Checksums; +import io.hstore.engine.page.FileHandle; import java.io.IOException; import java.io.UncheckedIOException; @@ -62,13 +63,13 @@ public interface Gap { private static final class Segment { final long start; final Path path; - final FileChannel channel; + final FileHandle file; volatile long firstGeneration = EMPTY; - Segment(long start, Path path, FileChannel channel) { + Segment(long start, Path path, FileHandle file) { this.start = start; this.path = path; - this.channel = channel; + this.file = file; } } @@ -111,7 +112,7 @@ private void restore() throws IOException { try (Stream listing = Files.list(directory)) { for (Path path : listing.filter(file -> file.getFileName().toString().endsWith(".feed")).toList()) { long start = Long.parseUnsignedLong(path.getFileName().toString().replace(".feed", ""), 16); - segments.put(start, new Segment(start, path, FileChannel.open(path, StandardOpenOption.READ, StandardOpenOption.WRITE))); + segments.put(start, new Segment(start, path, FileHandle.open(path))); } } if (segments.isEmpty()) { @@ -119,7 +120,7 @@ private void restore() throws IOException { } Segment last = segments.lastEntry().getValue(); long position = segments.firstKey(); - for (Iterator frames = new FrameReader(position, last.start + last.channel.size()); frames.hasNext(); ) { + for (Iterator frames = new FrameReader(position, last.start + last.file.size()); frames.hasNext(); ) { Frame frame = frames.next(); remember(frame.generation(), position); position = frame.next(); @@ -146,7 +147,8 @@ private void cutAt(long position) throws IOException { segments.remove(later.start); discard(later); } - holding.channel.truncate(position - holding.start); + long length = position - holding.start; + holding.file.run(channel -> channel.truncate(length)); if (position == holding.start) { holding.firstGeneration = EMPTY; } @@ -185,12 +187,14 @@ public void append(CommitEvent event) { try { Segment active = segments.isEmpty() || end - segments.lastKey() >= segmentBytes ? roll() : segments.lastEntry().getValue(); ByteBuffer buffer = frame.segment().asByteBuffer(); - long position = end; - while (buffer.hasRemaining()) { - position += active.channel.write(buffer, position - active.start); - } + long local = end - active.start; + active.file.run(channel -> { + while (buffer.hasRemaining()) { + channel.write(buffer, local + buffer.position()); + } + }); remember(event.generation(), end); - end = position; + end += buffer.limit(); appended.signalAll(); } catch (IOException e) { throw HStoreException.io("change feed append failed", e); @@ -201,10 +205,10 @@ public void append(CommitEvent event) { private Segment roll() throws IOException { if (!segments.isEmpty()) { - segments.lastEntry().getValue().channel.force(false); + segments.lastEntry().getValue().file.run(channel -> channel.force(false)); } Path path = directory.resolve("%016x.feed".formatted(end)); - Segment segment = new Segment(end, path, FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE)); + Segment segment = new Segment(end, path, FileHandle.open(path)); segments.put(end, segment); try (FileChannel listing = FileChannel.open(directory, StandardOpenOption.READ)) { listing.force(true); @@ -219,7 +223,7 @@ public void sync() { return; } try { - segments.lastEntry().getValue().channel.force(false); + segments.lastEntry().getValue().file.run(channel -> channel.force(false)); } catch (IOException e) { throw HStoreException.io("change feed sync failed", e); } @@ -262,7 +266,7 @@ public long retain(long keepAfter) { private void discard(Segment segment) { files.writeLock().lock(); try { - segment.channel.close(); + segment.file.close(); Files.deleteIfExists(segment.path); } catch (IOException e) { throw new UncheckedIOException(e); @@ -321,8 +325,8 @@ public void close() { try { appended.signalAll(); for (Segment segment : segments.values()) { - segment.channel.force(true); - segment.channel.close(); + segment.file.run(channel -> channel.force(true)); + segment.file.close(); } } catch (IOException e) { throw new UncheckedIOException(e); @@ -374,14 +378,14 @@ private Frame read() { private Frame readLocked() { Map.Entry entry = segments.floorEntry(position); - if (entry == null || !entry.getValue().channel.isOpen()) { + if (entry == null || !entry.getValue().file.isOpen()) { return null; } - FileChannel channel = entry.getValue().channel; + FileHandle file = entry.getValue().file; long local = position - entry.getKey(); try { ByteBuffer header = ByteBuffer.allocate(FRAME_HEADER); - if (!readFully(channel, header, local)) { + if (!readFully(file, header, local)) { return null; } ByteCursor in = ByteCursor.over(MemorySegment.ofBuffer(header.flip())); @@ -392,7 +396,7 @@ private Frame readLocked() { return null; } ByteBuffer body = ByteBuffer.allocate(length); - if (!readFully(channel, body, local + FRAME_HEADER)) { + if (!readFully(file, body, local + FRAME_HEADER)) { return null; } byte[] payload = body.array(); @@ -404,13 +408,15 @@ private Frame readLocked() { } } - private boolean readFully(FileChannel channel, ByteBuffer buffer, long at) throws IOException { - while (buffer.hasRemaining()) { - if (channel.read(buffer, at + buffer.position()) < 0) { - return false; + private boolean readFully(FileHandle file, ByteBuffer buffer, long at) throws IOException { + return file.call(channel -> { + while (buffer.hasRemaining()) { + if (channel.read(buffer, at + buffer.position()) < 0) { + return false; + } } - } - return true; + return true; + }); } } diff --git a/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java b/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java index dfddb11..0f4972c 100644 --- a/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java +++ b/engine/src/main/java/io/hstore/engine/maintenance/Checkpointer.java @@ -9,6 +9,7 @@ import io.hstore.engine.catalog.Generation; import io.hstore.engine.catalog.SlotRegistry; import io.hstore.engine.feed.ChangeFeed; +import io.hstore.engine.page.FileHandle; import io.hstore.engine.page.PageStore; import io.hstore.engine.tree.Epoch; import io.hstore.engine.tree.Materializer; @@ -73,6 +74,10 @@ public Generation materialized() { } public Result checkpoint() { + return FileHandle.uninterrupted(this::take); + } + + private Result take() { Generation target = transactions.current(); Epoch epoch = new Epoch(); LongStream.Builder written = LongStream.builder(); diff --git a/engine/src/main/java/io/hstore/engine/page/FileHandle.java b/engine/src/main/java/io/hstore/engine/page/FileHandle.java new file mode 100644 index 0000000..57961f7 --- /dev/null +++ b/engine/src/main/java/io/hstore/engine/page/FileHandle.java @@ -0,0 +1,114 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine.page; + +import java.io.IOException; +import java.nio.channels.ClosedByInterruptException; +import java.nio.channels.ClosedChannelException; +import java.nio.channels.FileChannel; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Supplier; + +public final class FileHandle implements AutoCloseable { + + private static final int MAX_REOPENS = 64; + + @FunctionalInterface + public interface Io { + T apply(FileChannel channel) throws IOException; + } + + @FunctionalInterface + public interface Action { + void run(FileChannel channel) throws IOException; + } + + private final Path path; + private final ReentrantLock reopening = new ReentrantLock(); + private volatile FileChannel channel; + private volatile boolean closed; + + private FileHandle(Path path, FileChannel channel) { + this.path = path; + this.channel = channel; + } + + public static FileHandle open(Path path) throws IOException { + return new FileHandle(path, FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE)); + } + + public static T uninterrupted(Supplier work) { + boolean interrupted = Thread.interrupted(); + try { + return work.get(); + } finally { + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + } + + public T call(Io io) throws IOException { + boolean interrupted = Thread.interrupted(); + try { + for (int attempt = 0; ; attempt++) { + FileChannel current = channel; + try { + return io.apply(current); + } catch (ClosedChannelException e) { + if (e instanceof ClosedByInterruptException) { + interrupted |= Thread.interrupted(); + } + if (closed || attempt >= MAX_REOPENS) { + throw e; + } + reopen(current); + } + } + } finally { + if (interrupted) { + Thread.currentThread().interrupt(); + } + } + } + + public void run(Action action) throws IOException { + call(channel -> { + action.run(channel); + return null; + }); + } + + public boolean isOpen() { + return !closed; + } + + public long size() throws IOException { + return call(FileChannel::size); + } + + private void reopen(FileChannel failed) throws IOException { + reopening.lock(); + try { + if (!closed && channel == failed) { + channel = FileChannel.open(path, StandardOpenOption.READ, StandardOpenOption.WRITE); + } + } finally { + reopening.unlock(); + } + } + + @Override + public void close() throws IOException { + reopening.lock(); + try { + closed = true; + channel.close(); + } finally { + reopening.unlock(); + } + } +} diff --git a/engine/src/main/java/io/hstore/engine/page/MappedLongs.java b/engine/src/main/java/io/hstore/engine/page/MappedLongs.java index ee86d5d..2f32c32 100644 --- a/engine/src/main/java/io/hstore/engine/page/MappedLongs.java +++ b/engine/src/main/java/io/hstore/engine/page/MappedLongs.java @@ -14,7 +14,6 @@ import java.nio.ByteOrder; import java.nio.channels.FileChannel; import java.nio.file.Path; -import java.nio.file.StandardOpenOption; import java.util.Arrays; final class MappedLongs implements AutoCloseable { @@ -28,20 +27,20 @@ final class MappedLongs implements AutoCloseable { private static final VarHandle ENTRY = LONG.varHandle(); private final Path path; - private final FileChannel channel; + private final FileHandle file; private boolean fresh; private volatile MemorySegment[] chunks = new MemorySegment[0]; - private MappedLongs(Path path, FileChannel channel) { + private MappedLongs(Path path, FileHandle file) { this.path = path; - this.channel = channel; + this.file = file; } static MappedLongs open(Path path, int magic, int format, String what) { try { - FileChannel channel = FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE); - long size = channel.size(); - MappedLongs longs = new MappedLongs(path, channel); + FileHandle file = FileHandle.open(path); + long size = file.size(); + MappedLongs longs = new MappedLongs(path, file); longs.mapThrough(size == 0 ? 0 : (int) ((size - 1) >>> CHUNK_SHIFT)); MemorySegment header = longs.chunks[0]; longs.fresh = header.get(INT, 0) == 0 && header.get(INT, 4) == 0 && header.get(LONG, 8) == 0; @@ -50,7 +49,7 @@ static MappedLongs open(Path path, int magic, int format, String what) { header.set(INT, 4, format); header.force(); } else if (header.get(INT, 0) != magic || header.get(INT, 4) != format) { - channel.close(); + file.close(); throw HStoreException.invalid(what + " " + path + " has an unknown format"); } return longs; @@ -101,7 +100,8 @@ private synchronized void mapThrough(int last) { MemorySegment[] grown = Arrays.copyOf(mapped, last + 1); try { for (int chunk = mapped.length; chunk <= last; chunk++) { - grown[chunk] = channel.map(FileChannel.MapMode.READ_WRITE, chunk * CHUNK_BYTES, CHUNK_BYTES, Arena.ofAuto()); + long offset = chunk * CHUNK_BYTES; + grown[chunk] = file.call(channel -> channel.map(FileChannel.MapMode.READ_WRITE, offset, CHUNK_BYTES, Arena.ofAuto())); } } catch (IOException e) { throw new UncheckedIOException("cannot grow " + path, e); @@ -112,7 +112,7 @@ private synchronized void mapThrough(int last) { @Override public void close() { try { - channel.close(); + file.close(); } catch (IOException e) { throw new UncheckedIOException("cannot close " + path, e); } 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 1068065..815a640 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.nio.file.StandardOpenOption; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.ReentrantLock; @@ -25,23 +24,22 @@ final class SegmentFile implements AutoCloseable { private final int id; private final Path path; - private final FileChannel channel; + private final FileHandle file; 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) { + private SegmentFile(int id, Path path, FileHandle file) { this.id = id; this.path = path; - this.channel = channel; + this.file = file; } static SegmentFile open(int id, Path path) { try { - FileChannel channel = FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE); - return new SegmentFile(id, path, channel); + return new SegmentFile(id, path, FileHandle.open(path)); } catch (IOException e) { throw new UncheckedIOException("cannot open segment " + path, e); } @@ -62,7 +60,7 @@ int id() { long unitsOnDisk() { try { - return PageId.unitsFor(channel.size()); + return PageId.unitsFor(file.size()); } catch (IOException e) { throw HStoreException.io("cannot size segment " + path, e); } @@ -104,7 +102,7 @@ private MemorySegment read(MemorySegment view, long position, int maximum) { MemorySegment view() { try { - MemorySegment whole = covering(channel.size()); + MemorySegment whole = covering(file.size()); return whole == null ? MemorySegment.NULL : whole; } catch (IOException e) { throw HStoreException.io("cannot size segment " + path, e); @@ -135,13 +133,13 @@ private MemorySegment covering(long required) { if (required <= current.byteSize()) { return current; } - long size = channel.size(); + long size = file.size(); if (size < required) { return null; } Arena replacement = Arena.ofShared(); try { - current = channel.map(FileChannel.MapMode.READ_ONLY, 0, size, replacement); + current = file.call(channel -> channel.map(FileChannel.MapMode.READ_ONLY, 0, size, replacement)); } catch (IOException | RuntimeException e) { replacement.close(); throw e; @@ -177,11 +175,13 @@ private MemorySegment copy(long position, int maximum) { private void fill(ByteBuffer buffer, long position) { try { - while (buffer.hasRemaining()) { - if (channel.read(buffer, position + buffer.position()) < 0) { - return; + file.run(channel -> { + while (buffer.hasRemaining()) { + if (channel.read(buffer, position + buffer.position()) < 0) { + return; + } } - } + }); } catch (IOException e) { throw HStoreException.io("read failed in segment " + id + " at byte " + position, e); } @@ -190,9 +190,11 @@ private void fill(ByteBuffer buffer, long position) { void write(long position, MemorySegment image) { ByteBuffer buffer = image.asByteBuffer(); try { - while (buffer.hasRemaining()) { - channel.write(buffer, position + buffer.position()); - } + file.run(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); @@ -202,7 +204,7 @@ void write(long position, MemorySegment image) { void force() { if (dirty.getAndSet(false)) { try { - channel.force(false); + file.run(channel -> channel.force(false)); } catch (IOException e) { dirty.set(true); throw HStoreException.io("sync failed for segment " + id, e); @@ -236,7 +238,7 @@ public void close() { remapping.unlock(); } try { - channel.close(); + file.close(); } catch (IOException e) { throw new UncheckedIOException(e); } diff --git a/engine/src/main/java/io/hstore/engine/wal/WriteAheadLog.java b/engine/src/main/java/io/hstore/engine/wal/WriteAheadLog.java index 4d1ad82..9df854a 100644 --- a/engine/src/main/java/io/hstore/engine/wal/WriteAheadLog.java +++ b/engine/src/main/java/io/hstore/engine/wal/WriteAheadLog.java @@ -6,6 +6,7 @@ import io.hstore.engine.HStoreException; import io.hstore.engine.page.ByteCursor; import io.hstore.engine.page.Checksums; +import io.hstore.engine.page.FileHandle; import java.io.IOException; import java.io.UncheckedIOException; @@ -44,7 +45,7 @@ public record Positioned(long lsn, long next, WalRecord record) { private final ReentrantLock lock = new ReentrantLock(); private final NavigableMap segments = new TreeMap<>(); private final AtomicLong bytesAppended = new AtomicLong(); - private FileChannel active; + private FileHandle active; private long activeStart; private long end; private boolean dirty; @@ -76,9 +77,12 @@ private void restore() { validEnd = positioned.next(); } activeStart = last.getKey(); - active = FileChannel.open(last.getValue(), StandardOpenOption.READ, StandardOpenOption.WRITE); - active.truncate(validEnd - activeStart); - active.force(true); + active = FileHandle.open(last.getValue()); + long length = validEnd - activeStart; + active.run(channel -> { + channel.truncate(length); + channel.force(true); + }); end = validEnd; } catch (IOException e) { throw new UncheckedIOException("cannot open write-ahead log " + directory, e); @@ -130,9 +134,11 @@ private void write(MemorySegment bytes) { } ByteBuffer buffer = bytes.asByteBuffer(); long position = end - activeStart; - while (buffer.hasRemaining()) { - position += active.write(buffer, position); - } + active.run(channel -> { + while (buffer.hasRemaining()) { + channel.write(buffer, position + buffer.position()); + } + }); dirty = true; } catch (IOException e) { throw HStoreException.io("write-ahead log append failed", e); @@ -141,12 +147,12 @@ private void write(MemorySegment bytes) { private void roll() throws IOException { if (active != null) { - active.force(true); + active.run(channel -> channel.force(true)); active.close(); } activeStart = end; Path path = pathFor(activeStart); - active = FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE); + active = FileHandle.open(path); segments.put(activeStart, path); syncDirectory(); } @@ -163,7 +169,7 @@ public void sync() { lock.lock(); try { if (dirty && active != null) { - active.force(false); + active.run(channel -> channel.force(false)); dirty = false; } } catch (IOException e) { @@ -225,7 +231,7 @@ public void close() { lock.lock(); try { if (active != null) { - active.force(true); + active.run(channel -> channel.force(true)); active.close(); active = null; } diff --git a/engine/src/test/java/io/hstore/engine/InterruptTest.java b/engine/src/test/java/io/hstore/engine/InterruptTest.java new file mode 100644 index 0000000..89bb417 --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/InterruptTest.java @@ -0,0 +1,65 @@ +// SPDX-FileCopyrightText: Metacog Labs +// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 + +package io.hstore.engine; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class InterruptTest { + + @TempDir + Path directory; + + @Test + void anInterruptedCallerLeavesTheEngineWorking() throws Exception { + StorageEngine engine = StorageEngine.open(directory, EngineTest.small()); + engine.write(txn -> txn.createNode(EngineTest.NODE, "before")); + onInterruptedThread(() -> { + engine.write(txn -> txn.createNode(EngineTest.NODE, "interrupted")); + engine.read(snapshot -> snapshot.resolve(0, EngineTest.NODE, "before")); + engine.feed().replay(0).count(); + engine.checkpoint(); + }); + long events = engine.feed().replay(0).count(); + for (int i = 0; i < 200; i++) { + String key = "after-" + i; + engine.write(txn -> txn.createNode(EngineTest.NODE, key)); + } + assertEquals(events + 200, engine.feed().replay(0).count()); + engine.verifyReferences(); + onInterruptedThread(engine::close); + try (StorageEngine reopened = StorageEngine.open(directory, EngineTest.small())) { + for (String key : List.of("before", "interrupted", "after-199")) { + assertTrue(reopened.read(snapshot -> snapshot.resolve(0, EngineTest.NODE, key)).isPresent(), key); + } + reopened.verifyReferences(); + } + } + + private static void onInterruptedThread(Runnable work) throws InterruptedException { + AtomicReference failure = new AtomicReference<>(); + AtomicBoolean keptInterrupt = new AtomicBoolean(); + Thread thread = Thread.ofPlatform().start(() -> { + Thread.currentThread().interrupt(); + try { + work.run(); + keptInterrupt.set(Thread.currentThread().isInterrupted()); + } catch (Throwable thrown) { + failure.set(thrown); + } + }); + thread.join(); + assertNull(failure.get()); + assertTrue(keptInterrupt.get()); + } +} diff --git a/engine/src/test/java/io/hstore/engine/page/FileHandleTest.java b/engine/src/test/java/io/hstore/engine/page/FileHandleTest.java new file mode 100644 index 0000000..ed9c4dd --- /dev/null +++ b/engine/src/test/java/io/hstore/engine/page/FileHandleTest.java @@ -0,0 +1,118 @@ +// 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.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.ClosedChannelException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class FileHandleTest { + + @TempDir + Path directory; + + @Test + void aThreadInterruptedBeforehandWritesAndKeepsItsInterrupt() throws Exception { + Path path = directory.resolve("before"); + try (FileHandle file = FileHandle.open(path)) { + Thread.currentThread().interrupt(); + try { + write(file, 0, "first"); + assertTrue(Thread.currentThread().isInterrupted()); + } finally { + Thread.interrupted(); + } + write(file, 5, "second"); + assertEquals("firstsecond", read(file, 11)); + } + assertEquals("firstsecond", Files.readString(path)); + } + + @Test + void anInterruptDuringACallReopensTheFileAndRetries() throws Exception { + try (FileHandle file = FileHandle.open(directory.resolve("during"))) { + write(file, 0, "kept"); + AtomicInteger attempts = new AtomicInteger(); + try { + String text = file.call(channel -> { + if (attempts.getAndIncrement() == 0) { + Thread.currentThread().interrupt(); + } + ByteBuffer buffer = ByteBuffer.allocate(4); + channel.read(buffer, 0); + return new String(buffer.array(), StandardCharsets.UTF_8); + }); + assertEquals("kept", text); + assertEquals(2, attempts.get()); + assertTrue(Thread.currentThread().isInterrupted()); + } finally { + Thread.interrupted(); + } + write(file, 4, "!"); + assertEquals("kept!", read(file, 5)); + } + } + + @Test + void aChannelClosedUnderAnotherThreadIsReopenedForIt() throws Exception { + try (FileHandle file = FileHandle.open(directory.resolve("other"))) { + write(file, 0, "data"); + AtomicInteger attempts = new AtomicInteger(); + long size = file.call(channel -> { + if (attempts.getAndIncrement() == 0) { + Thread closer = Thread.ofPlatform().start(() -> { + try { + channel.close(); + } catch (IOException e) { + throw new IllegalStateException(e); + } + }); + try { + closer.join(); + } catch (InterruptedException e) { + throw new IllegalStateException(e); + } + } + return channel.size(); + }); + assertEquals(4, size); + assertEquals(2, attempts.get()); + assertFalse(Thread.currentThread().isInterrupted()); + } + } + + @Test + void aClosedHandleStaysClosed() throws Exception { + FileHandle file = FileHandle.open(directory.resolve("closed")); + file.close(); + assertThrows(ClosedChannelException.class, file::size); + } + + private static void write(FileHandle file, long position, String text) throws IOException { + ByteBuffer buffer = ByteBuffer.wrap(text.getBytes(StandardCharsets.UTF_8)); + file.run(channel -> { + while (buffer.hasRemaining()) { + channel.write(buffer, position + buffer.position()); + } + }); + } + + private static String read(FileHandle file, int length) throws IOException { + ByteBuffer buffer = ByteBuffer.allocate(length); + file.run(channel -> channel.read(buffer, 0)); + return new String(buffer.array(), StandardCharsets.UTF_8); + } +} diff --git a/engine/src/test/java/io/hstore/engine/page/PageDirectoryTest.java b/engine/src/test/java/io/hstore/engine/page/PageDirectoryTest.java index 975a58d..bfc90cb 100644 --- a/engine/src/test/java/io/hstore/engine/page/PageDirectoryTest.java +++ b/engine/src/test/java/io/hstore/engine/page/PageDirectoryTest.java @@ -13,12 +13,31 @@ 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 PageDirectoryTest { @TempDir Path directory; + @Test + void anInterruptedThreadGrowsTheDirectoryWithoutClosingIt() throws Exception { + long far = 3L << 20; + boolean[] keptInterrupt = new boolean[1]; + try (PageDirectory pages = PageDirectory.open(directory.resolve(PageDirectory.FILE))) { + Thread grower = Thread.ofPlatform().start(() -> { + Thread.currentThread().interrupt(); + pages.put(far, 42); + keptInterrupt[0] = Thread.currentThread().isInterrupted(); + }); + grower.join(); + pages.put(far * 2, 43); + assertTrue(keptInterrupt[0]); + assertEquals(42, pages.get(far)); + assertEquals(43, pages.get(far * 2)); + } + } + @Test void addressesAndTheNextNumberSurviveReopenAcrossChunks() { Path file = directory.resolve(PageDirectory.FILE); 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 071e682..15c5105 100644 --- a/engine/src/test/java/io/hstore/engine/page/PageStoreTest.java +++ b/engine/src/test/java/io/hstore/engine/page/PageStoreTest.java @@ -99,6 +99,33 @@ void finishedSegmentsAreMappedAndTheGrowingOneIsNotUntilItGrows() { } } + @Test + void anInterruptedThreadReadsAndWritesPagesWithoutClosingTheSegment() throws Exception { + try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) { + long before = write(store, PAGE_SIZE); + byte[] expected = bytes(store.read(before)); + long[] written = new long[1]; + boolean[] stillInterrupted = new boolean[1]; + List seen = new ArrayList<>(); + Thread interrupted = Thread.ofPlatform().start(() -> { + Thread.currentThread().interrupt(); + written[0] = write(store, PAGE_SIZE / 2); + seen.add(bytes(store.read(before))); + store.sync(); + stillInterrupted[0] = Thread.currentThread().isInterrupted(); + }); + interrupted.join(); + assertTrue(stillInterrupted[0]); + assertArrayEquals(expected, seen.getFirst()); + long after = write(store, PAGE_SIZE); + assertEquals(store.segmentOf(before), store.segmentOf(after)); + assertEquals(0, store.mappedBytes(store.segmentOf(after))); + assertArrayEquals(expected, bytes(store.read(before))); + assertArrayEquals(image(new Random(PAGE_SIZE / 2), PAGE_SIZE / 2), bytes(store.read(written[0]))); + store.sync(); + } + } + @Test void pagesAreNumberedInOrderWhicheverSegmentHoldsThem() { try (PageStore store = PageStore.open(directory, PAGE_SIZE, PAGES_PER_SEGMENT, List.of())) {