refactor: Restructure project into multi-module setup and improve bundler
- Convert project to multi-module structure with 'bundler' subproject - Extract bundler functionality into separate module with proper dependencies - Improve IO state management with atomic reference for better synchronization - Enhance bundler performance with parallel processing support - Add CLI interface to bundler with various compression options - Fix file handling issues in bundle reader/writer - Update compression configuration and remove redundant options - Improve region file synchronization with proper state management - Add dictionary support for better compression ratios - Refactor backup strategies to use new bundler API - Update dependencies and build configuration - Fix path resolution issues in bundle operations `
This commit is contained in:
iceBear67 committed
1 parent
3a4256cee1
commit
28280bc162
26 files changed
+605
-536
No files matched your search
@@ -0,0 +1,21 @@
|
||||
plugins {
|
||||
id 'java'
|
||||
id 'java-library'
|
||||
id("io.freefair.lombok") version "9.0.0-rc2"
|
||||
id("com.gradleup.shadow") version '9.1.0'
|
||||
}
|
||||
|
||||
|
||||
dependencies {
|
||||
api 'com.github.luben:zstd-jni:1.5.7-4'
|
||||
compileOnly 'io.netty:netty-all:4.2.6.Final'
|
||||
api "io.ib67.kiwi:lang:1.2.3-SNAPSHOT"
|
||||
api 'io.netty:netty-all:4.2.6.Final'
|
||||
}
|
||||
|
||||
|
||||
jar {
|
||||
manifest {
|
||||
attributes(Map.of("Main-Class", "io.ib67.sfcraft.bundler.Main"))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
package io.ib67.sfcraft.bundler;
|
||||
|
||||
public record BundleEntry(
|
||||
short attribute,
|
||||
long fileLen,
|
||||
String name
|
||||
) {
|
||||
public static short ATTR_GUNZIP = 1;
|
||||
public static short MAGIC = (short) 0xA2E3;
|
||||
}
|
||||
@@ -0,0 +1,39 @@
|
||||
package io.ib67.sfcraft.bundler;
|
||||
|
||||
import com.github.luben.zstd.Zstd;
|
||||
import com.github.luben.zstd.ZstdInputStreamNoFinalizer;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.RandomAccessFile;
|
||||
import java.nio.channels.Channels;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.StandardOpenOption;
|
||||
|
||||
public class BundleReader {
|
||||
public static void extract(Path bundlePath, Path destination) throws IOException {
|
||||
var buffer = new byte[4096];
|
||||
try (var in = new EntryInputStream(
|
||||
new ZstdInputStreamNoFinalizer(Files.newInputStream(bundlePath)))) {
|
||||
while (in.available() > 0) {
|
||||
var entry = in.readEntry();
|
||||
var len = entry.fileLen();
|
||||
var dstPath = destination.resolve(Path.of(entry.name()));
|
||||
if (Files.notExists(dstPath.getParent())) {
|
||||
Files.createDirectories(dstPath.getParent());
|
||||
}
|
||||
try (var out = Files.newOutputStream(dstPath,
|
||||
StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE)) {
|
||||
var read = 0;
|
||||
var toRead = len;
|
||||
while (toRead > 0) {
|
||||
read = in.read(buffer, 0, (int) Math.min(buffer.length, toRead));
|
||||
toRead -= read;
|
||||
out.write(buffer, 0, read);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,185 @@
|
||||
package io.ib67.sfcraft.bundler;
|
||||
|
||||
import com.github.luben.zstd.ZstdDictCompress;
|
||||
import com.github.luben.zstd.ZstdOutputStreamNoFinalizer;
|
||||
import io.ib67.kiwi.routine.Uni;
|
||||
import io.netty.buffer.AdaptiveByteBufAllocator;
|
||||
import io.netty.buffer.ByteBufAllocator;
|
||||
import lombok.Builder;
|
||||
import lombok.SneakyThrows;
|
||||
|
||||
import java.io.*;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.*;
|
||||
import java.util.function.UnaryOperator;
|
||||
import java.util.zip.GZIPInputStream;
|
||||
|
||||
public class BundleWriter implements Closeable {
|
||||
public static Comparator<Path> SORT = BundleWriter::preferRegion;
|
||||
protected static final ByteBufAllocator ALLOC = new AdaptiveByteBufAllocator();
|
||||
protected final boolean allowGunzip;
|
||||
protected final int compressionLevel;
|
||||
protected final Path relativeRoot;
|
||||
protected final EntryOutputStream outputStream;
|
||||
protected final byte[] buffer;
|
||||
|
||||
@SneakyThrows
|
||||
@Builder
|
||||
public BundleWriter(
|
||||
boolean allowGunzip,
|
||||
int compressionLevel,
|
||||
int maxWorkers,
|
||||
Path relativeRoot,
|
||||
OutputStream outputStream,
|
||||
byte[] dictionary
|
||||
) {
|
||||
this.allowGunzip = allowGunzip;
|
||||
this.compressionLevel = compressionLevel;
|
||||
this.relativeRoot = relativeRoot;;
|
||||
var zstdOut = new ZstdOutputStreamNoFinalizer(outputStream, compressionLevel);
|
||||
if (dictionary != null) {
|
||||
zstdOut.setDict(new ZstdDictCompress(dictionary, compressionLevel));
|
||||
}
|
||||
if(maxWorkers > 1) zstdOut.setWorkers(maxWorkers);
|
||||
this.outputStream = new EntryOutputStream(zstdOut);
|
||||
this.buffer = new byte[4096];
|
||||
}
|
||||
|
||||
public static CompletableFuture<List<Path>> createBundleParallelized(
|
||||
Path parentOfBundle,
|
||||
List<Path> _paths,
|
||||
int threshold,
|
||||
Executor executor,
|
||||
UnaryOperator<BundleWriterBuilder> config
|
||||
) throws IOException {
|
||||
var paths = new ArrayList<>(_paths);
|
||||
if (Files.notExists(parentOfBundle)) {
|
||||
Files.createDirectories(parentOfBundle);
|
||||
}
|
||||
// Shuffle once to make elements distribute evenly, then split and sort for
|
||||
// the final sequence
|
||||
Collections.shuffle(paths);
|
||||
if (paths.size() >= threshold) {
|
||||
var parallelism = Math.min(
|
||||
Runtime.getRuntime().availableProcessors() / 2
|
||||
, paths.size() / threshold
|
||||
); //todo tuning and debug
|
||||
|
||||
var parted = partition(paths, paths.size() / parallelism);
|
||||
var futures = new ArrayList<CompletableFuture<Path>>();
|
||||
for (int i = 0; i < parted.size(); i++) {
|
||||
var pathList = parted.get(i);
|
||||
var output = parentOfBundle.resolve("Worker_" + System.currentTimeMillis() + "_" + i + ".jpack");
|
||||
var future = CompletableFuture.supplyAsync(() -> createBundleWorker(pathList, output, config), executor);
|
||||
futures.add(future);
|
||||
}
|
||||
return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).thenApply((v) ->
|
||||
Uni.from(futures::forEach).map(CompletableFuture::join).toList());
|
||||
} else {
|
||||
return CompletableFuture.supplyAsync(() -> {
|
||||
var singleResult = parentOfBundle.resolve("result.zst");
|
||||
try {
|
||||
createBundle(singleResult, paths, config);
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
return List.of(singleResult);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@SneakyThrows
|
||||
private static Path createBundleWorker(
|
||||
List<Path> task,
|
||||
Path output,
|
||||
UnaryOperator<BundleWriterBuilder> config
|
||||
) {
|
||||
task.sort(SORT);
|
||||
createBundle(output, task, config);
|
||||
return output;
|
||||
}
|
||||
|
||||
private static <E> List<List<E>> partition(List<E> toPartition, int eachSize) {
|
||||
var result = new ArrayList<List<E>>();
|
||||
var remain = toPartition.size() % eachSize;
|
||||
var groups = (toPartition.size() - remain) / eachSize;
|
||||
for (int i = 0; i < groups; i++) {
|
||||
result.add(toPartition.subList(i * eachSize, (i + 1) * eachSize));
|
||||
}
|
||||
if (remain != 0) result.add(toPartition.subList(groups * eachSize, groups * eachSize + remain));
|
||||
return result;
|
||||
}
|
||||
|
||||
public static int preferRegion(Path a, Path b) {
|
||||
boolean aMatch = a.getParent().toString().equals("region") && a.endsWith(".mca");
|
||||
boolean bMatch = b.getParent().toString().equals("region") && b.endsWith(".mca");
|
||||
|
||||
if (aMatch && !bMatch) return -1;
|
||||
if (!aMatch && bMatch) return 1;
|
||||
return a.compareTo(b);
|
||||
}
|
||||
|
||||
public static void createBundle(
|
||||
Path pathToBundle,
|
||||
Iterable<Path> path,
|
||||
UnaryOperator<BundleWriterBuilder> config
|
||||
) throws IOException {
|
||||
try (var writer = config.apply(BundleWriter.builder())
|
||||
.outputStream(Files.newOutputStream(pathToBundle))
|
||||
.build()) {
|
||||
for (var p : path) {
|
||||
if (!Files.isRegularFile(p) || Files.size(p) == 0) continue;
|
||||
var s = p.toString();
|
||||
writer.write(p, (s.contains("region") && s.endsWith(".mca")) || s.endsWith(".dat"));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void write(Path path, boolean suggestGunzip) throws IOException {
|
||||
var size = Files.size(path);
|
||||
if (suggestGunzip) {
|
||||
boolean isGzip;
|
||||
try (var raf = new RandomAccessFile(path.toFile(), "r")) {
|
||||
isGzip = raf.read() == 0x8b && raf.read() == 0x1f;
|
||||
}
|
||||
if (isGzip) {
|
||||
var buffer = ALLOC.buffer((int) size * 2);
|
||||
try (var gunzip = new GZIPInputStream(new FileInputStream(path.toFile()))) {
|
||||
while (buffer.writeBytes(gunzip, 4096) != -1) ;
|
||||
outputStream.writeEntryHeader( // bug
|
||||
relativeRoot.relativize(path).toString(),
|
||||
BundleEntry.ATTR_GUNZIP,
|
||||
size
|
||||
);
|
||||
buffer.readBytes(outputStream, buffer.readableBytes());
|
||||
} catch (IOException e) {
|
||||
writePlain(path);
|
||||
} finally {
|
||||
buffer.release();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
writePlain(path);
|
||||
}
|
||||
|
||||
private void writePlain(Path path) throws IOException {
|
||||
var rPath = relativeRoot.toAbsolutePath().relativize(path.toAbsolutePath());
|
||||
var buffer = this.buffer;
|
||||
var outputStream = this.outputStream;
|
||||
outputStream.writeEntryHeader(rPath.toString(), (short) 0, Files.size(path));
|
||||
var read = 0;
|
||||
try (var fs = Files.newInputStream(path)) {
|
||||
while ((read = fs.read(buffer)) > 0) {
|
||||
outputStream.write(buffer, 0, read);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
outputStream.close();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
package io.ib67.sfcraft.bundler;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.buffer.Unpooled;
|
||||
|
||||
import java.io.FilterInputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
public class EntryInputStream extends FilterInputStream {
|
||||
private final ByteBuffer buffer;
|
||||
|
||||
/**
|
||||
* Creates a {@code FilterInputStream}
|
||||
* by assigning the argument {@code in}
|
||||
* to the field {@code this.in}
|
||||
* to remember it for later use.
|
||||
*
|
||||
* @param in the underlying input stream, or {@code null} if
|
||||
* this instance is to be created without an underlying stream.
|
||||
*/
|
||||
protected EntryInputStream(InputStream in) {
|
||||
super(in);
|
||||
this.buffer = ByteBuffer.allocate(4 + 8 + 4);
|
||||
}
|
||||
|
||||
public BundleEntry readEntry() throws IOException {
|
||||
buffer.clear();
|
||||
in.read(buffer.array(), 0, 4+8+4);
|
||||
var magic = buffer.getShort();
|
||||
if (magic != BundleEntry.MAGIC) throw new IOException("Invalid magic number");
|
||||
var attribute = buffer.getShort();
|
||||
var fileLen = buffer.getLong();
|
||||
var lenName = buffer.getInt();
|
||||
var name = new String(in.readNBytes(lenName)).intern();
|
||||
return new BundleEntry(attribute, fileLen, name);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
super.close();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
package io.ib67.sfcraft.bundler;
|
||||
|
||||
import io.netty.buffer.ByteBufAllocator;
|
||||
import io.netty.buffer.Unpooled;
|
||||
|
||||
import java.io.FilterOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.OutputStream;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
|
||||
public class EntryOutputStream extends FilterOutputStream {
|
||||
protected final ByteBuffer buffer;
|
||||
/**
|
||||
* Creates an output stream filter built on top of the specified
|
||||
* underlying output stream.
|
||||
*
|
||||
* @param out the underlying output stream to be assigned to
|
||||
* the field {@code this.out} for later use, or
|
||||
* {@code null} if this instance is to be
|
||||
* created without an underlying stream.
|
||||
*/
|
||||
public EntryOutputStream(OutputStream out) {
|
||||
super(out);
|
||||
this.buffer = ByteBuffer.allocate(4+8+4);
|
||||
}
|
||||
|
||||
public void writeEntryHeader(String path, short attribute, long fileLen) throws IOException {
|
||||
var name = path.getBytes(StandardCharsets.UTF_8);
|
||||
buffer.clear();
|
||||
buffer.putShort(BundleEntry.MAGIC);
|
||||
buffer.putShort(attribute);
|
||||
buffer.putLong(fileLen);
|
||||
buffer.putInt(name.length);
|
||||
out.write(buffer.array());
|
||||
out.write(name);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void write(byte[] b, int off, int len) throws IOException {
|
||||
out.write(b, off, len);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void write(byte[] b) throws IOException {
|
||||
out.write(b);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
package io.ib67.sfcraft.bundler;
|
||||
|
||||
import com.github.luben.zstd.Zstd;
|
||||
import com.github.luben.zstd.ZstdOutputStreamNoFinalizer;
|
||||
import io.ib67.kiwi.ArgOpts;
|
||||
import lombok.SneakyThrows;
|
||||
|
||||
import java.io.BufferedOutputStream;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ForkJoinPool;
|
||||
import java.util.function.UnaryOperator;
|
||||
|
||||
public class Main {
|
||||
@SneakyThrows
|
||||
public static void main(String[] args) {
|
||||
var opts = ArgOpts.builder()
|
||||
.args(args).description("Tools for .jpack format bundle").programName("jpack").build();
|
||||
var in = opts.string("in", "Input. Can be directory or jpack bundle", null);
|
||||
var out = opts.string("out", "Output. Can be jpack bundle or directory", null);
|
||||
var help = opts.bool("help", false);
|
||||
var allowGunzip = opts.bool("gunzip",
|
||||
"Should we recompress NBT files for maximum compression rate (depends on compress)", true);
|
||||
var parallel = opts.bool("parallel", "Use parallelized bundler. This conflicts with compress options", false);
|
||||
var parallelThreshold = opts.integer("threshold", "The threshold of parallel bundling", 512);
|
||||
var parallelCompression = opts.integer("parallel-compression", "The worker threads used for parallel compression. Do not confuse this with --parallel, which focus on bundling.\n" +
|
||||
"This option will be ignored when --parallel is present.", 1);
|
||||
var compressionLevel = opts.integer("level", "Zstd compression level", Zstd.defaultCompressionLevel());
|
||||
var dictionaryPath = opts.string("dict", "Zstd dictionary file", null);
|
||||
if (help) {
|
||||
opts.printHelp(System.out);
|
||||
return;
|
||||
}
|
||||
var dict = (dictionaryPath != null) ? Files.readAllBytes(Path.of(dictionaryPath)) : null;
|
||||
if (in == null || out == null) {
|
||||
System.out.println("Both in and out should be provided");
|
||||
return;
|
||||
}
|
||||
var inPath = Path.of(in);
|
||||
var outPath = Path.of(out);
|
||||
if (!Files.exists(inPath)) {
|
||||
System.out.println("The input isn't exists.");
|
||||
return;
|
||||
}
|
||||
if (Files.exists(outPath)) {
|
||||
System.out.println("The output is already exists. Please move it to other places");
|
||||
return;
|
||||
}
|
||||
if (in.endsWith(".jpack")) {
|
||||
BundleReader.extract(inPath, outPath);
|
||||
} else if (out.endsWith(".jpack")) {
|
||||
if (!Files.isDirectory(inPath))
|
||||
throw new IllegalArgumentException("Your input should be a world directory.");
|
||||
UnaryOperator<BundleWriter.BundleWriterBuilder> cfg = it -> it
|
||||
.allowGunzip(allowGunzip)
|
||||
.relativeRoot(outPath.toAbsolutePath().getParent())
|
||||
.dictionary(dict)
|
||||
.compressionLevel(compressionLevel);
|
||||
List<Path> paths;
|
||||
try (var _paths = Files.walk(inPath)) {
|
||||
paths = _paths.toList();
|
||||
}
|
||||
if (parallel) {
|
||||
try (var fj = ForkJoinPool.commonPool()) {
|
||||
BundleWriter.createBundleParallelized(
|
||||
outPath.getParent(),
|
||||
paths,
|
||||
parallelThreshold, fj,
|
||||
b -> cfg.apply(b).maxWorkers(0)
|
||||
).join();
|
||||
}
|
||||
} else {
|
||||
BundleWriter.createBundle(
|
||||
outPath,
|
||||
paths,
|
||||
c -> cfg.apply(c).maxWorkers(parallelCompression)
|
||||
);
|
||||
}
|
||||
} else {
|
||||
System.out.println("Invalid arguments. You should provide a pair of jpack and directory to act as in/output.");
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user