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 df07737..2555fb9 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
@@ -40,8 +40,15 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
* {@code masterFileLock} — master file read / append / replace
* {@code regionObjectLock} — in-memory sector table + swap file channel
*
- * The atomic flags (closed / synced / beingSynced / lastWritten) and the bucket
- * epochs are lock-free and may be touched while holding any (or no) lock.
+ * The atomic flags (closed / synced / beingSynced / lastWritten), the bucket epochs
+ * and the swap space counters (currentAcquiredIndex / liveBytes, mutated only under
+ * the region write lock) are lock-free readable and may be touched while holding any
+ * (or no) lock.
+ *
+ * The swap file is fully transient: it is deleted at open, opened with
+ * DELETE_ON_CLOSE and never parsed back after a crash, so it carries no header and
+ * is never fsynced. Durability comes exclusively from the master file, whose v3
+ * on-disk format is unchanged.
*/
public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.RegionFile {
private static final double SWAP_FILE_AUTO_COMPACT_PERCENT = 3.0 / 5.0; // 60 %
@@ -52,9 +59,7 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
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
+ private static final int XXHASH32_SEED = 0x0721; // ～(∠・ω< )⌒★
private static final long MASTER_FILE_SUPER_BLOCK = -0x200812250269L;
private static final byte MASTER_FILE_VERSION = 0x02; // ver 2.0
@@ -68,6 +73,24 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
private static final long MAX_SIZE_PER_CHUNK = RegionFile.MAX_CHUNK_SIZE;
+ // on-disk sector layout in the swap file:
+ // dataLen(int) + timestamp(long) + xxhash32(int) + lz4(chunk data)
+ // the 16 meta bytes stay OUTSIDE the compression so neither the write nor the read
+ // path needs a full-size intermediate copy of the chunk data; dataLen doubles as
+ // the lz4 original size, so no separate length prefix is needed
+ private static final int SECTOR_META_SIZE = Integer.BYTES + Long.BYTES + Integer.BYTES;
+
+ // all three are stateless and thread-safe
+ private static final LZ4Compressor LZ4_COMPRESSOR = LZ4Factory.fastestInstance().fastCompressor();
+ private static final LZ4FastDecompressor LZ4_DECOMPRESSOR = LZ4Factory.fastestInstance().fastDecompressor();
+ private static final XXHash32 XX_HASH_32 = XXHashFactory.fastestInstance().hash32();
+
+ // per-thread staging buffer for the hot chunk read/write paths: the compressed
+ // bytes never outlive the single pread/pwrite they are staged for, so they never
+ // need to escape into a fresh allocation
+ private static final int SCRATCH_RETAIN_LIMIT = 2 * 1024 * 1024; // 2 MiB
+ private static final ThreadLocal SCRATCH = ThreadLocal.withInitial(() -> ByteBuffer.allocate(64 * 1024));
+
private static final StandardOpenOption[] SWAP_FILE_CHANNEL_OPTIONS = new StandardOpenOption[]{
StandardOpenOption.CREATE,
StandardOpenOption.WRITE,
@@ -93,15 +116,16 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
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];
- private long currentAcquiredIndex = this.headerSize();
- private int xxHash32Seed = SWAP_FILE_HASH_SEED;
private FileChannel swapFileChannel;
+ // mutated only under regionObjectLock's write lock; volatile so flushInternal()
+ // can run its garbage estimate without taking any lock at all
+ private volatile long currentAcquiredIndex;
+ private volatile long liveBytes;
+
private final byte compressionLevel;
private final MasterFileParser masterFileParser = new MasterFileParser();
- private final CompressingOps compressingOps = new CompressingOps();
// managed by VarHandles following
private boolean closed = false;
@@ -132,10 +156,31 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
this.initSwapFile();
this.tryLoadOldBlinearMasterFileData();
+ // resume WAL mode directly from an existing v3 master file: without this, the
+ // first sync after every open rewrites the whole file even for one dirty chunk
+ this.masterFileParser.tryEnterWalMode(this.masterFilePath);
+
this.flusher = flusher;
this.flusher.addFile(this);
}
+ private static @NotNull ByteBuffer acquireScratch(int capacity) {
+ ByteBuffer buf = SCRATCH.get();
+
+ if (buf.capacity() < capacity) {
+ buf = ByteBuffer.allocate(Math.max(capacity, buf.capacity() << 1));
+
+ // oversized one-off requests get a throwaway buffer instead of pinning
+ // megabytes onto every io thread forever
+ if (buf.capacity() <= SCRATCH_RETAIN_LIMIT) {
+ SCRATCH.set(buf);
+ }
+ }
+
+ buf.clear();
+ return buf;
+ }
+
private static void writeFullyAt(FileChannel channel, @NonNull ByteBuffer buf, long startOffset) throws IOException {
long offset = startOffset;
while (buf.hasRemaining()) {
@@ -182,6 +227,10 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
private void cleanUpSwapFile() throws IOException {
Files.deleteIfExists(this.swapFilePath);
+
+ // a crash between compact's tmp creation and the atomic replace leaves a stale
+ // .swp.tmp behind, which would make every future compact fail at CREATE_NEW
+ Files.deleteIfExists(Path.of(this.swapFilePath + ".tmp"));
}
private void ensureBucketLoaded(int chunkIndex) throws IOException {
@@ -309,99 +358,69 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
SWAP_FILE_CHANNEL_OPTIONS
);
- // fill default sectors
+ // fill default sectors; the swap file has no header, data starts at offset 0
for (int i = 0; i < 1024; i++) {
- this.sectors[i] = new Sector(i, this.headerSize(), 0);
+ this.sectors[i] = new Sector(i, 0, 0);
}
+
+ this.currentAcquiredIndex = 0;
+ this.liveBytes = 0;
}
- private void recalculateAcquiredIndex() {
- long newValue = this.headerSize();
+ private void recalculateCounters() {
+ long acquired = 0;
+ long live = 0;
for (Sector sector : this.sectors) {
+ // cleared sectors keep their stale extent for in-place reuse (see store()),
+ // so their extent MUST still be counted into the acquired watermark here,
+ // or later appends could land inside it and get overwritten by a reuse
+ acquired = Math.max(acquired, sector.offset + sector.length);
+
if (sector.hasData()) {
- newValue = Math.max(newValue, sector.offset + sector.length);
+ live += sector.length;
}
}
- this.currentAcquiredIndex = newValue;
- }
-
- private void writeSwapFileHeaders(boolean forceFile, boolean forceMeta) throws IOException {
- final ByteBuffer buffer = ByteBuffer.allocate(this.headerSize());
-
- buffer.putLong(SWAP_FILE_SUPER_BLOCK); // Magic
- buffer.put(SWAP_FILE_VERSION); // Version
- buffer.putInt(this.xxHash32Seed); // XXHash32 seed
- buffer.putLong(this.currentAcquiredIndex); // Acquired index
-
- for (Sector sector : this.sectors) {
- // encode each sector
- buffer.put(sector.getEncoded());
- }
-
- buffer.flip();
-
- writeFullyAt(this.swapFileChannel, buffer, 0);
-
- if (forceFile) {
- this.swapFileChannel.force(forceMeta);
- }
- }
-
- private int sectorSize() {
- return this.sectors.length * Sector.sizeOfSingle();
- }
-
- private int headerSize() {
- int result = 0;
-
- result += Long.BYTES; // Magic
- result += Byte.BYTES; // Version
- result += Integer.BYTES; // XXHash32 seed
- result += Long.BYTES; // Acquired index
- result += this.sectorSize(); // Sectors
-
- return result;
+ this.currentAcquiredIndex = acquired;
+ this.liveBytes = live;
}
private void flushInternal() throws IOException {
- boolean initiallySyncRequired;
-
- this.regionObjectLock.writeLock().lock();
- try {
- if (this.isClosedRaw()) {
- return;
- }
-
- long liveSize = 0;
- for (Sector sector : this.sectors) {
- // skip no data sectors
- if (!sector.hasData()) {
- continue;
- }
-
- liveSize += sector.length;
- }
-
- // everything acquired but not covered by a live sector is garbage
- final long spareSize = this.currentAcquiredIndex - this.headerSize() - liveSize;
-
- 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) {
- // do compact
- this.compactSwapFile();
- }
-
- // prevent syncing after compact because it could be time costing sometimes
- initiallySyncRequired = !Files.exists(this.masterFilePath) && !compactRequested;
- } finally {
- this.regionObjectLock.writeLock().unlock();
+ if (this.isClosedRaw()) {
+ return;
}
- if (initiallySyncRequired) {
+ // lock-free garbage estimate from the incrementally maintained counters:
+ // this runs after EVERY chunk write, so no write lock, no O(1024) sector
+ // scan and no Files.exists() stat on the hot path
+ final long live = this.liveBytes;
+ final long spare = this.currentAcquiredIndex - live;
+ final boolean compactRequested = spare > SWAP_FILE_AUTO_COMPACT_SIZE && (double) spare > (double) live * SWAP_FILE_AUTO_COMPACT_PERCENT;
+
+ // try auto compact to clean the garbage area
+ if (compactRequested) {
+ this.regionObjectLock.writeLock().lock();
+ try {
+ if (!this.isClosedRaw()) {
+ // recheck with the authoritative values under the lock
+ final long liveNow = this.liveBytes;
+ final long spareNow = this.currentAcquiredIndex - liveNow;
+
+ if (spareNow > SWAP_FILE_AUTO_COMPACT_SIZE && (double) spareNow > (double) liveNow * SWAP_FILE_AUTO_COMPACT_PERCENT) {
+ // do compact
+ this.compactSwapFile();
+ }
+ }
+ } finally {
+ this.regionObjectLock.writeLock().unlock();
+ }
+ }
+
+ // create the master file eagerly on the very first write of a fresh region;
+ // afterwards this is a single volatile read per chunk write.
+ // prevent syncing after compact because it could be time costing sometimes
+ if (!compactRequested && !this.masterFileParser.masterFileExists()) {
this.syncToMasterFile(false, false);
}
}
@@ -462,8 +481,6 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
}
private void compactSwapFile() throws IOException {
- this.writeSwapFileHeaders(true, true); // save headers for compact
-
final Sector[] newSectorsToBeReplaced = new Sector[this.sectors.length];
for (int i = 0; i < this.sectors.length; i++) {
@@ -477,7 +494,7 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
// note:
// we reset length to 0 and this would make length <= newLength(which is >= 0) is always true.
// so that the following write operation wouldn't override the data of other sectors
- // see the write method in Sector class
+ // see the store method in Sector class
newSectorsToBeReplaced[i] = new Sector(i, 0, 0);
}
@@ -492,7 +509,7 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
StandardOpenOption.READ,
StandardOpenOption.TRUNCATE_EXISTING
)) {
- long offsetPointer = this.headerSize();
+ long offsetPointer = 0;
for (Sector sector : newSectorsToBeReplaced) {
// skip cleared or no data-contained sectors
@@ -511,12 +528,13 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
newSectorsToBeReplaced[sector.index] = newRecalculated; // update sector infos
}
- tempChannel.force(true);
+ // note: NO force here — the swap file is transient and never read back
+ // after a crash, so fsyncing it (twice, like before) was pure overhead
newAcquiredIndex = offsetPointer;
} catch (Throwable ex) {
- // recalculate acquired index
- this.recalculateAcquiredIndex();
+ // recalculate counters
+ this.recalculateCounters();
// delete the target temp file
Files.deleteIfExists(targetTemp);
// fast-fail
@@ -530,8 +548,8 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
try {
atomicReplace(targetTemp, this.swapFilePath);
} catch (Throwable e) {
- // recalculate acquired index
- this.recalculateAcquiredIndex();
+ // recalculate counters
+ this.recalculateCounters();
// reopen closed channel
this.reopenSwapFileChannel();
// fast-fail
@@ -543,12 +561,10 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
// reopen file channel
this.reopenSwapFileChannel();
- // replace with recalculated file headers
+ // replace with recalculated infos: after a compact everything left is live
this.sectors = newSectorsToBeReplaced;
this.currentAcquiredIndex = newAcquiredIndex;
-
- // flush to file
- this.writeSwapFileHeaders(true, true);
+ this.liveBytes = newAcquiredIndex;
} catch (Throwable ex) {
// we are totally failed here,
// directly mark as closed as the swap file is already replaced, and we failed to update the
@@ -571,14 +587,13 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
);
}
- private void writeChunkDataRaw(int index, ByteBuffer chunkData, boolean skipSync) throws IOException {
- final ByteBuffer committed = this.compressingOps.compress(chunkData); // run compression out of lock
-
+ // stores an already lz4-encoded sector (meta + compressed data), typically staged
+ // in the thread-local scratch: nothing here escapes to the heap
+ private void storeSector(int index, @NotNull ByteBuffer encoded, boolean skipSync) throws IOException {
this.regionObjectLock.writeLock().lock();
try {
- final Sector sector = this.sectors[index];
+ this.sectors[index].store(encoded, this.swapFileChannel);
- sector.store(committed, this.swapFileChannel);
if (!skipSync) {
this.markBucketDirty(index);
}
@@ -593,23 +608,26 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
this.markAsToSync();
}
- private @Nullable ByteBuffer readChunkDataRaw(int index) throws IOException {
- final ByteBuffer raw;
-
- this.regionObjectLock.readLock().lock();
- try {
- final Sector sector = this.sectors[index];
-
- if (!sector.hasData()) {
- return null;
- }
-
- raw = sector.read(this.swapFileChannel);
- } finally {
- this.regionObjectLock.readLock().unlock();
+ // section = dataLen(int) + timestamp(long) + xxhash32(int) + data, i.e. the exact
+ // per-chunk byte layout persisted inside master file bucket records
+ private void writeSection(int index, @NotNull ByteBuffer section, boolean skipSync) throws IOException {
+ if (section.remaining() < SECTOR_META_SIZE) {
+ throw new IOException("Truncated chunk section (" + section.remaining() + " bytes) for index " + index);
}
- return this.compressingOps.decompress(raw);
+ final int dataLen = section.remaining() - SECTOR_META_SIZE;
+ final ByteBuffer out = acquireScratch(SECTOR_META_SIZE + LZ4_COMPRESSOR.maxCompressedLength(dataLen));
+
+ // meta bytes are carried over verbatim, only the chunk data goes through lz4
+ final int oldLimit = section.limit();
+ section.limit(section.position() + SECTOR_META_SIZE);
+ out.put(section);
+ section.limit(oldLimit);
+
+ LZ4_COMPRESSOR.compress(section, out);
+ out.flip();
+
+ this.storeSector(index, out, skipSync);
}
private void clearChunkData(int index) throws IOException {
@@ -617,9 +635,7 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
this.regionObjectLock.writeLock().lock();
try {
- final Sector sector = this.sectors[index];
-
- sector.clear();
+ this.sectors[index].clear();
this.markBucketDirty(index);
} finally {
this.regionObjectLock.writeLock().unlock();
@@ -653,24 +669,26 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
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());
+ final int dataLen = data.remaining();
+
+ if (dataLen > MAX_SIZE_PER_CHUNK) {
+ throw new RegionFileStorage.RegionFileSizeException("Writing too large chunk, limit : " + MAX_SIZE_PER_CHUNK + " but got : " + dataLen);
}
- final int oldPositionOfData = data.position();
- final int xxHash32OfData = this.xxHash32.hash(data, this.xxHash32Seed);
- data.position(oldPositionOfData);
+ // absolute-offset hash: no position save/restore dance needed
+ final int xxHash32OfData = XX_HASH_32.hash(data, data.position(), dataLen, XXHASH32_SEED);
- // uncompressed length(int) + timestamp(long) + xxhash32(int)
- final ByteBuffer chunkSectionBuilder = ByteBuffer.allocate(data.remaining() + 4 + 8 + 4);
+ // meta + compressed data are built directly in the reusable scratch: no
+ // full-size intermediate copy of the chunk data, no allocation that escapes
+ final ByteBuffer out = acquireScratch(SECTOR_META_SIZE + LZ4_COMPRESSOR.maxCompressedLength(dataLen));
- chunkSectionBuilder.putInt(data.remaining()); // Length(int)
- chunkSectionBuilder.putLong(System.currentTimeMillis()); // Timestamp(long)
- chunkSectionBuilder.putInt(xxHash32OfData); // xxHash32 of the original data(int)
- chunkSectionBuilder.put(data); // Data(bytes)
- chunkSectionBuilder.flip();
+ out.putInt(dataLen); // uncompressed length, doubles as the lz4 original size
+ out.putLong(System.currentTimeMillis()); // timestamp
+ out.putInt(xxHash32OfData); // xxHash32 of the original data
+ LZ4_COMPRESSOR.compress(data, out);
+ out.flip();
- this.writeChunkDataRaw(chunkIndex, chunkSectionBuilder, false);
+ this.storeSector(chunkIndex, out, false);
}
private @Nullable ByteBuffer readChunk(int x, int z) throws IOException {
@@ -678,34 +696,42 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
this.ensureBucketLoaded(chunkIndex);
- final ByteBuffer data = this.readChunkDataRaw(chunkIndex);
+ final ByteBuffer stage;
- if (data == null) {
- return null;
+ this.regionObjectLock.readLock().lock();
+ try {
+ final Sector sector = this.sectors[chunkIndex];
+
+ if (!sector.hasData()) {
+ return null;
+ }
+
+ // only the pread runs under the lock, staged into the reusable scratch
+ stage = acquireScratch((int) sector.length);
+ stage.limit((int) sector.length);
+
+ readFullyAt(this.swapFileChannel, stage, sector.offset);
+ } finally {
+ this.regionObjectLock.readLock().unlock();
}
- final int length = data.getInt(); // compressed length(int)
- final long timestamp = data.getLong(); // TODO use this timestamp(long) for something?
- final int dataXXHash32 = data.getInt(); // XXHash32 for validation(int)
+ stage.flip();
- final IOException xxHash32CheckFailedEx = this.checkXXHash32(dataXXHash32, data);
- if (xxHash32CheckFailedEx != null) {
- throw xxHash32CheckFailedEx; // prevent from loading
+ final int dataLen = stage.getInt();
+ stage.getLong(); // TODO use this timestamp(long) for something?
+ final int expectedXXHash32 = stage.getInt();
+
+ // lz4 decompresses straight from the scratch into the result buffer: the
+ // compressed bytes are never copied into an intermediate array
+ final byte[] data = new byte[dataLen];
+ LZ4_DECOMPRESSOR.decompress(stage.array(), stage.arrayOffset() + SECTOR_META_SIZE, data, 0, dataLen);
+
+ final int actualXXHash32 = XX_HASH_32.hash(data, 0, dataLen, XXHASH32_SEED);
+ if (actualXXHash32 != expectedXXHash32) {
+ throw new IOException("XXHash32 check failed ! Expected: " + expectedXXHash32 + ",but got: " + actualXXHash32); // prevent from loading
}
- return data;
- }
-
- private @Nullable IOException checkXXHash32(long originalXXHash32, @NotNull ByteBuffer input) {
- final int oldPositionOfInput = input.position();
- final int currentXXHash32 = this.xxHash32.hash(input, this.xxHash32Seed);
- input.position(oldPositionOfInput);
-
- if (originalXXHash32 != currentXXHash32) {
- return new IOException("XXHash32 check failed ! Expected: " + originalXXHash32 + ",but got: " + currentXXHash32);
- }
-
- return null;
+ return ByteBuffer.wrap(data);
}
@Override
@@ -821,34 +847,6 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
}
}
- // here we use this tool to prevent the swap file goes too large
- // sometimes when a region contains all chunks, it might be very huge without any compressions(around 100MiB)
- private static class CompressingOps {
- private final LZ4Compressor lz4Compressor = LZ4Factory.fastestInstance().fastCompressor();
- private final LZ4FastDecompressor lz4Decompressor = LZ4Factory.fastestInstance().fastDecompressor();
-
- public @NotNull ByteBuffer compress(@NotNull ByteBuffer in) {
- final int bufferLenToAllocate = this.lz4Compressor.maxCompressedLength(in.remaining());
- final ByteBuffer result = ByteBuffer.allocate(bufferLenToAllocate + 4);
-
- result.putInt(in.remaining());
- this.lz4Compressor.compress(in, result);
-
- return result.flip();
- }
-
- public @NotNull ByteBuffer decompress(@NotNull ByteBuffer flippedIn) {
- final int originalLen = flippedIn.getInt();
- final byte[] raw = new byte[flippedIn.remaining()];
- flippedIn.get(raw);
-
- final byte[] decompressed = new byte[originalLen];
- this.lz4Decompressor.decompress(raw, decompressed);
-
- return ByteBuffer.wrap(decompressed);
- }
- }
-
public class Sector {
private final int index;
private long offset;
@@ -861,76 +859,51 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
this.length = length;
}
- public @NotNull ByteBuffer read(@NotNull FileChannel channel) throws IOException {
- final ByteBuffer result = ByteBuffer.allocate((int) this.length);
-
- readFullyAt(channel, result, this.offset);
-
- result.flip();
- return result;
- }
-
public void store(@NotNull ByteBuffer newData, @NotNull FileChannel channel) throws IOException {
final long oldLength = this.length;
+ final long oldLive = this.hasData ? oldLength : 0L;
final long newDataLength = newData.remaining();
this.hasData = true;
this.length = newDataLength;
- // data is smaller or its length equals to the local buffer we hold, write it directly
+ // data fits into the extent this sector already owns (a cleared sector keeps
+ // its stale extent exactly for this reuse), write it in place
if (newDataLength <= oldLength) {
writeFullyAt(channel, newData, this.offset);
+ } else {
+ // or we will append to the end of file
+ this.offset = BufferedLinearRegionFile.this.currentAcquiredIndex;
+ BufferedLinearRegionFile.this.currentAcquiredIndex = this.offset + newDataLength;
- return;
+ writeFullyAt(channel, newData, this.offset);
}
- // or we will append to the end of file
- this.offset = BufferedLinearRegionFile.this.currentAcquiredIndex;
-
- BufferedLinearRegionFile.this.currentAcquiredIndex += this.length;
-
- writeFullyAt(channel, newData, this.offset);
- }
-
- private @NotNull ByteBuffer getEncoded() {
- final ByteBuffer buffer = ByteBuffer.allocate(sizeOfSingle());
-
- buffer.putLong(this.offset);
- buffer.putLong(this.length);
- buffer.put((byte) (this.hasData ? 1 : 0));
- buffer.flip();
-
- return buffer;
- }
-
- public void restoreFrom(@NotNull ByteBuffer buffer) {
- this.offset = buffer.getLong();
- this.length = buffer.getLong();
- this.hasData = buffer.get() == 1;
-
- if (this.length < 0 || this.offset < 0) {
- throw new IllegalStateException("Invalid sector data: " + this);
- }
+ // single mutator under the region write lock; keeps the garbage estimate
+ // in flushInternal() lock-free and scan-free
+ BufferedLinearRegionFile.this.liveBytes += newDataLength - oldLive;
}
public void clear() {
+ if (this.hasData) {
+ BufferedLinearRegionFile.this.liveBytes -= this.length;
+ }
+
this.hasData = false;
}
public boolean hasData() {
return this.hasData;
}
-
- static int sizeOfSingle() {
- // offset + length hasData
- return Long.BYTES * 2 + 1;
- }
}
private class ChunkBufferHelper extends ByteArrayOutputStream {
private final ChunkPos pos;
private ChunkBufferHelper(ChunkPos pos) {
+ // chunk NBT payloads are tens to hundreds of KiB: BAOS's default 32 bytes
+ // means a dozen grow-and-copy rounds per single chunk serialization
+ super(8192);
this.pos = pos;
}
@@ -945,7 +918,7 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
}
private class MasterFileParser {
- // V3 bucketed format layout:
+ // V3 bucketed format layout (UNCHANGED, fully compatible with existing files):
// [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
// [142, EOF): bucket records — originalLen(int) + compressedLen(int) + compressedData
@@ -957,9 +930,10 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
private final ReadWriteLock masterFileLock = new ReentrantReadWriteLock();
- // 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.
+ // WAL(append) state, guarded by masterFileLock: non-null whenever a valid v3
+ // master file is open for appending — restored directly at open time by
+ // tryEnterWalMode(), or (re)established by rewriteFully(); syncs then 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;
@@ -967,16 +941,103 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
private long[] recordSizes;
private long appendOffset;
+ // single volatile read instead of a Files.exists() stat per chunk write
+ private volatile boolean fileExists;
+
// 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) {
+ private record BucketRecord(long epoch, @Nullable ByteBuffer payload) {
+ }
+
+ public boolean masterFileExists() {
+ return this.fileExists;
+ }
+
+ // resumes WAL mode from an existing, structurally valid v3 master file so the
+ // first sync after open can append instead of rewriting the entire file.
+ // bails out silently (leaving the full-rewrite path armed) if the file is
+ // missing, not v3, or its position table doesn't validate
+ public void tryEnterWalMode(@NotNull Path mainFile) throws IOException {
+ this.masterFileLock.writeLock().lock();
+ try {
+ // legacy migration in tryParseMainFileOld() may have entered WAL already
+ if (this.appendChannel != null) {
+ return;
+ }
+
+ if (!Files.exists(mainFile)) {
+ return;
+ }
+
+ this.fileExists = true;
+
+ final FileChannel channel = FileChannel.open(mainFile, StandardOpenOption.READ, StandardOpenOption.WRITE);
+ boolean success = false;
+ try {
+ final long fileSize = channel.size();
+
+ if (fileSize < V3_DATA_AREA_OFFSET) {
+ return;
+ }
+
+ 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;
+ }
+
+ final long[] table = this.parseOffsetTable(channel);
+ final long[] sizes = new long[BUCKET_COUNT];
+ long dataEnd = V3_DATA_AREA_OFFSET;
+
+ for (int i = 0; i < BUCKET_COUNT; i++) {
+ final long recordOffset = table[i];
+
+ if (recordOffset == 0) {
+ continue;
+ }
+
+ if (recordOffset < V3_DATA_AREA_OFFSET || recordOffset + V3_RECORD_HEADER_SIZE > fileSize) {
+ return; // corrupted table: stay in full-rewrite mode
+ }
+
+ final ByteBuffer lens = this.readRecordLengths(channel, recordOffset);
+ final int originalLen = lens.getInt();
+ final int compressedLen = lens.getInt();
+
+ if (originalLen < 0 || compressedLen < 0 || recordOffset + V3_RECORD_HEADER_SIZE + compressedLen > fileSize) {
+ return; // corrupted record header: stay in full-rewrite mode
+ }
+
+ sizes[i] = V3_RECORD_HEADER_SIZE + (long) compressedLen;
+ dataEnd = Math.max(dataEnd, recordOffset + sizes[i]);
+ }
+
+ // append after the last referenced record: anything past that is
+ // uncommitted garbage from a torn previous append and may be reused
+ this.appendChannel = channel;
+ this.positionTable = table;
+ this.recordSizes = sizes;
+ this.appendOffset = dataEnd;
+ success = true;
+ } finally {
+ if (!success) {
+ channel.close();
+ }
+ }
+ } finally {
+ this.masterFileLock.writeLock().unlock();
+ }
}
// must be called under syncLock (see syncToMasterFile)
public void sync(@NotNull Path mainFile, boolean forceCompact) throws IOException {
this.masterFileLock.writeLock().lock();
try {
- // full rewrite on the first sync after open, and afterwards whenever the
+ // full rewrite whenever no valid append state exists (fresh region /
+ // corrupted table / legacy migration), 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() || forceCompact) {
@@ -1003,22 +1064,28 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
}
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 boolean wal = 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;
- // 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);
+ FileChannel legacySource = null;
+ try {
+ final FileChannel oldChannel;
+ final long[] oldPositionTable;
+
+ if (wal) {
+ // reuse the live append channel as the copy source together with the
+ // cached table/sizes: no reopen and no per-bucket length pread needed
+ oldChannel = this.appendChannel;
+ oldPositionTable = this.positionTable;
+ } else {
+ legacySource = this.openV3MasterFile(mainFile);
+ oldChannel = legacySource;
+ oldPositionTable = oldChannel == null ? null : this.parseOffsetTable(oldChannel);
+ }
try (FileChannel outChannel = FileChannel.open(tmpFilePath,
StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING)) {
@@ -1034,10 +1101,12 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
final BucketRecord record = this.buildBucketRecord(bucketIndex);
if (record.payload() != null) {
- writeFullyAt(outChannel, ByteBuffer.wrap(record.payload()), dataOffset);
+ final int recordSize = record.payload().remaining();
+
+ writeFullyAt(outChannel, record.payload(), dataOffset);
newPositionTable[bucketIndex] = dataOffset;
- newRecordSizes[bucketIndex] = record.payload().length;
- dataOffset += record.payload().length;
+ newRecordSizes[bucketIndex] = recordSize;
+ dataOffset += recordSize;
}
// else: the bucket is empty now, its table entry stays 0
@@ -1045,9 +1114,15 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
} 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 recordSize;
+
+ if (wal) {
+ recordSize = this.recordSizes[bucketIndex];
+ } else {
+ final ByteBuffer lens = this.readRecordLengths(oldChannel, oldOffset);
+ lens.getInt(); // skip originalLen
+ recordSize = V3_RECORD_HEADER_SIZE + (long) lens.getInt();
+ }
transferFully(oldChannel, oldOffset, recordSize, outChannel, dataOffset);
newPositionTable[bucketIndex] = dataOffset;
@@ -1063,15 +1138,39 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
newAppendOffset = dataOffset;
}
+ } catch (Throwable e) {
+ // don't leak the half-written tmp file; in WAL mode the append state is
+ // untouched so the next sync just retries the compact, in legacy mode
+ // the next sync retries this full-rewrite path
+ try {
+ Files.deleteIfExists(tmpFilePath);
+ } catch (Throwable e2) {
+ e.addSuppressed(e2);
+ }
+
+ throw e instanceof IOException io ? io : new IOException("Failed to rewrite master file!", e);
+ } finally {
+ if (legacySource != null) {
+ legacySource.close();
+ }
+ }
+
+ // close the append channel before the replace: some platforms (windows)
+ // refuse to replace a file that still has open handles
+ if (wal) {
+ final FileChannel toClose = this.appendChannel;
+ this.appendChannel = null; // if close() throws, fall back to full rewrite next sync
+ toClose.close();
}
atomicReplace(tmpFilePath, mainFile);
- // enter WAL mode: keep the freshly written master file open for appending syncs
+ // (re)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.fileExists = true;
this.markBucketsSynced(syncedBucketEpochs);
}
@@ -1081,7 +1180,9 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
final long[] syncedBucketEpochs = new long[BUCKET_COUNT];
final long[] newPositionTable = this.positionTable.clone();
final long[] newRecordSizes = this.recordSizes.clone();
+ final ByteBuffer[] pending = new ByteBuffer[BUCKET_COUNT];
long dataOffset = this.appendOffset;
+ int pendingCount = 0;
boolean anyDirty = false;
for (int bucketIndex = 0; bucketIndex < BUCKET_COUNT; bucketIndex++) {
@@ -1090,12 +1191,13 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
}
final BucketRecord record = this.buildBucketRecord(bucketIndex);
+ final ByteBuffer payload = record.payload();
- if (record.payload() != null) {
- writeFullyAt(channel, ByteBuffer.wrap(record.payload()), dataOffset);
+ if (payload != null) {
+ pending[pendingCount++] = payload;
newPositionTable[bucketIndex] = dataOffset;
- newRecordSizes[bucketIndex] = record.payload().length;
- dataOffset += record.payload().length;
+ newRecordSizes[bucketIndex] = payload.remaining();
+ dataOffset += payload.remaining();
} else {
// the bucket is empty now
newPositionTable[bucketIndex] = 0;
@@ -1110,8 +1212,19 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
return;
}
- // make the appended records durable before the position table may point at them
- channel.force(false);
+ if (pendingCount > 0) {
+ // all records land contiguously at the tail: one gathering write (writev)
+ // instead of one pwrite per dirty bucket
+ channel.position(this.appendOffset);
+
+ final ByteBuffer last = pending[pendingCount - 1];
+ while (last.hasRemaining()) {
+ channel.write(pending, 0, pendingCount);
+ }
+
+ // 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
@@ -1126,12 +1239,19 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
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
+ // snapshots one bucket under a short read lock (raw sector bytes only, with
+ // sectors that sit back to back in the swap file coalesced into single preads);
+ // 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 ByteBuffer[] rawSectors = new ByteBuffer[BUCKET_SIZE]; // slices into run buffers, null = no data
+
+ final long[] offsets = new long[BUCKET_SIZE];
+ final long[] lengths = new long[BUCKET_SIZE];
+ final int[] slots = new int[BUCKET_SIZE];
+ int liveCount = 0;
+
final long epoch;
BufferedLinearRegionFile.this.regionObjectLock.readLock().lock();
@@ -1143,48 +1263,119 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
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;
+ if (!sector.hasData()) {
+ continue;
+ }
+
+ offsets[liveCount] = sector.offset;
+ lengths[liveCount] = sector.length;
+ slots[liveCount] = i;
+ liveCount++;
+ }
+
+ if (liveCount == 0) {
+ return new BucketRecord(epoch, null);
+ }
+
+ sortByOffset(offsets, lengths, slots, liveCount);
+
+ int i = 0;
+ while (i < liveCount) {
+ int j = i;
+ long runEnd = offsets[i] + lengths[i];
+
+ while (j + 1 < liveCount && offsets[j + 1] == runEnd) {
+ j++;
+ runEnd += lengths[j];
+ }
+
+ final ByteBuffer run = ByteBuffer.allocate((int) (runEnd - offsets[i]));
+ readFullyAt(BufferedLinearRegionFile.this.swapFileChannel, run, offsets[i]);
+
+ for (int k = i; k <= j; k++) {
+ rawSectors[slots[k]] = run.slice((int) (offsets[k] - offsets[i]), (int) lengths[k]);
+ }
+
+ i = j + 1;
}
} finally {
BufferedLinearRegionFile.this.regionObjectLock.readLock().unlock();
}
- final ByteArrayOutputStream rawBuf = new ByteArrayOutputStream();
- final DataOutputStream rawOut = new DataOutputStream(rawBuf);
- boolean hasAny = false;
+ // exact size budget up front: 4 bytes size prefix per chunk slot plus
+ // meta + decompressed data for the live ones — one allocation, no growing
+ // ByteArrayOutputStream and no toByteArray() copy at the end
+ int sectionSize = BUCKET_SIZE * Integer.BYTES;
+ for (int i = 0; i < BUCKET_SIZE; i++) {
+ final ByteBuffer raw = rawSectors[i];
+
+ if (raw != null) {
+ sectionSize += SECTOR_META_SIZE + raw.getInt(raw.position());
+ }
+ }
+
+ final byte[] section = new byte[sectionSize];
+ final ByteBuffer sectionBuf = ByteBuffer.wrap(section);
for (int i = 0; i < BUCKET_SIZE; i++) {
- final ByteBuffer rawSector = rawSectors[i];
+ final ByteBuffer raw = rawSectors[i];
// note: null -> no data contained
- if (rawSector == null) {
- rawOut.writeInt(0);
+ if (raw == null) {
+ sectionBuf.putInt(0);
continue;
}
- final ByteBuffer chunkData = BufferedLinearRegionFile.this.compressingOps.decompress(rawSector);
- final byte[] arr = new byte[chunkData.remaining()];
- chunkData.get(arr);
+ final byte[] runArray = raw.array();
+ final int rawBase = raw.arrayOffset() + raw.position();
+ final int dataLen = raw.getInt(raw.position());
- rawOut.writeInt(arr.length);
- rawOut.write(arr);
- hasAny = true;
- }
- rawOut.flush();
+ sectionBuf.putInt(SECTOR_META_SIZE + dataLen);
+ sectionBuf.put(runArray, rawBase, SECTOR_META_SIZE); // meta bytes carried over verbatim
- if (!hasAny) {
- return new BucketRecord(epoch, null);
+ // lz4 decompresses straight into the section buffer, no intermediate arrays
+ final int destPos = sectionBuf.position();
+ LZ4_DECOMPRESSOR.decompress(runArray, rawBase + SECTOR_META_SIZE, section, destPos, dataLen);
+ sectionBuf.position(destPos + dataLen);
}
- final byte[] raw = rawBuf.toByteArray();
- final byte[] compressed = Zstd.compress(raw, BufferedLinearRegionFile.this.compressionLevel);
+ // zstd compresses straight into the final payload: skips Zstd.compress()'s
+ // internal bound-sized temp array plus its exact-size copy at the end
+ final int bound = (int) Zstd.compressBound(sectionSize);
+ final byte[] payload = new byte[V3_RECORD_HEADER_SIZE + bound];
+ final long compressedLen = Zstd.compressByteArray(payload, V3_RECORD_HEADER_SIZE, bound, section, 0, sectionSize, 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);
+ if (Zstd.isError(compressedLen)) {
+ throw new IOException("Failed to zstd compress bucket " + bucketIndex + ": " + Zstd.getErrorName(compressedLen));
+ }
- return new BucketRecord(epoch, payload.array());
+ final ByteBuffer result = ByteBuffer.wrap(payload, 0, V3_RECORD_HEADER_SIZE + (int) compressedLen);
+ result.putInt(sectionSize); // original (uncompressed) length
+ result.putInt((int) compressedLen); // compressed length
+ result.position(0);
+
+ return new BucketRecord(epoch, result);
+ }
+
+ private static void sortByOffset(long[] offsets, long[] lengths, int[] slots, int count) {
+ // n <= 64, insertion sort is plenty and allocation-free
+ for (int i = 1; i < count; i++) {
+ final long offset = offsets[i];
+ final long length = lengths[i];
+ final int slot = slots[i];
+ int j = i - 1;
+
+ while (j >= 0 && offsets[j] > offset) {
+ offsets[j + 1] = offsets[j];
+ lengths[j + 1] = lengths[j];
+ slots[j + 1] = slots[j];
+ j--;
+ }
+
+ offsets[j + 1] = offset;
+ lengths[j + 1] = length;
+ slots[j + 1] = slot;
+ }
}
private void markBucketsSynced(long[] syncedBucketEpochs) {
@@ -1233,7 +1424,7 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
header.putLong(MASTER_FILE_SUPER_BLOCK);
header.put(MASTER_FILE_VERSION_BUCKET);
header.put(BufferedLinearRegionFile.this.compressionLevel);
- header.putInt(BufferedLinearRegionFile.this.xxHash32Seed);
+ header.putInt(XXHASH32_SEED);
header.flip();
writeFullyAt(channel, header, 0);
@@ -1354,10 +1545,11 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
final int chunkSectionDataSize = decompressed.getInt();
if (chunkSectionDataSize <= 0) continue;
- final byte[] chunkSectionData = new byte[chunkSectionDataSize];
- decompressed.get(chunkSectionData);
+ // slice instead of copying the section bytes out
+ final ByteBuffer section = decompressed.slice(decompressed.position(), chunkSectionDataSize);
+ decompressed.position(decompressed.position() + chunkSectionDataSize);
- BufferedLinearRegionFile.this.writeChunkDataRaw(chunkIndex, ByteBuffer.wrap(chunkSectionData), true);
+ BufferedLinearRegionFile.this.writeSection(chunkIndex, section, true);
}
}
@@ -1479,7 +1671,8 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
final ByteBuffer sectorDataNioBuffer = ByteBuffer.wrap(sectorData);
BufferedLinearRegionFile.this.markBucketLoaded(index);
- BufferedLinearRegionFile.this.writeChunkDataRaw(index, sectorDataNioBuffer, false);
+ // blinear v2 stored the exact section layout, feed it through the section path
+ BufferedLinearRegionFile.this.writeSection(index, sectorDataNioBuffer, false);
}
}
}
@@ -1598,4 +1791,4 @@ public class BufferedLinearRegionFile implements io.nanachiyo0721.shiroha.data.R
throw new IOException("Unknown or unsupported super block : " + superBlock);
}
}
-}
\ No newline at end of file
+}