2026-07-14 12:22:17 +08:00
package io.nanachiyo0721.shiroha.data ;
2026-07-09 00:03:09 +08:00
import ca.spottedleaf.concurrentutil.util.ConcurrentUtil ;
import ca.spottedleaf.moonrise.patches.chunk_system.io.MoonriseRegionFileIO ;
import com.github.luben.zstd.Zstd ;
import com.github.luben.zstd.ZstdInputStream ;
import net.jpountz.lz4.LZ4Compressor ;
import net.jpountz.lz4.LZ4Factory ;
import net.jpountz.lz4.LZ4FastDecompressor ;
import net.jpountz.xxhash.XXHash32 ;
import net.jpountz.xxhash.XXHashFactory ;
import net.minecraft.nbt.CompoundTag ;
import net.minecraft.world.level.ChunkPos ;
import net.minecraft.world.level.chunk.storage.RegionFile ;
import net.minecraft.world.level.chunk.storage.RegionFileStorage ;
import net.openhft.hashing.LongHashFunction ;
import org.apache.commons.lang3.Validate ;
import org.jetbrains.annotations.Contract ;
import org.jetbrains.annotations.NotNull ;
import org.jetbrains.annotations.Nullable ;
import org.jspecify.annotations.NonNull ;
import java.io.* ;
import java.lang.invoke.VarHandle ;
import java.nio.ByteBuffer ;
import java.nio.channels.FileChannel ;
import java.nio.file.Files ;
import java.nio.file.Path ;
import java.nio.file.StandardCopyOption ;
import java.nio.file.StandardOpenOption ;
import java.util.concurrent.atomic.AtomicLong ;
import java.util.concurrent.locks.ReadWriteLock ;
import java.util.concurrent.locks.ReentrantReadWriteLock ;
2026-07-17 22:10:57 +08:00
/**
* Lock hierarchy (always acquire top to bottom, never the reverse):
* <ol>
* <li>{@code syncLock} — serializes master file syncs against close</li>
* <li>{@code Bucket.lock} — per-bucket lazy-load guard</li>
* <li>{@code masterFileLock} — master file read / append / replace</li>
* <li>{@code regionObjectLock} — in-memory sector table + swap file channel</li>
* </ol>
2026-07-26 22:30:02 +08:00
* 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.
* <p>
* 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.
2026-07-17 22:10:57 +08:00
*/
2026-07-14 12:22:17 +08:00
public class BufferedLinearRegionFile implements io . nanachiyo0721 . shiroha . data . RegionFile {
2026-07-09 00:03:09 +08:00
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
2026-07-17 22:10:57 +08:00
// 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 ;
2026-07-26 22:30:02 +08:00
private static final int XXHASH32_SEED = 0x0721 ; // ~ (∠・ω< )⌒★
2026-07-09 00:03:09 +08:00
private static final long MASTER_FILE_SUPER_BLOCK = - 0x200812250269L ;
private static final byte MASTER_FILE_VERSION = 0x02 ; // ver 2.0
private static final byte MASTER_FILE_VERSION_BUCKET = 0x03 ; // ver 3.0
private static final long LINEAR_FILE_SUPER_BLOCK = 0xc3ff13183cca9d9aL ;
private static final int BUCKET_SHIFT = 6 ;
private static final int BUCKET_SIZE = 1 << BUCKET_SHIFT ;
private static final int BUCKET_COUNT = 1024 / BUCKET_SIZE ;
private static final long MAX_SIZE_PER_CHUNK = RegionFile . MAX_CHUNK_SIZE ;
2026-07-26 22:30:02 +08:00
// 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
2026-08-03 20:01:42 +08:00
private static final LZ4Compressor LZ4_COMPRESSOR = LZ4Factory . fastestInstance (). highCompressor ();
2026-07-26 22:30:02 +08:00
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 < ByteBuffer > SCRATCH = ThreadLocal . withInitial (() -> ByteBuffer . allocate ( 64 * 1024 ));
2026-07-09 00:03:09 +08:00
private static final StandardOpenOption [] SWAP_FILE_CHANNEL_OPTIONS = new StandardOpenOption [] {
StandardOpenOption . CREATE ,
StandardOpenOption . WRITE ,
StandardOpenOption . READ ,
StandardOpenOption . DELETE_ON_CLOSE
};
private static final class Bucket {
private final Object lock = new Object ();
private final AtomicLong writeEpoch = new AtomicLong ();
private final AtomicLong syncedEpoch = new AtomicLong ();
private volatile boolean loaded = false ;
}
private final Bucket [] buckets = new Bucket [ BUCKET_COUNT ] ;
private final Path masterFilePath ;
private final Path swapFilePath ;
2026-07-17 22:10:57 +08:00
// 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 ();
2026-07-09 00:03:09 +08:00
private final ReadWriteLock regionObjectLock = new ReentrantReadWriteLock ();
private Sector [] sectors = new Sector [ 1024 ] ;
private FileChannel swapFileChannel ;
2026-07-26 22:30:02 +08:00
// 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 ;
2026-07-09 00:03:09 +08:00
private final byte compressionLevel ;
2026-07-24 21:52:44 +08:00
private final MasterFileParser masterFileParser = new MasterFileParser ();
2026-07-09 00:03:09 +08:00
// managed by VarHandles following
private boolean closed = false ;
private boolean beingSynced = false ;
private boolean synced = false ;
private long lastWritten = System . nanoTime ();
private static final VarHandle CLOSED_HANDLE = ConcurrentUtil . getVarHandle ( BufferedLinearRegionFile . class , "closed" , boolean . class );
private static final VarHandle SYNCED_HANDLE = ConcurrentUtil . getVarHandle ( BufferedLinearRegionFile . class , "synced" , boolean . class );
private static final VarHandle BEING_SYNCED_HANDLE = ConcurrentUtil . getVarHandle ( BufferedLinearRegionFile . class , "beingSynced" , boolean . class );
private static final VarHandle LAST_WRITTEN_HANDLE = ConcurrentUtil . getVarHandle ( BufferedLinearRegionFile . class , "lastWritten" , long . class );
private final BufferedLinearRegionFileFlusher flusher ;
public BufferedLinearRegionFile ( Path masterFilePath , int compressionLevel , @NotNull BufferedLinearRegionFileFlusher flusher ) throws IOException {
this . masterFilePath = masterFilePath ;
this . swapFilePath = Path . of ( this . masterFilePath . toString () + ".swp" );
Validate . inclusiveBetween ( 1 , 22 , compressionLevel );
for ( int i = 0 ; i < this . buckets . length ; i ++ ) {
this . buckets [ i ] = new Bucket ();
}
this . compressionLevel = ( byte ) compressionLevel ;
this . cleanUpSwapFile ();
this . initSwapFile ();
this . tryLoadOldBlinearMasterFileData ();
2026-07-26 22:30:02 +08:00
// 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 );
2026-07-09 00:03:09 +08:00
this . flusher = flusher ;
this . flusher . addFile ( this );
}
2026-07-26 22:30:02 +08:00
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 ;
}
2026-07-09 00:03:09 +08:00
private static void writeFullyAt ( FileChannel channel , @NonNull ByteBuffer buf , long startOffset ) throws IOException {
long offset = startOffset ;
while ( buf . hasRemaining ()) {
offset += channel . write ( buf , offset );
}
}
private static void readFullyAt ( FileChannel channel , @NonNull ByteBuffer buf , long startOffset ) throws IOException {
long offset = startOffset ;
while ( buf . hasRemaining ()) {
final int read = channel . read ( buf , offset );
if ( read < 0 ) throw new EOFException ( "Unexpected EOF at offset " + offset );
offset += read ;
}
}
2026-07-17 22:10:57 +08:00
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 );
}
}
}
2026-07-09 00:03:09 +08:00
private void cleanUpSwapFile () throws IOException {
Files . deleteIfExists ( this . swapFilePath );
2026-07-26 22:30:02 +08:00
// 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" ));
2026-07-09 00:03:09 +08:00
}
private void ensureBucketLoaded ( int chunkIndex ) throws IOException {
final int bucketIndex = chunkIndex >> BUCKET_SHIFT ;
final Bucket bucket = this . buckets [ bucketIndex ] ;
2026-07-17 22:10:57 +08:00
if ( bucket . loaded ) { // volatile fast path
return ;
}
2026-07-09 00:03:09 +08:00
// bucket lock -> master read lock -> swap write lock
synchronized ( bucket . lock ) {
if ( bucket . loaded ) {
return ;
}
this . masterFileParser . loadBucketsFor ( this . masterFilePath , bucketIndex );
bucket . loaded = true ;
}
}
2026-07-17 22:10:57 +08:00
// 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 ;
}
2026-07-09 00:03:09 +08:00
}
2026-07-17 22:10:57 +08:00
private void markBucketDirty ( int chunkIndex ) {
this . buckets [ chunkIndex >> BUCKET_SHIFT ] . writeEpoch . incrementAndGet ();
2026-07-09 00:03:09 +08:00
}
private long getBucketWriteEpoch ( int bucketIndex ) {
2026-07-17 22:10:57 +08:00
return this . buckets [ bucketIndex ] . writeEpoch . get ();
2026-07-09 00:03:09 +08:00
}
private void markBucketSynced ( int bucketIndex , long syncedEpoch ) {
2026-07-17 22:10:57 +08:00
this . buckets [ bucketIndex ] . syncedEpoch . accumulateAndGet ( syncedEpoch , Math :: max );
2026-07-09 00:03:09 +08:00
}
private boolean isBucketDirty ( int bucketIndex ) {
final Bucket bucket = this . buckets [ bucketIndex ] ;
return bucket . writeEpoch . get () != bucket . syncedEpoch . get ();
}
public boolean markAsBeingSynced () {
return BEING_SYNCED_HANDLE . compareAndSet ( this , false , true );
}
public long getLastWritten () {
return ( long ) LAST_WRITTEN_HANDLE . getVolatile ( this );
}
public boolean shouldSync () {
return ! (( boolean ) SYNCED_HANDLE . getVolatile ( this ));
}
public boolean softReadLock () {
// not done close logic yet
return this . regionObjectLock . readLock (). tryLock ();
}
public void releaseReadLock () {
this . regionObjectLock . readLock (). unlock ();
}
2026-08-13 13:45:28 +08:00
private void guardAgainstClosed () throws IOException {
if ( this . isClosedRaw ()) {
throw new IOException ( "Closed" );
}
}
2026-07-09 00:03:09 +08:00
public boolean isClosedRaw () {
return ( boolean ) CLOSED_HANDLE . getVolatile ( this );
}
public boolean isClosed () {
this . regionObjectLock . readLock (). lock ();
try {
return ( boolean ) CLOSED_HANDLE . getVolatile ( this );
} finally {
this . regionObjectLock . readLock (). unlock ();
}
}
public void syncIfNeeded () throws IOException {
try {
2026-08-13 13:45:28 +08:00
this . syncToMasterFile ( false , false , false , false );
2026-07-09 00:03:09 +08:00
} finally {
BEING_SYNCED_HANDLE . setVolatile ( this , false ); // mark as not being synced
}
}
2026-08-13 13:45:28 +08:00
// noSwapLock/noMasterLock: caller (closeInternal) must already hold
// regionObjectLock.write and masterFileLock.write respectively.
private void syncToMasterFile ( boolean forceSync , boolean forceCompact , boolean noSwapLock , boolean noMasterLock ) throws IOException {
2026-07-17 22:10:57 +08:00
// serialized against close: the swap channel cannot go away under a running sync
synchronized ( this . syncLock ) {
// skip if closed already
if ( this . isClosedRaw ()) {
return ;
}
2026-07-09 00:03:09 +08:00
2026-07-17 22:10:57 +08:00
// fast skip when there is nothing to sync; writers flip the flag back
// via markAsToSync() which triggers the next round
2026-07-18 18:30:52 +08:00
if ( ! SYNCED_HANDLE . compareAndSet ( this , false , true ) && ! forceSync ) {
2026-07-17 22:10:57 +08:00
return ;
}
2026-07-09 00:03:09 +08:00
2026-07-17 22:10:57 +08:00
try {
2026-08-13 13:45:28 +08:00
this . masterFileParser . sync ( this . masterFilePath , forceCompact , noSwapLock , noMasterLock );
2026-07-17 22:10:57 +08:00
} catch ( Throwable e ) {
// set back
SYNCED_HANDLE . setVolatile ( this , false );
throw new IOException ( "Failed to sync to master file!" , e );
}
2026-07-09 00:03:09 +08:00
}
}
private void tryLoadOldBlinearMasterFileData () throws IOException {
this . masterFileParser . tryParseMainFileOld ( this . masterFilePath );
}
private void initSwapFile () throws IOException {
this . swapFileChannel = FileChannel . open (
this . swapFilePath ,
SWAP_FILE_CHANNEL_OPTIONS
);
2026-07-26 22:30:02 +08:00
// fill default sectors; the swap file has no header, data starts at offset 0
2026-07-09 00:03:09 +08:00
for ( int i = 0 ; i < 1024 ; i ++ ) {
2026-07-26 22:30:02 +08:00
this . sectors [ i ] = new Sector ( i , 0 , 0 );
2026-07-09 00:03:09 +08:00
}
2026-07-26 22:30:02 +08:00
this . currentAcquiredIndex = 0 ;
this . liveBytes = 0 ;
2026-07-09 00:03:09 +08:00
}
2026-07-26 22:30:02 +08:00
private void recalculateCounters () {
long acquired = 0 ;
long live = 0 ;
2026-07-09 00:03:09 +08:00
for ( Sector sector : this . sectors ) {
2026-07-26 22:30:02 +08:00
// 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 );
2026-07-09 00:03:09 +08:00
if ( sector . hasData ()) {
2026-07-26 22:30:02 +08:00
live += sector . length ;
2026-07-09 00:03:09 +08:00
}
}
2026-07-26 22:30:02 +08:00
this . currentAcquiredIndex = acquired ;
this . liveBytes = live ;
2026-07-09 00:03:09 +08:00
}
private void flushInternal () throws IOException {
2026-07-26 22:30:02 +08:00
if ( this . isClosedRaw ()) {
return ;
2026-07-09 00:03:09 +08:00
}
2026-07-26 22:30:02 +08:00
// 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 ()) {
2026-08-13 13:45:28 +08:00
this . syncToMasterFile ( false , false , false , false );
2026-07-09 00:03:09 +08:00
}
}
private void closeInternal () throws IOException {
2026-08-13 13:45:28 +08:00
// note: any new sync attempt is blocked inside this block
2026-07-17 22:10:57 +08:00
synchronized ( this . syncLock ) {
2026-08-13 13:45:28 +08:00
this . masterFileParser . masterFileLock . writeLock (). lock ();
try {
// note: any read/write ops is blocked inside this block
2026-07-17 22:10:57 +08:00
this . regionObjectLock . writeLock (). lock ();
2026-08-13 13:45:28 +08:00
2026-07-17 22:10:57 +08:00
try {
2026-08-13 13:45:28 +08:00
if ( this . isClosedRaw ()) {
boolean duplicateClosed = false ;
if ( this . swapFileChannel . isOpen ()) {
this . swapFileChannel . close ();
duplicateClosed = true ;
}
duplicateClosed &= this . masterFileParser . tryCloseNoLock ();
if ( ! duplicateClosed ) {
throw new IOException ( "Already closed" );
}
return ;
}
2026-08-19 11:36:01 +08:00
// remove from flusher
this . markClosed ();
2026-08-13 13:45:28 +08:00
IOException failure = null ;
// 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
// since we hold the write lock and any read/write/sync ops is currently blocked all along the close logic, acquiring the locks inside sync is a disaster
try {
2026-08-15 16:08:52 +08:00
this . syncToMasterFile ( true , true , true , true );
} catch ( IOException ex ) {
2026-08-13 13:45:28 +08:00
failure = ex ;
}
try {
this . swapFileChannel . close ();
} catch ( IOException ex ) {
2026-08-15 16:08:52 +08:00
if ( failure == null ) failure = ex ;
else failure . addSuppressed ( ex );
2026-08-13 13:45:28 +08:00
}
try {
this . masterFileParser . closeNoLock ();
} catch ( IOException e ) {
2026-08-15 16:08:52 +08:00
if ( failure == null ) failure = e ;
else failure . addSuppressed ( e );
2026-08-13 13:45:28 +08:00
}
if ( failure != null ) {
throw failure ;
}
2026-08-15 16:08:52 +08:00
} finally {
2026-07-17 22:10:57 +08:00
this . regionObjectLock . writeLock (). unlock ();
}
2026-08-15 16:08:52 +08:00
} finally {
2026-08-13 13:45:28 +08:00
this . masterFileParser . masterFileLock . writeLock (). unlock ();
2026-07-17 22:10:57 +08:00
}
2026-07-09 00:03:09 +08:00
}
}
2026-07-17 22:10:57 +08:00
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 );
2026-07-09 00:03:09 +08:00
}
}
private void compactSwapFile () throws IOException {
final Sector [] newSectorsToBeReplaced = new Sector [ this . sectors . length ] ;
for ( int i = 0 ; i < this . sectors . length ; i ++ ) {
final Sector old = this . sectors [ i ] ;
if ( old . hasData ()) {
newSectorsToBeReplaced [ i ] = old ;
continue ;
}
// 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
2026-07-26 22:30:02 +08:00
// see the store method in Sector class
2026-07-09 00:03:09 +08:00
newSectorsToBeReplaced [ i ] = new Sector ( i , 0 , 0 );
}
long newAcquiredIndex ;
2026-07-17 22:10:57 +08:00
final Path targetTemp = Path . of ( this . swapFilePath + ".tmp" );
2026-07-09 00:03:09 +08:00
try ( FileChannel tempChannel = FileChannel . open (
targetTemp ,
StandardOpenOption . CREATE_NEW ,
StandardOpenOption . WRITE ,
StandardOpenOption . READ ,
StandardOpenOption . TRUNCATE_EXISTING
)) {
2026-07-26 22:30:02 +08:00
long offsetPointer = 0 ;
2026-07-09 00:03:09 +08:00
for ( Sector sector : newSectorsToBeReplaced ) {
// skip cleared or no data-contained sectors
if ( ! sector . hasData ()) {
continue ;
}
// transfer to target
2026-07-17 22:10:57 +08:00
transferFully ( this . swapFileChannel , sector . offset , sector . length , tempChannel , offsetPointer );
2026-07-09 00:03:09 +08:00
// recalculate the offset and length
final Sector newRecalculated = new Sector ( sector . index , offsetPointer , sector . length );
newRecalculated . hasData = true ;
offsetPointer += sector . length ;
newSectorsToBeReplaced [ sector . index ] = newRecalculated ; // update sector infos
}
2026-07-26 22:30:02 +08:00
// 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
2026-07-09 00:03:09 +08:00
newAcquiredIndex = offsetPointer ;
} catch ( Throwable ex ) {
2026-07-26 22:30:02 +08:00
// recalculate counters
this . recalculateCounters ();
2026-07-09 00:03:09 +08:00
// delete the target temp file
Files . deleteIfExists ( targetTemp );
// fast-fail
// note: we don't block new write operations here as this is recoverable
throw new IOException ( "Failed to compact swap file!" , ex );
}
this . swapFileChannel . close ();
// replace swap file
try {
2026-07-17 22:10:57 +08:00
atomicReplace ( targetTemp , this . swapFilePath );
2026-07-09 00:03:09 +08:00
} catch ( Throwable e ) {
2026-07-17 22:10:57 +08:00
// fast-fail
this . markClosed (); // prevent new writing & sync operations
throw new IOException ( "Failed to replace original swap file!" , e );
2026-07-09 00:03:09 +08:00
}
try {
// reopen file channel
this . reopenSwapFileChannel ();
2026-07-26 22:30:02 +08:00
// replace with recalculated infos: after a compact everything left is live
2026-07-09 00:03:09 +08:00
this . sectors = newSectorsToBeReplaced ;
this . currentAcquiredIndex = newAcquiredIndex ;
2026-07-26 22:30:02 +08:00
this . liveBytes = newAcquiredIndex ;
2026-07-09 00:03:09 +08:00
} 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
// data which is still in the memory
//
// which means we might write any data into any incorrect indexed sectors which will blow the whole data
this . markClosed ();
throw new IOException ( ex );
}
}
private void reopenSwapFileChannel () throws IOException {
if ( this . swapFileChannel . isOpen ()) {
this . swapFileChannel . close ();
}
this . swapFileChannel = FileChannel . open (
this . swapFilePath ,
SWAP_FILE_CHANNEL_OPTIONS
);
}
2026-07-26 22:30:02 +08:00
// 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 {
2026-07-09 00:03:09 +08:00
this . regionObjectLock . writeLock (). lock ();
try {
2026-08-13 13:45:28 +08:00
this . guardAgainstClosed ();
2026-07-26 22:30:02 +08:00
this . sectors [ index ] . store ( encoded , this . swapFileChannel );
2026-07-09 00:03:09 +08:00
if ( ! skipSync ) {
this . markBucketDirty ( index );
}
} finally {
this . regionObjectLock . writeLock (). unlock ();
}
if ( skipSync ) {
return ;
}
this . markAsToSync ();
}
2026-07-26 22:30:02 +08:00
// 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 );
2026-07-09 00:03:09 +08:00
}
2026-07-26 22:30:02 +08:00
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 );
2026-07-09 00:03:09 +08:00
}
private void clearChunkData ( int index ) throws IOException {
2026-08-13 13:45:28 +08:00
this . guardAgainstClosed ();
2026-07-09 00:03:09 +08:00
this . ensureBucketLoaded ( index );
this . regionObjectLock . writeLock (). lock ();
try {
2026-08-13 13:45:28 +08:00
this . guardAgainstClosed ();
2026-07-26 22:30:02 +08:00
this . sectors [ index ] . clear ();
2026-07-09 00:03:09 +08:00
this . markBucketDirty ( index );
} finally {
this . regionObjectLock . writeLock (). unlock ();
}
this . markAsToSync ();
}
private void markAsToSync () {
SYNCED_HANDLE . setVolatile ( this , false ); // mark as unsynced
LAST_WRITTEN_HANDLE . setVolatile ( this , System . nanoTime ()); // update last written time
}
private static int getChunkIndex ( int x , int z ) {
return ( x & 31 ) + (( z & 31 ) << 5 );
}
private boolean hasData ( int index ) throws IOException {
2026-08-13 13:45:28 +08:00
this . guardAgainstClosed ();
2026-07-09 00:03:09 +08:00
this . ensureBucketLoaded ( index );
this . regionObjectLock . readLock (). lock ();
try {
2026-08-13 13:45:28 +08:00
this . guardAgainstClosed ();
2026-07-09 00:03:09 +08:00
return this . sectors [ index ] . hasData ();
} finally {
this . regionObjectLock . readLock (). unlock ();
}
}
private void writeChunk ( int x , int z , @NotNull ByteBuffer data ) throws IOException {
2026-08-13 13:45:28 +08:00
this . guardAgainstClosed ();
2026-07-09 00:03:09 +08:00
final int chunkIndex = getChunkIndex ( x , z );
2026-07-17 22:10:57 +08:00
this . ensureBucketLoaded ( chunkIndex );
2026-07-26 22:30:02 +08:00
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 );
2026-07-09 00:03:09 +08:00
}
2026-07-26 22:30:02 +08:00
// absolute-offset hash: no position save/restore dance needed
final int xxHash32OfData = XX_HASH_32 . hash ( data , data . position (), dataLen , XXHASH32_SEED );
2026-07-09 00:03:09 +08:00
2026-07-26 22:30:02 +08:00
// 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 ));
2026-07-09 00:03:09 +08:00
2026-07-26 22:30:02 +08:00
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 ();
2026-07-09 00:03:09 +08:00
2026-07-26 22:30:02 +08:00
this . storeSector ( chunkIndex , out , false );
2026-07-09 00:03:09 +08:00
}
private @Nullable ByteBuffer readChunk ( int x , int z ) throws IOException {
2026-08-13 13:45:28 +08:00
this . guardAgainstClosed ();
2026-07-09 00:03:09 +08:00
final int chunkIndex = getChunkIndex ( x , z );
this . ensureBucketLoaded ( chunkIndex );
2026-07-26 22:30:02 +08:00
final ByteBuffer stage ;
2026-07-09 00:03:09 +08:00
2026-07-26 22:30:02 +08:00
this . regionObjectLock . readLock (). lock ();
try {
2026-08-13 13:45:28 +08:00
this . guardAgainstClosed ();
2026-07-26 22:30:02 +08:00
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 ();
2026-07-09 00:03:09 +08:00
}
2026-07-26 22:30:02 +08:00
stage . flip ();
2026-07-09 00:03:09 +08:00
2026-07-26 22:30:02 +08:00
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
2026-07-09 00:03:09 +08:00
}
2026-07-26 22:30:02 +08:00
return ByteBuffer . wrap ( data );
2026-07-09 00:03:09 +08:00
}
@Override
public Path getPath () {
return this . masterFilePath ;
}
@Override
public DataInputStream getChunkDataInputStream ( @NotNull ChunkPos pos ) throws IOException {
final ByteBuffer data = this . readChunk ( pos . x (), pos . z ());
if ( data == null ) {
return null ;
}
return new DataInputStream ( new ByteBufferInputStream ( data ));
}
@Override
public boolean doesChunkExist ( @NotNull ChunkPos pos ) throws IOException {
return this . hasData ( getChunkIndex ( pos . x (), pos . z ()));
}
@Override
public DataOutputStream getChunkDataOutputStream ( ChunkPos pos ) {
return new DataOutputStream ( new ChunkBufferHelper ( pos ));
}
@Override
public void clear ( @NotNull ChunkPos pos ) throws IOException {
this . clearChunkData ( getChunkIndex ( pos . x (), pos . z ()));
}
@Override
public boolean hasChunk ( @NotNull ChunkPos pos ) {
try {
return this . hasData ( getChunkIndex ( pos . x (), pos . z ()));
} catch ( IOException e ) {
throw new RuntimeException ( e );
}
}
@Override
public void write ( @NotNull ChunkPos pos , ByteBuffer buf ) throws IOException {
this . writeChunk ( pos . x (), pos . z (), buf );
}
// MCC 的玩意,这东西也用不上给Linear了()
@Override
public CompoundTag getOversizedData ( int x , int z ) {
return null ;
}
@Override
public boolean isOversized ( int x , int z ) {
return false ;
}
@Override
public boolean recalculateHeader () {
return false ;
}
@Override
public void setOversized ( int x , int z , boolean oversized ) {
}
// MCC end
@Override
public MoonriseRegionFileIO . RegionDataController . WriteData moonrise$startWrite ( CompoundTag data , ChunkPos pos ) {
final DataOutputStream out = this . getChunkDataOutputStream ( pos );
return new MoonriseRegionFileIO . RegionDataController . WriteData (
data , MoonriseRegionFileIO . RegionDataController . WriteData . WriteResult . WRITE ,
out , regionFile -> out . close ()
);
}
@Override
public void flush () throws IOException {
this . flushInternal ();
}
@Override
public void close () throws IOException {
this . closeInternal ();
}
public static class ByteBufferInputStream extends InputStream {
protected final ByteBuffer internal ;
public ByteBufferInputStream ( ByteBuffer buf ) {
this . internal = buf ;
}
@Override
public int available () {
return this . internal . remaining ();
}
@Override
public int read () throws IOException {
return this . internal . hasRemaining () ? ( this . internal . get () & 0xFF ) : - 1 ;
}
@Override
public int read ( byte @NotNull [] bytes , int off , int len ) throws IOException {
if ( ! this . internal . hasRemaining ()) return - 1 ;
len = Math . min ( len , this . internal . remaining ());
this . internal . get ( bytes , off , len );
return len ;
}
}
public class Sector {
private final int index ;
private long offset ;
private long length ;
private boolean hasData = false ;
private Sector ( int index , long offset , long length ) {
this . index = index ;
this . offset = offset ;
this . length = length ;
}
public void store ( @NotNull ByteBuffer newData , @NotNull FileChannel channel ) throws IOException {
final long oldLength = this . length ;
2026-07-26 22:30:02 +08:00
final long oldLive = this . hasData ? oldLength : 0L ;
2026-07-09 00:03:09 +08:00
final long newDataLength = newData . remaining ();
this . hasData = true ;
this . length = newDataLength ;
2026-07-26 22:30:02 +08:00
// data fits into the extent this sector already owns (a cleared sector keeps
// its stale extent exactly for this reuse), write it in place
2026-07-09 00:03:09 +08:00
if ( newDataLength <= oldLength ) {
writeFullyAt ( channel , newData , this . offset );
2026-07-26 22:30:02 +08:00
} else {
// or we will append to the end of file
this . offset = BufferedLinearRegionFile . this . currentAcquiredIndex ;
BufferedLinearRegionFile . this . currentAcquiredIndex = this . offset + newDataLength ;
2026-07-09 00:03:09 +08:00
2026-07-26 22:30:02 +08:00
writeFullyAt ( channel , newData , this . offset );
2026-07-09 00:03:09 +08:00
}
2026-07-26 22:30:02 +08:00
// single mutator under the region write lock; keeps the garbage estimate
// in flushInternal() lock-free and scan-free
BufferedLinearRegionFile . this . liveBytes += newDataLength - oldLive ;
2026-07-09 00:03:09 +08:00
}
public void clear () {
2026-07-26 22:30:02 +08:00
if ( this . hasData ) {
BufferedLinearRegionFile . this . liveBytes -= this . length ;
}
2026-07-09 00:03:09 +08:00
this . hasData = false ;
}
public boolean hasData () {
return this . hasData ;
}
}
private class ChunkBufferHelper extends ByteArrayOutputStream {
private final ChunkPos pos ;
private ChunkBufferHelper ( ChunkPos pos ) {
2026-07-26 22:30:02 +08:00
// 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 );
2026-07-09 00:03:09 +08:00
this . pos = pos ;
}
@Override
public void close () throws IOException {
ByteBuffer bytebuffer = ByteBuffer . wrap ( this . buf , 0 , this . count );
BufferedLinearRegionFile . this . writeChunk ( this . pos . x (), this . pos . z (), bytebuffer );
BufferedLinearRegionFile . this . flushInternal ();
}
}
2026-07-24 21:52:44 +08:00
private class MasterFileParser {
2026-07-26 22:30:02 +08:00
// V3 bucketed format layout (UNCHANGED, fully compatible with existing files):
2026-07-17 22:10:57 +08:00
// [0, 14): header — superblock(8) + version(1) + compressionLevel(1) + xxHash32Seed(4)
2026-07-09 00:03:09 +08:00
// [14, 142): position table — BUCKET_COUNT(16) × long(8) each; 0 = no data for that bucket
2026-07-17 22:10:57 +08:00
// [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 ;
2026-07-09 00:03:09 +08:00
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
2026-07-17 22:10:57 +08:00
private static final int V3_RECORD_HEADER_SIZE = Integer . BYTES * 2 ; // originalLen + compressedLen
2026-07-09 00:03:09 +08:00
private final ReadWriteLock masterFileLock = new ReentrantReadWriteLock ();
2026-07-26 22:30:02 +08:00
// 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.
2026-07-17 22:10:57 +08:00
// 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 ;
2026-07-26 22:30:02 +08:00
// single volatile read instead of a Files.exists() stat per chunk write
private volatile boolean fileExists ;
2026-07-17 22:10:57 +08:00
// a consistent snapshot of one bucket taken from the swap file;
// payload == null means the bucket holds no chunks at all
2026-07-26 22:30:02 +08:00
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 ();
}
2026-07-17 22:10:57 +08:00
}
// must be called under syncLock (see syncToMasterFile)
2026-08-13 13:45:28 +08:00
public void sync ( @NotNull Path mainFile , boolean forceCompact , boolean noSwapLock , boolean noMasterLock ) throws IOException {
if ( ! noMasterLock ) this . masterFileLock . writeLock (). lock ();
2026-07-17 22:10:57 +08:00
try {
2026-07-26 22:30:02 +08:00
// full rewrite whenever no valid append state exists (fresh region /
// corrupted table / legacy migration), and afterwards whenever the
2026-07-17 22:10:57 +08:00
// appended garbage passed the auto-compact threshold: writes a tmp file,
// then atomically replaces the master file with it
2026-07-18 18:30:52 +08:00
if ( this . appendChannel == null || this . shouldCompactMasterFile () || forceCompact ) {
2026-08-13 13:45:28 +08:00
this . rewriteFully ( mainFile , noSwapLock );
2026-07-17 22:10:57 +08:00
} else {
// WAL-style otherwise: only append the dirty buckets
2026-08-13 13:45:28 +08:00
this . appendDirtyBuckets ( noSwapLock );
2026-07-17 22:10:57 +08:00
}
} finally {
2026-08-13 13:45:28 +08:00
if ( ! noMasterLock ) this . masterFileLock . writeLock (). unlock ();
2026-07-17 22:10:57 +08:00
}
}
// 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 ;
}
2026-08-13 13:45:28 +08:00
private void rewriteFully ( @NotNull Path mainFile , boolean noSwapLock ) throws IOException {
2026-07-26 22:30:02 +08:00
final boolean wal = this . appendChannel != null ;
2026-07-09 00:03:09 +08:00
final Path tmpFilePath = Path . of ( mainFile + ".tmp" );
final long [] syncedBucketEpochs = new long [ BUCKET_COUNT ] ;
final long [] newPositionTable = new long [ BUCKET_COUNT ] ;
2026-07-17 22:10:57 +08:00
final long [] newRecordSizes = new long [ BUCKET_COUNT ] ;
final long newAppendOffset ;
2026-07-09 00:03:09 +08:00
2026-07-26 22:30:02 +08:00
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 );
}
2026-07-09 00:03:09 +08:00
try ( FileChannel outChannel = FileChannel . open ( tmpFilePath ,
StandardOpenOption . CREATE , StandardOpenOption . WRITE , StandardOpenOption . TRUNCATE_EXISTING )) {
2026-07-17 22:10:57 +08:00
this . writeV3Header ( outChannel );
2026-07-09 00:03:09 +08:00
2026-07-17 22:10:57 +08:00
// position table placeholder (all zeros, filled in at the end)
2026-07-09 00:03:09 +08:00
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 ++ ) {
2026-07-17 22:10:57 +08:00
if ( BufferedLinearRegionFile . this . isBucketDirty ( bucketIndex )) {
2026-08-13 13:45:28 +08:00
final BucketRecord record = this . buildBucketRecord ( bucketIndex , noSwapLock );
2026-07-09 00:03:09 +08:00
2026-07-17 22:10:57 +08:00
if ( record . payload () != null ) {
2026-07-26 22:30:02 +08:00
final int recordSize = record . payload (). remaining ();
writeFullyAt ( outChannel , record . payload (), dataOffset );
2026-07-09 00:03:09 +08:00
newPositionTable [ bucketIndex ] = dataOffset ;
2026-07-26 22:30:02 +08:00
newRecordSizes [ bucketIndex ] = recordSize ;
dataOffset += recordSize ;
2026-07-09 00:03:09 +08:00
}
2026-07-17 22:10:57 +08:00
// else: the bucket is empty now, its table entry stays 0
2026-07-09 00:03:09 +08:00
2026-07-17 22:10:57 +08:00
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 ] ;
2026-07-26 22:30:02 +08:00
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 ();
}
2026-07-09 00:03:09 +08:00
2026-07-17 22:10:57 +08:00
transferFully ( oldChannel , oldOffset , recordSize , outChannel , dataOffset );
newPositionTable [ bucketIndex ] = dataOffset ;
newRecordSizes [ bucketIndex ] = recordSize ;
dataOffset += recordSize ;
2026-07-09 00:03:09 +08:00
}
}
2026-07-17 22:10:57 +08:00
// write the finalized position table
writeFullyAt ( outChannel , this . encodePositionTable ( newPositionTable ), V3_POS_TABLE_OFFSET );
2026-07-09 00:03:09 +08:00
outChannel . force ( true );
2026-07-17 22:10:57 +08:00
newAppendOffset = dataOffset ;
2026-07-09 00:03:09 +08:00
}
2026-07-26 22:30:02 +08:00
} 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 ();
2026-07-09 00:03:09 +08:00
}
2026-07-17 22:10:57 +08:00
atomicReplace ( tmpFilePath , mainFile );
2026-07-26 22:30:02 +08:00
// (re)enter WAL mode: keep the freshly written master file open for appending syncs
2026-07-17 22:10:57 +08:00
this . appendChannel = FileChannel . open ( mainFile , StandardOpenOption . READ , StandardOpenOption . WRITE );
this . positionTable = newPositionTable ;
this . recordSizes = newRecordSizes ;
this . appendOffset = newAppendOffset ;
2026-07-26 22:30:02 +08:00
this . fileExists = true ;
2026-07-17 22:10:57 +08:00
this . markBucketsSynced ( syncedBucketEpochs );
}
2026-08-13 13:45:28 +08:00
private void appendDirtyBuckets ( boolean noSwapLock ) throws IOException {
2026-07-17 22:10:57 +08:00
final FileChannel channel = this . appendChannel ;
final long [] syncedBucketEpochs = new long [ BUCKET_COUNT ] ;
final long [] newPositionTable = this . positionTable . clone ();
final long [] newRecordSizes = this . recordSizes . clone ();
2026-07-26 22:30:02 +08:00
final ByteBuffer [] pending = new ByteBuffer [ BUCKET_COUNT ] ;
2026-07-17 22:10:57 +08:00
long dataOffset = this . appendOffset ;
2026-07-26 22:30:02 +08:00
int pendingCount = 0 ;
2026-07-17 22:10:57 +08:00
boolean anyDirty = false ;
for ( int bucketIndex = 0 ; bucketIndex < BUCKET_COUNT ; bucketIndex ++ ) {
if ( ! BufferedLinearRegionFile . this . isBucketDirty ( bucketIndex )) {
continue ;
}
2026-08-13 13:45:28 +08:00
final BucketRecord record = this . buildBucketRecord ( bucketIndex , noSwapLock );
2026-07-26 22:30:02 +08:00
final ByteBuffer payload = record . payload ();
2026-07-17 22:10:57 +08:00
2026-07-26 22:30:02 +08:00
if ( payload != null ) {
pending [ pendingCount ++] = payload ;
2026-07-17 22:10:57 +08:00
newPositionTable [ bucketIndex ] = dataOffset ;
2026-07-26 22:30:02 +08:00
newRecordSizes [ bucketIndex ] = payload . remaining ();
dataOffset += payload . remaining ();
2026-07-17 22:10:57 +08:00
} else {
// the bucket is empty now
newPositionTable [ bucketIndex ] = 0 ;
newRecordSizes [ bucketIndex ] = 0 ;
}
syncedBucketEpochs [ bucketIndex ] = record . epoch ();
anyDirty = true ;
}
if ( ! anyDirty ) {
return ;
}
2026-07-26 22:30:02 +08:00
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 );
}
2026-07-17 22:10:57 +08:00
// 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 );
}
2026-07-26 22:30:02 +08:00
// 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
2026-08-13 13:45:28 +08:00
private @NotNull BucketRecord buildBucketRecord ( int bucketIndex , final boolean noLock ) throws IOException {
2026-07-17 22:10:57 +08:00
final int baseChunkIndex = bucketIndex << BUCKET_SHIFT ;
2026-07-26 22:30:02 +08:00
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 ;
2026-07-17 22:10:57 +08:00
final long epoch ;
2026-08-13 13:45:28 +08:00
if ( ! noLock ) BufferedLinearRegionFile . this . regionObjectLock . readLock (). lock ();
2026-07-17 22:10:57 +08:00
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 ] ;
2026-07-26 22:30:02 +08:00
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 ;
2026-07-17 22:10:57 +08:00
}
} finally {
2026-08-13 13:45:28 +08:00
if ( ! noLock ) BufferedLinearRegionFile . this . regionObjectLock . readLock (). unlock ();
2026-07-17 22:10:57 +08:00
}
2026-07-26 22:30:02 +08:00
// 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 );
2026-07-17 22:10:57 +08:00
for ( int i = 0 ; i < BUCKET_SIZE ; i ++ ) {
2026-07-26 22:30:02 +08:00
final ByteBuffer raw = rawSectors [ i ] ;
2026-07-17 22:10:57 +08:00
// note: null -> no data contained
2026-07-26 22:30:02 +08:00
if ( raw == null ) {
sectionBuf . putInt ( 0 );
2026-07-17 22:10:57 +08:00
continue ;
}
2026-07-26 22:30:02 +08:00
final byte [] runArray = raw . array ();
final int rawBase = raw . arrayOffset () + raw . position ();
final int dataLen = raw . getInt ( raw . position ());
2026-07-17 22:10:57 +08:00
2026-07-26 22:30:02 +08:00
sectionBuf . putInt ( SECTOR_META_SIZE + dataLen );
sectionBuf . put ( runArray , rawBase , SECTOR_META_SIZE ); // meta bytes carried over verbatim
2026-07-17 22:10:57 +08:00
2026-07-26 22:30:02 +08:00
// 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 );
2026-07-17 22:10:57 +08:00
}
2026-07-26 22:30:02 +08:00
// 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 );
2026-07-17 22:10:57 +08:00
2026-07-26 22:30:02 +08:00
if ( Zstd . isError ( compressedLen )) {
throw new IOException ( "Failed to zstd compress bucket " + bucketIndex + ": " + Zstd . getErrorName ( compressedLen ));
}
2026-07-17 22:10:57 +08:00
2026-07-26 22:30:02 +08:00
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 ;
}
2026-07-17 22:10:57 +08:00
}
2026-08-13 22:36:52 +08:00
private void markBucketsSynced ( long @NonNull [] syncedBucketEpochs ) {
2026-07-09 00:03:09 +08:00
for ( int i = 0 ; i < syncedBucketEpochs . length ; i ++ ) {
2026-07-17 22:10:57 +08:00
// note: a dirty bucket always has a write epoch >= 1, so 0 = untouched
if ( syncedBucketEpochs [ i ] != 0L ) {
BufferedLinearRegionFile . this . markBucketSynced ( i , syncedBucketEpochs [ i ] );
2026-07-09 00:03:09 +08:00
}
}
}
2026-07-17 22:10:57 +08:00
// 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 );
2026-07-26 22:30:02 +08:00
header . putInt ( XXHASH32_SEED );
2026-07-17 22:10:57 +08:00
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 ();
}
2026-08-13 13:45:28 +08:00
public boolean tryCloseNoLock () throws IOException {
if ( this . appendChannel != null && this . appendChannel . isOpen ()) {
this . appendChannel . close ();
this . appendChannel = null ;
return true ;
}
return false ;
}
public void closeNoLock () throws IOException {
if ( this . appendChannel != null ) {
this . appendChannel . close ();
this . appendChannel = null ;
2026-07-17 22:10:57 +08:00
}
}
private void loadBucketsFor ( @NotNull Path file , int bucketIndex ) throws IOException {
2026-07-09 00:03:09 +08:00
final int beginChunkIndex = bucketIndex << BUCKET_SHIFT ;
this . masterFileLock . readLock (). lock ();
try {
2026-08-13 13:45:28 +08:00
BufferedLinearRegionFile . this . guardAgainstClosed ();
2026-07-17 22:10:57 +08:00
final ByteBuffer decompressed ;
2026-07-09 00:03:09 +08:00
2026-07-17 22:10:57 +08:00
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 )) {
2026-07-09 00:03:09 +08:00
return ;
}
2026-07-17 22:10:57 +08:00
try ( FileChannel channel = FileChannel . open ( file , StandardOpenOption . READ )) {
if ( channel . size () < V3_DATA_AREA_OFFSET ) {
return ;
}
2026-07-09 00:03:09 +08:00
2026-07-17 22:10:57 +08:00
this . checkV3Header ( channel );
2026-07-09 00:03:09 +08:00
2026-07-17 22:10:57 +08:00
decompressed = this . readBucketData ( channel , this . parseOffsetTable ( channel ) [ bucketIndex ] );
}
}
2026-07-09 00:03:09 +08:00
2026-07-17 22:10:57 +08:00
if ( decompressed != null ) {
2026-07-09 00:03:09 +08:00
this . loadChunksFromBucketData ( decompressed , beginChunkIndex );
}
} finally {
this . masterFileLock . readLock (). unlock ();
}
}
2026-07-17 22:10:57 +08:00
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 ));
}
2026-07-09 00:03:09 +08:00
private long @NonNull [] parseOffsetTable ( FileChannel channel ) throws IOException {
final ByteBuffer buf = ByteBuffer . allocate ( V3_POS_TABLE_SIZE );
readFullyAt ( channel , buf , V3_POS_TABLE_OFFSET );
buf . flip ();
final long [] table = new long [ BUCKET_COUNT ] ;
for ( int i = 0 ; i < BUCKET_COUNT ; i ++ ) {
2026-07-17 22:10:57 +08:00
table [ i ] = buf . getLong ();
2026-07-09 00:03:09 +08:00
}
return table ;
}
private void loadChunksFromBucketData ( ByteBuffer decompressed , int beginChunkIndex ) throws IOException {
for ( int chunkIndex = beginChunkIndex ; chunkIndex < beginChunkIndex + BUCKET_SIZE ; chunkIndex ++ ) {
final int chunkSectionDataSize = decompressed . getInt ();
if ( chunkSectionDataSize <= 0 ) continue ;
2026-07-26 22:30:02 +08:00
// slice instead of copying the section bytes out
final ByteBuffer section = decompressed . slice ( decompressed . position (), chunkSectionDataSize );
decompressed . position ( decompressed . position () + chunkSectionDataSize );
2026-07-09 00:03:09 +08:00
2026-07-26 22:30:02 +08:00
BufferedLinearRegionFile . this . writeSection ( chunkIndex , section , true );
2026-07-09 00:03:09 +08:00
}
}
private void parseLinearV2 ( @NonNull DataInputStream ioStream , Path file ) throws IOException {
try ( ioStream ) {
ioStream . readLong (); // Skip newestTimestamp (Long)
byte gridSize = ioStream . readByte ();
if ( gridSize != 1 && gridSize != 2 && gridSize != 4 && gridSize != 8 && gridSize != 16 && gridSize != 32 )
throw new RuntimeException ( "Invalid grid size: " + gridSize + " file " + file );
int bucketSize = 32 / gridSize ;
ioStream . readInt (); // Skip region_x (Int)
ioStream . readInt (); // Skip region_z (Int)
ioStream . skipBytes ( 128 ); // Skip existence bitmap
// Skip NBT features
while ( true ) {
byte featureNameLength = ioStream . readByte ();
if ( featureNameLength == 0 ) break ;
byte [] featureNameBytes = new byte [ featureNameLength ] ;
ioStream . readFully ( featureNameBytes );
ioStream . readInt (); // featureValue
}
// Read bucket metadata
int totalBuckets = gridSize * gridSize ;
int [] bucketSizes = new int [ totalBuckets ] ;
byte [] bucketCompressionLevels = new byte [ totalBuckets ] ;
long [] bucketHashes = new long [ totalBuckets ] ;
for ( int i = 0 ; i < totalBuckets ; i ++ ) {
bucketSizes [ i ] = ioStream . readInt ();
bucketCompressionLevels [ i ] = ioStream . readByte ();
bucketHashes [ i ] = ioStream . readLong ();
}
// Read and decompress each bucket, load chunks into swap
for ( int bx = 0 ; bx < gridSize ; bx ++ ) {
for ( int bz = 0 ; bz < gridSize ; bz ++ ) {
int bucketIdx = bx * gridSize + bz ;
if ( bucketSizes [ bucketIdx ] <= 0 ) continue ;
byte [] compressedBucket = new byte [ bucketSizes [ bucketIdx ]] ;
ioStream . readFully ( compressedBucket );
long rawHash = LongHashFunction . xx (). hashBytes ( compressedBucket );
if ( rawHash != bucketHashes [ bucketIdx ] ) {
throw new IOException ( "Region file hash incorrect for bucket " + bucketIdx + " in " + file );
}
ByteArrayInputStream bucketByteStream = new ByteArrayInputStream ( compressedBucket );
ZstdInputStream zstdStream = new ZstdInputStream ( bucketByteStream );
ByteBuffer bucketBuffer = ByteBuffer . wrap ( zstdStream . readAllBytes ());
zstdStream . close ();
for ( int cx = 0 ; cx < bucketSize ; cx ++ ) {
for ( int cz = 0 ; cz < bucketSize ; cz ++ ) {
int chunkX = bx * bucketSize + cx ;
int chunkZ = bz * bucketSize + cz ;
int chunkIndex = chunkX + chunkZ * 32 ;
int chunkSize = bucketBuffer . getInt ();
long timestamp = bucketBuffer . getLong ();
if ( chunkSize > 0 ) {
// chunkSize includes the 8 bytes of timestamp already written
int dataLen = chunkSize - 8 ;
byte [] chunkData = new byte [ dataLen ] ;
bucketBuffer . get ( chunkData );
// Mark bucket as loaded. writeChunk() bumps the bucket epoch so it gets synced to the new master format.
2026-07-17 22:10:57 +08:00
BufferedLinearRegionFile . this . markBucketLoaded ( chunkIndex );
2026-07-09 00:03:09 +08:00
// Use writeChunk to go through the full path (adds length + timestamp + xxhash header)
BufferedLinearRegionFile . this . writeChunk ( chunkX , chunkZ , ByteBuffer . wrap ( chunkData ));
}
}
}
}
}
// Footer validation
long footerSuperBlock = ioStream . readLong ();
if ( footerSuperBlock != LINEAR_FILE_SUPER_BLOCK ) {
throw new IOException ( "Footer superblock invalid " + file );
}
}
}
private boolean tryParseBlinearV2 ( @NotNull DataInputStream ioStream , Path file ) throws IOException {
final byte version = ioStream . readByte ();
// we will parse dynamically (V3)
if ( version == MASTER_FILE_VERSION_BUCKET ) {
ioStream . close ();
return false ;
}
if ( version != MASTER_FILE_VERSION )
throw new RuntimeException ( "Invalid version: " + version + " in " + file );
// Skip newestTimestamp (Long) + Compression level (Byte): Unused.
ioStream . skipBytes ( 9 );
try ( final ZstdInputStream decompressStream = new ZstdInputStream ( ioStream )) {
// only used as a helper stream
// the parent stream will be closed in the try-catch block upper
final DataInputStream decompressedStreamHelper = new DataInputStream ( decompressStream );
for ( int index = 0 ; index < 1024 ; index ++ ) {
int size = decompressedStreamHelper . readInt (); // len
if ( size > 0 ) {
byte [] sectorData = new byte [ size ] ;
decompressedStreamHelper . readFully ( sectorData , 0 , size ); // data
final ByteBuffer sectorDataNioBuffer = ByteBuffer . wrap ( sectorData );
2026-07-17 22:10:57 +08:00
BufferedLinearRegionFile . this . markBucketLoaded ( index );
2026-07-26 22:30:02 +08:00
// blinear v2 stored the exact section layout, feed it through the section path
BufferedLinearRegionFile . this . writeSection ( index , sectorDataNioBuffer , false );
2026-07-09 00:03:09 +08:00
}
}
}
return true ;
}
@Contract ( value = "_ -> new" , pure = true )
public static int @NotNull [] coordinatesFromIndex ( int chunkIndex ) {
int x = chunkIndex & 31 ;
int z = ( chunkIndex >> 5 ) & 31 ;
return new int [] { x , z };
}
private void parseLinearV1 ( @NotNull DataInputStream ioStream ) throws IOException {
// Skip newestTimestamp (Long) + Compression level (Byte) + Chunk count (Short): Unused.
ioStream . skipBytes ( 11 );
// Skip chunk data len(Int)(Unused).
ioStream . skipBytes ( 4 );
// Skip data hash (Long): Unused.
ioStream . skipBytes ( 8 );
try ( final ZstdInputStream decompressedStream = new ZstdInputStream ( ioStream )) {
// only used as a helper stream
// the parent stream will be closed in the try-catch block upper
final DataInputStream bufferHelper = new DataInputStream ( decompressedStream );
final int [] chunkStarts = new int [ 1024 ] ;
for ( int i = 0 ; i < 1024 ; i ++ ) {
chunkStarts [ i ] = bufferHelper . readInt ();
bufferHelper . skipBytes ( 4 ); // Skip timestamps (Int): Unused.
}
for ( int i = 0 ; i < 1024 ; i ++ ) {
if ( chunkStarts [ i ] > 0 ) {
int size = chunkStarts [ i ] ;
byte [] chunkData = new byte [ size ] ;
bufferHelper . readFully ( chunkData );
final ByteBuffer chunkDataNioBuffer = ByteBuffer . wrap ( chunkData );
final int [] posByAxis = coordinatesFromIndex ( i );
final int x = posByAxis [ 0 ] ;
final int z = posByAxis [ 1 ] ;
2026-07-17 22:10:57 +08:00
BufferedLinearRegionFile . this . markBucketLoaded ( i );
2026-07-09 00:03:09 +08:00
BufferedLinearRegionFile . this . writeChunk ( x , z , chunkDataNioBuffer );
}
}
}
}
2026-07-20 21:25:40 +08:00
// won't and need not hold any region locks as we are calling this in a safe point (initially newed)
2026-07-09 00:03:09 +08:00
public void tryParseMainFileOld ( @NotNull Path mainFilePath ) throws IOException {
final File file = mainFilePath . toFile ();
if ( ! file . exists () || ! file . canRead ()) {
return ;
}
// those streams will be closed in the parse logic, or we will close it manually
final FileInputStream fileStream = new FileInputStream ( file );
final DataInputStream rawDataStream = new DataInputStream ( fileStream );
boolean oldParsed = false ;
final long superBlock ;
try {
superBlock = rawDataStream . readLong ();
if ( superBlock == MASTER_FILE_SUPER_BLOCK ) {
oldParsed = this . tryParseBlinearV2 ( rawDataStream , mainFilePath );
// false -> v3 -> closed in parse block
if ( ! oldParsed ) {
return ;
}
}
if ( superBlock == LINEAR_FILE_SUPER_BLOCK ) {
final byte version = rawDataStream . readByte ();
if ( version == 1 || version == 2 ) {
this . parseLinearV1 ( rawDataStream );
oldParsed = true ;
}
if ( version == 3 ) {
this . parseLinearV2 ( rawDataStream , mainFilePath );
oldParsed = true ;
}
}
} catch ( Throwable ex ) {
try {
rawDataStream . close ();
} catch ( IOException ex2 ) {
ex . addSuppressed ( ex2 );
}
throw new IOException ( "Failed to parse master file: " + mainFilePath , ex );
}
// old parsed, remove the original file, and we will recreate it as we sync
if ( oldParsed ) {
// immediately do sync operation
2026-08-13 13:45:28 +08:00
BufferedLinearRegionFile . this . syncToMasterFile ( true , true , false , false );
2026-07-09 00:03:09 +08:00
return ;
}
// anyone non-matched, close stream and throw the error
rawDataStream . close ();
throw new IOException ( "Unknown or unsupported super block : " + superBlock );
}
}
2026-07-26 22:30:02 +08:00
}