diff --git a/shiroha-server/src/main/java/io/nanachiyo0721/shiroha/data/BufferedLinearRegionFile.java b/shiroha-server/src/main/java/io/nanachiyo0721/shiroha/data/BufferedLinearRegionFile.java index 61acc34..485c782 100644 --- a/shiroha-server/src/main/java/io/nanachiyo0721/shiroha/data/BufferedLinearRegionFile.java +++ b/shiroha-server/src/main/java/io/nanachiyo0721/shiroha/data/BufferedLinearRegionFile.java @@ -29,15 +29,30 @@ import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.StandardCopyOption; import java.nio.file.StandardOpenOption; -import java.util.Arrays; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; +/** + * Lock hierarchy (always acquire top to bottom, never the reverse): + *
    + *
  1. {@code syncLock} — serializes master file syncs against close
    2. + *
  2. {@code Bucket.lock} — per-bucket lazy-load guard
    3. + *
  3. {@code masterFileLock} — master file read / append / replace
    4. + *
  4. {@code regionObjectLock} — in-memory sector table + swap file channel
    5. + *
+ * The atomic flags (closed / synced / beingSynced / lastWritten) and the bucket + * epochs are lock-free and may be touched while holding any (or no) lock. + */ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.RegionFile { private static final double SWAP_FILE_AUTO_COMPACT_PERCENT = 3.0 / 5.0; // 60 % private static final long SWAP_FILE_AUTO_COMPACT_SIZE = 1024 * 1024; // 1 MiB + // master file WAL appends leave the replaced bucket records behind as garbage; once + // it piles up past this threshold the next sync compacts via a full tmp-file rewrite + private static final double MASTER_FILE_AUTO_COMPACT_PERCENT = SWAP_FILE_AUTO_COMPACT_PERCENT; + private static final long MASTER_FILE_AUTO_COMPACT_SIZE = SWAP_FILE_AUTO_COMPACT_SIZE; + private static final long SWAP_FILE_SUPER_BLOCK = 0x1145141919810L; private static final int SWAP_FILE_HASH_SEED = 0x0721; // ～(∠・ω< )⌒★ private static final byte SWAP_FILE_VERSION = 0x02; // ver 2.0 @@ -74,6 +89,10 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R private final Path masterFilePath; private final Path swapFilePath; + // outermost lock: serializes syncToMasterFile() against closeInternal(), so the + // swap channel can never be torn down while a sync is still reading from it + private final Object syncLock = new Object(); + private final ReadWriteLock regionObjectLock = new ReentrantReadWriteLock(); private final XXHash32 xxHash32 = XXHashFactory.fastestInstance().hash32(); private Sector[] sectors = new Sector[1024]; @@ -134,6 +153,34 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R } } + private static void transferFully(FileChannel source, long sourceOffset, long count, FileChannel target, long targetOffset) throws IOException { + target.position(targetOffset); + + long transferred = 0; + while (transferred < count) { + transferred += source.transferTo(sourceOffset + transferred, count - transferred, target); + } + } + + // replaces target with source, deleting source if both attempts fail + private static void atomicReplace(Path source, Path target) throws IOException { + try { + Files.move(source, target, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE); + } catch (Throwable e) { + // atomic move might be unsupported on some file systems, so give it an attempt to retry without atomic move + try { + Files.move(source, target, StandardCopyOption.REPLACE_EXISTING); + } catch (Throwable ex) { + e.addSuppressed(ex); + + // delete file that failed to replace + Files.deleteIfExists(source); + + throw new IOException("Failed to replace " + target + "!", e); + } + } + } + private void cleanUpSwapFile() throws IOException { Files.deleteIfExists(this.swapFilePath); } @@ -142,6 +189,10 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R final int bucketIndex = chunkIndex >> BUCKET_SHIFT; final Bucket bucket = this.buckets[bucketIndex]; + if (bucket.loaded) { // volatile fast path + return; + } + // bucket lock -> master read lock -> swap write lock synchronized (bucket.lock) { if (bucket.loaded) { @@ -153,32 +204,26 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R } } - private long markBucketDirty(int chunkIndex) { - return this.markBucketDirtyByIndex(chunkIndex >> BUCKET_SHIFT); + // used by the legacy parsers: their data goes through the write path directly, + // so the bucket must be flagged loaded first to avoid a recursive lazy-load + private void markBucketLoaded(int chunkIndex) { + final Bucket bucket = this.buckets[chunkIndex >> BUCKET_SHIFT]; + + synchronized (bucket.lock) { + bucket.loaded = true; + } } - private long markBucketDirtyByIndex(int bucketIndex) { - final Bucket bucket = this.buckets[bucketIndex]; - - return bucket.writeEpoch.incrementAndGet(); + private void markBucketDirty(int chunkIndex) { + this.buckets[chunkIndex >> BUCKET_SHIFT].writeEpoch.incrementAndGet(); } private long getBucketWriteEpoch(int bucketIndex) { - final Bucket bucket = this.buckets[bucketIndex]; - - return bucket.writeEpoch.get(); - } - - private long getBucketSyncedEpoch(int bucketIndex) { - final Bucket bucket = this.buckets[bucketIndex]; - - return bucket.syncedEpoch.get(); + return this.buckets[bucketIndex].writeEpoch.get(); } private void markBucketSynced(int bucketIndex, long syncedEpoch) { - final Bucket bucket = this.buckets[bucketIndex]; - - bucket.syncedEpoch.accumulateAndGet(syncedEpoch, Math::max); + this.buckets[bucketIndex].syncedEpoch.accumulateAndGet(syncedEpoch, Math::max); } private boolean isBucketDirty(int bucketIndex) { @@ -223,14 +268,7 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R } public void syncIfNeeded() throws IOException { - // the sync operation is just coping the data from swap file to the master file - // so we could acquire read lock simply so that we won't block any other read operations try { - // skip if closed already - if (this.isClosed()) { - return; - } - this.syncToMasterFile(); } finally { BEING_SYNCED_HANDLE.setVolatile(this, false); // mark as not being synced @@ -238,19 +276,27 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R } private void syncToMasterFile() throws IOException { - // prevent multiple syncs in the same time - if (!SYNCED_HANDLE.compareAndSet(this, false, true)) { - return; - } + // serialized against close: the swap channel cannot go away under a running sync + synchronized (this.syncLock) { + // skip if closed already + if (this.isClosedRaw()) { + return; + } - try { - // this.masterFileParser.writeMainFile(this.masterFilePath); - this.masterFileParser.writeMainFileBucketed(this.masterFilePath); - } catch (Throwable e) { - // set back - SYNCED_HANDLE.setVolatile(this, false); + // fast skip when there is nothing to sync; writers flip the flag back + // via markAsToSync() which triggers the next round + if (!SYNCED_HANDLE.compareAndSet(this, false, true)) { + return; + } - throw new IOException("Failed to sync to master file!", e); + try { + this.masterFileParser.sync(this.masterFilePath); + } catch (Throwable e) { + // set back + SYNCED_HANDLE.setVolatile(this, false); + + throw new IOException("Failed to sync to master file!", e); + } } } @@ -329,29 +375,20 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R return; } - long spareSize = this.currentAcquiredIndex; - - spareSize -= this.headerSize(); + long liveSize = 0; for (Sector sector : this.sectors) { // skip no data sectors if (!sector.hasData()) { continue; } - spareSize -= sector.length; + liveSize += sector.length; } - long sectorSize = 0; - for (Sector sector : this.sectors) { - // skip no data sectors - if (!sector.hasData()) { - continue; - } + // everything acquired but not covered by a live sector is garbage + final long spareSize = this.currentAcquiredIndex - this.headerSize() - liveSize; - sectorSize += sector.length; - } - - final boolean compactRequested = spareSize > SWAP_FILE_AUTO_COMPACT_SIZE && (double) spareSize > ((double) sectorSize) * SWAP_FILE_AUTO_COMPACT_PERCENT; + final boolean compactRequested = spareSize > SWAP_FILE_AUTO_COMPACT_SIZE && (double) spareSize > ((double) liveSize) * SWAP_FILE_AUTO_COMPACT_PERCENT; // try auto compact to clean the garbage area if (compactRequested) { @@ -371,24 +408,58 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R } private void closeInternal() throws IOException { - this.syncIfNeeded(); + synchronized (this.syncLock) { + if (this.isClosedRaw()) { + // already closed (possibly by a compact disaster path): just make sure + // both channels are really gone — close is idempotent + this.regionObjectLock.writeLock().lock(); + try { + this.swapFileChannel.close(); + } finally { + this.regionObjectLock.writeLock().unlock(); + } - this.regionObjectLock.writeLock().lock(); - try { - this.markClosed(); + this.masterFileParser.close(); + return; + } - this.swapFileChannel.close(); - } finally { - this.regionObjectLock.writeLock().unlock(); + // final sync so no buffered data is lost; holding syncLock also guarantees no + // concurrent flusher sync is still running when we tear down below. + // if this throws we deliberately stay open: the flusher can retry the sync + // later, and the not-yet-synced swap data is not dropped on the floor + this.syncToMasterFile(); + + IOException failure = null; + + this.regionObjectLock.writeLock().lock(); + try { + this.markClosed(); + + this.swapFileChannel.close(); + } catch (IOException e) { + failure = e; + } finally { + this.regionObjectLock.writeLock().unlock(); + } + + try { + // acquired after the region lock is fully released, never inside it (lock hierarchy) + this.masterFileParser.close(); + } catch (IOException e) { + if (failure == null) failure = e; else failure.addSuppressed(e); + } + + if (failure != null) { + throw failure; + } } } - private void markClosed() throws IOException { - if (!CLOSED_HANDLE.compareAndSet(this, false, true)) { - throw new IOException("Already closed!"); + private void markClosed() { + // lenient CAS: the disaster path of compactSwapFile() may have closed us already + if (CLOSED_HANDLE.compareAndSet(this, false, true)) { + this.flusher.removeFile(this); } - - this.flusher.removeFile(this); } private void compactSwapFile() throws IOException { @@ -413,7 +484,7 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R long newAcquiredIndex; - final Path targetTemp = new File(this.swapFilePath.toString() + ".tmp").toPath(); + final Path targetTemp = Path.of(this.swapFilePath + ".tmp"); try (FileChannel tempChannel = FileChannel.open( targetTemp, @@ -423,7 +494,6 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R StandardOpenOption.TRUNCATE_EXISTING )) { long offsetPointer = this.headerSize(); - tempChannel.position(offsetPointer); for (Sector sector : newSectorsToBeReplaced) { // skip cleared or no data-contained sectors @@ -432,7 +502,7 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R } // transfer to target - sector.transferTo(this.swapFileChannel, tempChannel); + transferFully(this.swapFileChannel, sector.offset, sector.length, tempChannel, offsetPointer); // recalculate the offset and length final Sector newRecalculated = new Sector(sector.index, offsetPointer, sector.length); @@ -459,37 +529,17 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R // replace swap file try { - Files.move( - targetTemp, - this.swapFilePath, - StandardCopyOption.REPLACE_EXISTING, - StandardCopyOption.ATOMIC_MOVE - ); + atomicReplace(targetTemp, this.swapFilePath); } catch (Throwable e) { - // atomic move might be unsupported on some file systems, so give it an attempt to retry without atomic move - try { - Files.move( - targetTemp, - this.swapFilePath, - StandardCopyOption.REPLACE_EXISTING - ); - } catch (Throwable ex) { - // now we are totally failed - e.addSuppressed(ex); - - // delete file that failed to replace - Files.deleteIfExists(targetTemp); - // recalculate acquired index - this.recalculateAcquiredIndex(); - // reopen closed channel - this.reopenSwapFileChannel(); - // fast-fail - this.markClosed(); // prevent new writing & sync operations - throw new IOException("Failed to replace original swap file!", e); - } + // recalculate acquired index + this.recalculateAcquiredIndex(); + // reopen closed channel + this.reopenSwapFileChannel(); + // fast-fail + this.markClosed(); // prevent new writing & sync operations + throw new IOException("Failed to replace original swap file!", e); } - try { // reopen file channel this.reopenSwapFileChannel(); @@ -545,10 +595,6 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R } private @Nullable ByteBuffer readChunkDataRaw(int index) throws IOException { - return this.readChunkDataRaw(index, true); - } - - private @Nullable ByteBuffer readChunkDataRaw(int index, boolean acquireLock) throws IOException { final ByteBuffer raw; this.regionObjectLock.readLock().lock(); @@ -606,6 +652,8 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R private void writeChunk(int x, int z, @NotNull ByteBuffer data) throws IOException { final int chunkIndex = getChunkIndex(x, z); + this.ensureBucketLoaded(chunkIndex); + if (data.remaining() > MAX_SIZE_PER_CHUNK) { throw new RegionFileStorage.RegionFileSizeException("Writing too large chunk, limit : " + MAX_SIZE_PER_CHUNK + " but got : " + data.remaining()); } @@ -703,11 +751,6 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R @Override public void write(@NotNull ChunkPos pos, ByteBuffer buf) throws IOException { - - final int chunkIndex = getChunkIndex(pos.x(), pos.z()); - - this.ensureBucketLoaded(chunkIndex); - this.writeChunk(pos.x(), pos.z(), buf); } @@ -819,16 +862,6 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R this.length = length; } - public void transferTo(@NotNull FileChannel source, @NotNull FileChannel target) throws IOException { - long transferred = 0; - while (transferred < this.length) { - transferred += source.transferTo( - this.offset + transferred, - this.length - transferred, - target); - } - } - public @NotNull ByteBuffer read(@NotNull FileChannel channel) throws IOException { final ByteBuffer result = ByteBuffer.allocate((int) this.length); @@ -906,9 +939,6 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R public void close() throws IOException { ByteBuffer bytebuffer = ByteBuffer.wrap(this.buf, 0, this.count); - final int chunkIndex = getChunkIndex(this.pos.x(), this.pos.z()); - - BufferedLinearRegionFile.this.ensureBucketLoaded(chunkIndex); BufferedLinearRegionFile.this.writeChunk(this.pos.x(), this.pos.z(), bytebuffer); BufferedLinearRegionFile.this.flushInternal(); @@ -916,238 +946,357 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R } private class LinearMasterFileParser { - // V3 new format layout: - // [0, 14): header — superblock(8) + version(1) + compressionLevel(1) + xxHash32Seed(4) + // V3 bucketed format layout: + // [0, 14): header — superblock(8) + version(1) + compressionLevel(1) + xxHash32Seed(4) // [14, 142): position table — BUCKET_COUNT(16) × long(8) each; 0 = no data for that bucket - // [526, EOF): bucket data — originalLen(int) + compressedLen(int) + compressedData - private static final long V3_POS_TABLE_OFFSET = 14L; + // [142, EOF): bucket records — originalLen(int) + compressedLen(int) + compressedData + private static final int V3_HEADER_SIZE = 14; + private static final long V3_POS_TABLE_OFFSET = V3_HEADER_SIZE; private static final int V3_POS_TABLE_SIZE = BUCKET_COUNT * Long.BYTES; // 128 private static final long V3_DATA_AREA_OFFSET = V3_POS_TABLE_OFFSET + V3_POS_TABLE_SIZE; // 142 + private static final int V3_RECORD_HEADER_SIZE = Integer.BYTES * 2; // originalLen + compressedLen private final ReadWriteLock masterFileLock = new ReentrantReadWriteLock(); - public void writeMainFileBucketed(@NotNull Path mainFile) throws IOException { + // WAL(append) state, guarded by masterFileLock: null until the first sync after + // open has fully rewritten the master file; afterwards syncs only append changed + // buckets to the tail and update the position table in place. + // recordSizes mirrors positionTable (size of each live record) so the garbage + // ratio can be computed without touching the disk + private @Nullable FileChannel appendChannel; + private long[] positionTable; + private long[] recordSizes; + private long appendOffset; + + // a consistent snapshot of one bucket taken from the swap file; + // payload == null means the bucket holds no chunks at all + private record BucketRecord(long epoch, byte @Nullable [] payload) { + } + + // must be called under syncLock (see syncToMasterFile) + public void sync(@NotNull Path mainFile) throws IOException { + this.masterFileLock.writeLock().lock(); + try { + // full rewrite on the first sync after open, and afterwards whenever the + // appended garbage passed the auto-compact threshold: writes a tmp file, + // then atomically replaces the master file with it + if (this.appendChannel == null || this.shouldCompactMasterFile()) { + this.rewriteFully(mainFile); + } else { + // WAL-style otherwise: only append the dirty buckets + this.appendDirtyBuckets(); + } + } finally { + this.masterFileLock.writeLock().unlock(); + } + } + + // only valid in WAL mode (appendChannel != null); mirrors the swap file heuristic + private boolean shouldCompactMasterFile() { + long liveSize = 0; + for (final long size : this.recordSizes) { + liveSize += size; + } + + final long spareSize = this.appendOffset - V3_DATA_AREA_OFFSET - liveSize; + + return spareSize > MASTER_FILE_AUTO_COMPACT_SIZE && (double) spareSize > ((double) liveSize) * MASTER_FILE_AUTO_COMPACT_PERCENT; + } + + private void rewriteFully(@NotNull Path mainFile) throws IOException { + // a compacting rewrite replaces the whole file, so drop the old append channel + // first; if anything below fails, the next sync just retries via this path + if (this.appendChannel != null) { + this.appendChannel.close(); + this.appendChannel = null; + } + final Path tmpFilePath = Path.of(mainFile + ".tmp"); final long[] syncedBucketEpochs = new long[BUCKET_COUNT]; final long[] newPositionTable = new long[BUCKET_COUNT]; + final long[] newRecordSizes = new long[BUCKET_COUNT]; + final long newAppendOffset; - // note: there is no necessary hold the write lock for this stuff - // as the truly write operations only happens on replacing the master file (see the file move call below this hunk) - // and we had CAS flags to prevent multiple synchronization happening at the same time + // open the old file to copy the non-dirty buckets over + try (FileChannel oldChannel = this.openV3MasterFile(mainFile)) { + final long[] oldPositionTable = oldChannel == null ? null : this.parseOffsetTable(oldChannel); - // Open old file to copy non-dirty buckets - long[] oldPositionTable = null; - FileChannel oldChannel = null; - - this.masterFileLock.writeLock().lock(); - try { - if (Files.exists(mainFile)) { - try { - oldChannel = FileChannel.open(mainFile, StandardOpenOption.READ); - if (oldChannel.size() >= V3_DATA_AREA_OFFSET) { - final ByteBuffer hdr = ByteBuffer.allocate(14); - readFullyAt(oldChannel, hdr, 0); - hdr.flip(); - if (hdr.getLong() == MASTER_FILE_SUPER_BLOCK && hdr.get() == MASTER_FILE_VERSION_BUCKET) { - oldPositionTable = this.parseOffsetTable(oldChannel); - } else { - oldChannel.close(); - oldChannel = null; - } - } else { - oldChannel.close(); - oldChannel = null; - } - } catch (Throwable e) { - if (oldChannel != null) { - try { - oldChannel.close(); - } catch (IOException e2) { - e.addSuppressed(e2); - } - } - - throw new RuntimeException(e); - } - } - - BufferedLinearRegionFile.this.regionObjectLock.readLock().lock(); try (FileChannel outChannel = FileChannel.open(tmpFilePath, StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING)) { + this.writeV3Header(outChannel); - // Write header (14 bytes) - final ByteBuffer header = ByteBuffer.allocate(14); - header.putLong(MASTER_FILE_SUPER_BLOCK); - header.put(MASTER_FILE_VERSION_BUCKET); - header.put(BufferedLinearRegionFile.this.compressionLevel); - header.putInt(BufferedLinearRegionFile.this.xxHash32Seed); - header.flip(); - writeFullyAt(outChannel, header, 0); - - // Write position table placeholder (all zeros, filled in at the end) + // position table placeholder (all zeros, filled in at the end) writeFullyAt(outChannel, ByteBuffer.allocate(V3_POS_TABLE_SIZE), V3_POS_TABLE_OFFSET); long dataOffset = V3_DATA_AREA_OFFSET; for (int bucketIndex = 0; bucketIndex < BUCKET_COUNT; bucketIndex++) { - final long bucketWriteEpoch = BufferedLinearRegionFile.this.getBucketWriteEpoch(bucketIndex); - final boolean isBucketDirty = bucketWriteEpoch != BufferedLinearRegionFile.this.getBucketSyncedEpoch(bucketIndex); - - if (isBucketDirty) { - final int baseChunk = bucketIndex << BUCKET_SHIFT; - final ByteArrayOutputStream rawBuf = new ByteArrayOutputStream(); - final DataOutputStream rawOut = new DataOutputStream(rawBuf); - boolean hasAny = false; - - for (int i = 0; i < BUCKET_SIZE; i++) { - // swap read lock - final ByteBuffer data = BufferedLinearRegionFile.this.readChunkDataRaw(baseChunk + i, false); - - // note: null -> no data contained - if (data == null) { - rawOut.writeInt(0); - } else { - final byte[] arr = new byte[data.remaining()]; - data.get(arr); - rawOut.writeInt(arr.length); - rawOut.write(arr); - hasAny = true; - } - } - rawOut.flush(); - - if (hasAny) { - final byte[] raw = rawBuf.toByteArray(); - final byte[] compressed = Zstd.compress(raw, BufferedLinearRegionFile.this.compressionLevel); + if (BufferedLinearRegionFile.this.isBucketDirty(bucketIndex)) { + final BucketRecord record = this.buildBucketRecord(bucketIndex); + if (record.payload() != null) { + writeFullyAt(outChannel, ByteBuffer.wrap(record.payload()), dataOffset); newPositionTable[bucketIndex] = dataOffset; - - final ByteBuffer bucketBuf = ByteBuffer.allocate(8 + compressed.length); - bucketBuf.putInt(raw.length); // original (uncompressed) length - bucketBuf.putInt(compressed.length); // compressed length - bucketBuf.put(compressed); - bucketBuf.flip(); - writeFullyAt(outChannel, bucketBuf, dataOffset); - dataOffset += bucketBuf.limit(); + newRecordSizes[bucketIndex] = record.payload().length; + dataOffset += record.payload().length; } - // else: newPositionTable[bucketIndex] stays 0 + // else: the bucket is empty now, its table entry stays 0 - syncedBucketEpochs[bucketIndex] = bucketWriteEpoch; - } else { - // Not dirty: copy bytes from old file if available - if (oldPositionTable != null && oldPositionTable[bucketIndex] != 0) { - final long oldOffset = oldPositionTable[bucketIndex]; - final ByteBuffer lensBuf = ByteBuffer.allocate(8); - readFullyAt(oldChannel, lensBuf, oldOffset); - lensBuf.flip(); - lensBuf.getInt(); // skip originalLen - final int compressedLen = lensBuf.getInt(); + syncedBucketEpochs[bucketIndex] = record.epoch(); + } else if (oldPositionTable != null && oldPositionTable[bucketIndex] != 0) { + // not dirty: copy the record bytes straight from the old file + final long oldOffset = oldPositionTable[bucketIndex]; + final ByteBuffer lens = this.readRecordLengths(oldChannel, oldOffset); + lens.getInt(); // skip originalLen + final long recordSize = V3_RECORD_HEADER_SIZE + (long) lens.getInt(); - final long bucketTotalSize = 8L + compressedLen; - newPositionTable[bucketIndex] = dataOffset; - outChannel.position(dataOffset); - long remaining = bucketTotalSize; - long srcPos = oldOffset; - while (remaining > 0) { - final long transferred = oldChannel.transferTo(srcPos, remaining, outChannel); - srcPos += transferred; - remaining -= transferred; - } - dataOffset += bucketTotalSize; - } + transferFully(oldChannel, oldOffset, recordSize, outChannel, dataOffset); + newPositionTable[bucketIndex] = dataOffset; + newRecordSizes[bucketIndex] = recordSize; + dataOffset += recordSize; } } - // Write the finalized position table - final ByteBuffer posTableBuf = ByteBuffer.allocate(V3_POS_TABLE_SIZE); - for (final long pos : newPositionTable) { - posTableBuf.putLong(pos); - } - posTableBuf.flip(); - writeFullyAt(outChannel, posTableBuf, V3_POS_TABLE_OFFSET); + // write the finalized position table + writeFullyAt(outChannel, this.encodePositionTable(newPositionTable), V3_POS_TABLE_OFFSET); outChannel.force(true); - } finally { - BufferedLinearRegionFile.this.regionObjectLock.readLock().unlock(); - if (oldChannel != null) { - oldChannel.close(); - } + newAppendOffset = dataOffset; } - - try { - Files.move(tmpFilePath, mainFile, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE); - } catch (Throwable e) { - - try { - Files.move(tmpFilePath, mainFile, StandardCopyOption.REPLACE_EXISTING); - } catch (Throwable ex) { - Files.deleteIfExists(tmpFilePath); - - e.addSuppressed(ex); - - throw new IOException("Failed to replace master file!", e); - } - } - } finally { - this.masterFileLock.writeLock().unlock(); } + atomicReplace(tmpFilePath, mainFile); + + // enter WAL mode: keep the freshly written master file open for appending syncs + this.appendChannel = FileChannel.open(mainFile, StandardOpenOption.READ, StandardOpenOption.WRITE); + this.positionTable = newPositionTable; + this.recordSizes = newRecordSizes; + this.appendOffset = newAppendOffset; + + this.markBucketsSynced(syncedBucketEpochs); + } + + private void appendDirtyBuckets() throws IOException { + final FileChannel channel = this.appendChannel; + final long[] syncedBucketEpochs = new long[BUCKET_COUNT]; + final long[] newPositionTable = this.positionTable.clone(); + final long[] newRecordSizes = this.recordSizes.clone(); + long dataOffset = this.appendOffset; + boolean anyDirty = false; + + for (int bucketIndex = 0; bucketIndex < BUCKET_COUNT; bucketIndex++) { + if (!BufferedLinearRegionFile.this.isBucketDirty(bucketIndex)) { + continue; + } + + final BucketRecord record = this.buildBucketRecord(bucketIndex); + + if (record.payload() != null) { + writeFullyAt(channel, ByteBuffer.wrap(record.payload()), dataOffset); + newPositionTable[bucketIndex] = dataOffset; + newRecordSizes[bucketIndex] = record.payload().length; + dataOffset += record.payload().length; + } else { + // the bucket is empty now + newPositionTable[bucketIndex] = 0; + newRecordSizes[bucketIndex] = 0; + } + + syncedBucketEpochs[bucketIndex] = record.epoch(); + anyDirty = true; + } + + if (!anyDirty) { + return; + } + + // make the appended records durable before the position table may point at them + channel.force(false); + + // commit the new tail first: even a torn position table write can then never + // cause a later append to overwrite records the on-disk table already references + this.appendOffset = dataOffset; + + writeFullyAt(channel, this.encodePositionTable(newPositionTable), V3_POS_TABLE_OFFSET); + channel.force(true); + + this.positionTable = newPositionTable; + this.recordSizes = newRecordSizes; + + this.markBucketsSynced(syncedBucketEpochs); + } + + // snapshots one bucket under a short read lock (raw sector bytes only); + // LZ4 decompression and zstd compression both run outside any lock so + // writers are only blocked while the raw bytes are copied + private @NotNull BucketRecord buildBucketRecord(int bucketIndex) throws IOException { + final int baseChunkIndex = bucketIndex << BUCKET_SHIFT; + final ByteBuffer[] rawSectors = new ByteBuffer[BUCKET_SIZE]; + final long epoch; + + BufferedLinearRegionFile.this.regionObjectLock.readLock().lock(); + try { + // the epoch is taken before the data: writes completing afterwards bump + // it further, so they simply get picked up by the next sync round + epoch = BufferedLinearRegionFile.this.getBucketWriteEpoch(bucketIndex); + + for (int i = 0; i < BUCKET_SIZE; i++) { + final Sector sector = BufferedLinearRegionFile.this.sectors[baseChunkIndex + i]; + + rawSectors[i] = sector.hasData() ? sector.read(BufferedLinearRegionFile.this.swapFileChannel) : null; + } + } finally { + BufferedLinearRegionFile.this.regionObjectLock.readLock().unlock(); + } + + final ByteArrayOutputStream rawBuf = new ByteArrayOutputStream(); + final DataOutputStream rawOut = new DataOutputStream(rawBuf); + boolean hasAny = false; + + for (int i = 0; i < BUCKET_SIZE; i++) { + final ByteBuffer rawSector = rawSectors[i]; + + // note: null -> no data contained + if (rawSector == null) { + rawOut.writeInt(0); + continue; + } + + final ByteBuffer chunkData = BufferedLinearRegionFile.this.compressingOps.decompress(rawSector); + final byte[] arr = new byte[chunkData.remaining()]; + chunkData.get(arr); + + rawOut.writeInt(arr.length); + rawOut.write(arr); + hasAny = true; + } + rawOut.flush(); + + if (!hasAny) { + return new BucketRecord(epoch, null); + } + + final byte[] raw = rawBuf.toByteArray(); + final byte[] compressed = Zstd.compress(raw, BufferedLinearRegionFile.this.compressionLevel); + + final ByteBuffer payload = ByteBuffer.allocate(V3_RECORD_HEADER_SIZE + compressed.length); + payload.putInt(raw.length); // original (uncompressed) length + payload.putInt(compressed.length); // compressed length + payload.put(compressed); + + return new BucketRecord(epoch, payload.array()); + } + + private void markBucketsSynced(long[] syncedBucketEpochs) { for (int i = 0; i < syncedBucketEpochs.length; i++) { - final long syncedEpoch = syncedBucketEpochs[i]; - if (syncedEpoch != 0L) { - BufferedLinearRegionFile.this.markBucketSynced(i, syncedEpoch); + // note: a dirty bucket always has a write epoch >= 1, so 0 = untouched + if (syncedBucketEpochs[i] != 0L) { + BufferedLinearRegionFile.this.markBucketSynced(i, syncedBucketEpochs[i]); } } } - private void loadBucketsFor(Path file, int bucketIndex) throws IOException { + // opens the master file for reading if it exists and is a valid V3 bucketed file, else null + private @Nullable FileChannel openV3MasterFile(@NotNull Path mainFile) throws IOException { + if (!Files.exists(mainFile)) { + return null; + } + + final FileChannel channel = FileChannel.open(mainFile, StandardOpenOption.READ); + try { + if (channel.size() >= V3_DATA_AREA_OFFSET) { + final ByteBuffer header = ByteBuffer.allocate(V3_HEADER_SIZE); + readFullyAt(channel, header, 0); + header.flip(); + + if (header.getLong() == MASTER_FILE_SUPER_BLOCK && header.get() == MASTER_FILE_VERSION_BUCKET) { + return channel; + } + } + } catch (Throwable e) { + try { + channel.close(); + } catch (IOException e2) { + e.addSuppressed(e2); + } + + throw e; + } + + channel.close(); + return null; + } + + private void writeV3Header(@NotNull FileChannel channel) throws IOException { + final ByteBuffer header = ByteBuffer.allocate(V3_HEADER_SIZE); + + header.putLong(MASTER_FILE_SUPER_BLOCK); + header.put(MASTER_FILE_VERSION_BUCKET); + header.put(BufferedLinearRegionFile.this.compressionLevel); + header.putInt(BufferedLinearRegionFile.this.xxHash32Seed); + header.flip(); + + writeFullyAt(channel, header, 0); + } + + private @NotNull ByteBuffer encodePositionTable(long[] table) { + final ByteBuffer buf = ByteBuffer.allocate(V3_POS_TABLE_SIZE); + + for (final long pos : table) { + buf.putLong(pos); + } + + return buf.flip(); + } + + private @NotNull ByteBuffer readRecordLengths(@NotNull FileChannel channel, long recordOffset) throws IOException { + final ByteBuffer lens = ByteBuffer.allocate(V3_RECORD_HEADER_SIZE); + + readFullyAt(channel, lens, recordOffset); + + return lens.flip(); + } + + public void close() throws IOException { + this.masterFileLock.writeLock().lock(); + try { + if (this.appendChannel != null) { + this.appendChannel.close(); + this.appendChannel = null; + } + } finally { + this.masterFileLock.writeLock().unlock(); + } + } + + private void loadBucketsFor(@NotNull Path file, int bucketIndex) throws IOException { final int beginChunkIndex = bucketIndex << BUCKET_SHIFT; this.masterFileLock.readLock().lock(); - try { - if (!Files.exists(file)) { - return; - } + final ByteBuffer decompressed; - try (FileChannel channel = FileChannel.open(file, StandardOpenOption.READ)) { - if (channel.size() < V3_DATA_AREA_OFFSET) { + if (this.appendChannel != null) { + // WAL mode: reuse the always-open channel and the cached position table + decompressed = this.readBucketData(this.appendChannel, this.positionTable[bucketIndex]); + } else { + if (!Files.exists(file)) { return; } - final ByteBuffer headerBuf = ByteBuffer.allocate(14); - readFullyAt(channel, headerBuf, 0); - headerBuf.flip(); + try (FileChannel channel = FileChannel.open(file, StandardOpenOption.READ)) { + if (channel.size() < V3_DATA_AREA_OFFSET) { + return; + } - final long superblock = headerBuf.getLong(); - if (superblock != MASTER_FILE_SUPER_BLOCK) - throw new IOException("Invalid superblock " + superblock + "!"); + this.checkV3Header(channel); - final byte version = headerBuf.get(); - if (version != MASTER_FILE_VERSION_BUCKET) - throw new IOException("Unknown version: " + version); + decompressed = this.readBucketData(channel, this.parseOffsetTable(channel)[bucketIndex]); + } + } - // compressionLevel and hashSeed consumed but not used here - headerBuf.get(); - headerBuf.getInt(); - - final long[] posTable = this.parseOffsetTable(channel); - - // New format: jump directly to bucket data - final long bucketDataOffset = posTable[bucketIndex]; - if (bucketDataOffset == 0) return; - - final ByteBuffer lensBuf = ByteBuffer.allocate(8); - readFullyAt(channel, lensBuf, bucketDataOffset); - lensBuf.flip(); - final int originalLen = lensBuf.getInt(); - final int compressedLen = lensBuf.getInt(); - - final byte[] compressedData = new byte[compressedLen]; - readFullyAt(channel, ByteBuffer.wrap(compressedData), bucketDataOffset + 8); - - final ByteBuffer decompressed = ByteBuffer.wrap(Zstd.decompress(compressedData, originalLen)); + if (decompressed != null) { this.loadChunksFromBucketData(decompressed, beginChunkIndex); } } finally { @@ -1155,6 +1304,38 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R } } + private void checkV3Header(@NotNull FileChannel channel) throws IOException { + final ByteBuffer headerBuf = ByteBuffer.allocate(V3_HEADER_SIZE); + readFullyAt(channel, headerBuf, 0); + headerBuf.flip(); + + final long superblock = headerBuf.getLong(); + if (superblock != MASTER_FILE_SUPER_BLOCK) + throw new IOException("Invalid superblock " + superblock + "!"); + + final byte version = headerBuf.get(); + if (version != MASTER_FILE_VERSION_BUCKET) + throw new IOException("Unknown version: " + version); + + // compressionLevel and hashSeed are not used here + } + + // reads and decompresses one bucket record; null when the table entry is empty + private @Nullable ByteBuffer readBucketData(@NotNull FileChannel channel, long recordOffset) throws IOException { + if (recordOffset == 0) { + return null; + } + + final ByteBuffer lens = this.readRecordLengths(channel, recordOffset); + final int originalLen = lens.getInt(); + final int compressedLen = lens.getInt(); + + final byte[] compressedData = new byte[compressedLen]; + readFullyAt(channel, ByteBuffer.wrap(compressedData), recordOffset + V3_RECORD_HEADER_SIZE); + + return ByteBuffer.wrap(Zstd.decompress(compressedData, originalLen)); + } + private long @NonNull [] parseOffsetTable(FileChannel channel) throws IOException { final ByteBuffer buf = ByteBuffer.allocate(V3_POS_TABLE_SIZE); readFullyAt(channel, buf, V3_POS_TABLE_OFFSET); @@ -1162,10 +1343,8 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R final long[] table = new long[BUCKET_COUNT]; - Arrays.fill(table, 0L); for (int i = 0; i < BUCKET_COUNT; i++) { - final long pos = buf.getLong(); - table[i] = pos; + table[i] = buf.getLong(); } return table; @@ -1253,12 +1432,7 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R bucketBuffer.get(chunkData); // Mark bucket as loaded. writeChunk() bumps the bucket epoch so it gets synced to the new master format. - final int blinearBucketIndex = chunkIndex >> BUCKET_SHIFT; - final Bucket bucket = BufferedLinearRegionFile.this.buckets[blinearBucketIndex]; - - synchronized (bucket.lock) { - bucket.loaded = true; - } + BufferedLinearRegionFile.this.markBucketLoaded(chunkIndex); // Use writeChunk to go through the full path (adds length + timestamp + xxhash header) BufferedLinearRegionFile.this.writeChunk(chunkX, chunkZ, ByteBuffer.wrap(chunkData)); @@ -1305,13 +1479,7 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R final ByteBuffer sectorDataNioBuffer = ByteBuffer.wrap(sectorData); - final int bucketIndex = index >> BUCKET_SHIFT; - final Bucket bucket = BufferedLinearRegionFile.this.buckets[bucketIndex]; - - synchronized (bucket.lock) { - bucket.loaded = true; - } - + BufferedLinearRegionFile.this.markBucketLoaded(index); BufferedLinearRegionFile.this.writeChunkDataRaw(index, sectorDataNioBuffer, false); } } @@ -1359,14 +1527,7 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R final int x = posByAxis[0]; final int z = posByAxis[1]; - - final int bucketIndex = i >> BUCKET_SHIFT; - final Bucket bucket = BufferedLinearRegionFile.this.buckets[bucketIndex]; - - synchronized (bucket.lock) { - bucket.loaded = true; - } - + BufferedLinearRegionFile.this.markBucketLoaded(i); BufferedLinearRegionFile.this.writeChunk(x, z, chunkDataNioBuffer); } }