Optimize world generation

- Limit terrain cleanup to affected columns and batch cave boundary reads.
- Reuse offline generation workers and native air states.
- Gather history writes and defer unpublished append synchronization until atomic publication.
- Cover terrain parity, worker lifecycle, history durability, and publication failures.
This commit is contained in:
Brian Neumann-Fopiano committed 2026-09-23 21:32:17 -04:00
1 parent 7986698c1c
commit 38cf4ac7c7
25 files changed
+1362 -226

No files matched your search

@@ -37,6 +37,7 @@ final class HeadlessNativePlatform implements IrisPlatform, PlatformRegistries,
private final PlatformBiomeWriter biomeWriter;
private final PlatformGenerationRegistry generationRegistry;
private final NativeBlockResolver blocks = new NativeBlockResolver(this);
private final NativeBlockState air = blocks.getAir();
private final ConcurrentHashMap<String, NativeBlockState> states = new ConcurrentHashMap<>();
HeadlessNativePlatform(File root, Supplier<HolderLookup.Provider> registries) {
@@ -101,7 +102,7 @@ final class HeadlessNativePlatform implements IrisPlatform, PlatformRegistries,
@Override
public NativeBlockState air() {
return blocks.getAir();
return air;
}
@Override
@@ -22,16 +22,26 @@ import art.arcane.volmlib.nativelib.minecraft26_2.modded.NativeStateMerger;
import java.io.File;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.file.Path;
import net.minecraft.core.HolderLookup;
import net.minecraft.core.RegistryAccess;
import net.minecraft.data.registries.VanillaRegistries;
import net.minecraft.server.packs.resources.ResourceManager;
final class HeadlessNativeRuntime implements RealPackProbeSupport.RuntimeBindings {
private final Path stagingRoot;
private NativeRegionTerrainWriter writer;
private HeadlessNativeRegistries nativeRegistries;
private HolderLookup.Provider registries;
HeadlessNativeRuntime() {
stagingRoot = null;
}
HeadlessNativeRuntime(Path stagingRoot) throws IOException {
this.stagingRoot = stagingRoot.toRealPath();
}
@Override
public IrisPlatform create(File root) {
HeadlessNativeBootstrap.initialize();
@@ -68,6 +78,12 @@ final class HeadlessNativeRuntime implements RealPackProbeSupport.RuntimeBinding
@Override
public GenerationHistory createHistory(RealPackProbeSupport.HistoryRequest request) throws IOException {
if (stagingRoot != null) {
if (!request.worldRoot().getParent().toRealPath().startsWith(stagingRoot)) {
throw new IOException("Unpublished native generation must remain inside its private staging directory");
}
return HeadlessGenerationHistorySession.createUnpublished(request, new DataFixerV263());
}
return HeadlessGenerationHistorySession.create(request, new DataFixerV263());
}
@@ -45,14 +45,15 @@ final class HeadlessRegionTerrain {
static final class Session {
private final Map<Long, HeadlessStructurePlanner.PlannedChunk<NativeStructureOwnershipRecord>> chunks = new HashMap<>();
private final RegionGenerationWindow.Policy workerPolicy;
private final RegionGenerationWindow workers;
private HeadlessTerrainContext context;
Session(RegionGenerationWindow.Policy workerPolicy) {
this.workerPolicy = Objects.requireNonNull(workerPolicy, "workerPolicy");
Session(RegionGenerationWindow workers) {
this.workers = Objects.requireNonNull(workers, "workers");
}
Result generate(Request request, Consumer<String> progress) throws Exception {
workers.requireOpen();
if (context != null && context != request.context()) {
throw new IllegalArgumentException("Rolling regions require one native terrain context");
}
@@ -85,7 +86,7 @@ final class HeadlessRegionTerrain {
}
long started = System.nanoTime();
AtomicInteger completed = new AtomicInteger();
RegionGenerationWindow.process(new RegionGenerationWindow.Request<>(eligible.size(), request.parallelism(),
workers.process(new RegionGenerationWindow.Request<>(eligible.size(), request.parallelism(),
index -> apply(request.context(), eligible.get(index), status, chunks),
(index, result) -> {
if (status == ChunkStatus.STRUCTURE_STARTS) {
@@ -96,7 +97,7 @@ final class HeadlessRegionTerrain {
progress.accept("stage=" + status.getName() + " chunks=" + count + "/" + eligible.size()
+ " seconds=" + (System.nanoTime() - started) / 1_000_000_000.0);
}
}), workerPolicy);
}));
failReported();
}
List<ProtoChunk> ordered = new ArrayList<>(plan.size());
@@ -82,7 +82,7 @@ public final class OfflineRegionGenerator {
Path capturedConfiguration = staged.resolve("spigot.yml");
Files.write(capturedConfiguration, configurationBytes, StandardOpenOption.CREATE_NEW);
String configurationHash = HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(configurationBytes));
HeadlessNativeRuntime runtime = new HeadlessNativeRuntime();
HeadlessNativeRuntime runtime = new HeadlessNativeRuntime(staged);
long started = System.nanoTime();
long writes = 0;
long terrainWrites = 0;
@@ -102,12 +102,14 @@ public final class OfflineRegionGenerator {
: hydrology.preparePregeneration(configuration.hydrologyArea());
HeadlessTerrainContext context = HeadlessTerrainContext.create(new HeadlessTerrainContext.Options(
session.engine(), runtime, staged.resolve("templates"), configuration.levelKey(),
configuration.noiseSettings(), spigot))) {
HeadlessRegionTerrain.Session rolling = new HeadlessRegionTerrain.Session(RegionGenerationWindow.Policy.STANDALONE);
configuration.noiseSettings(), spigot));
RegionGenerationWindow workers = new RegionGenerationWindow(configuration.parallelism(),
RegionGenerationWindow.Policy.STANDALONE)) {
HeadlessRegionTerrain.Session rolling = new HeadlessRegionTerrain.Session(workers);
Path regionDirectory = session.engine().getTarget().getWorld().worldFolder().toPath().resolve("region");
Files.createDirectory(regionDirectory);
OfflineRegionWriter writer = new OfflineRegionWriter(new OfflineRegionWriter.Options(
context, regionDirectory, configuration.parallelism(), RegionGenerationWindow.Policy.STANDALONE));
context, regionDirectory, workers));
for (int row = 0; row < configuration.regionsZ(); row++) {
for (int column = 0; column < configuration.regionsX(); column++) {
int regionX = row % 2 == 0 ? column : configuration.regionsX() - column - 1;
@@ -142,9 +144,7 @@ public final class OfflineRegionGenerator {
wallSeconds, generationSeconds, serializationSeconds);
Files.writeString(staged.resolve("checkpoint.properties"), report, StandardOpenOption.CREATE_NEW);
long publicationStarted = System.nanoTime();
forceTree(staged);
Files.move(staged, configuration.output().resolve("native-terrain-checkpoint"), StandardCopyOption.ATOMIC_MOVE);
force(configuration.output());
publish(staged, configuration.output().resolve("native-terrain-checkpoint"));
System.out.println(report);
System.out.println("publicationSeconds=" + (System.nanoTime() - publicationStarted) / 1_000_000_000.0);
System.out.println("totalSeconds=" + (System.nanoTime() - started) / 1_000_000_000.0);
@@ -207,6 +207,12 @@ public final class OfflineRegionGenerator {
});
}
static void publish(Path staged, Path destination) throws IOException {
forceTree(staged);
Files.move(staged, destination, StandardCopyOption.ATOMIC_MOVE);
force(destination.getParent());
}
private static void force(Path path) throws IOException {
try (FileChannel channel = FileChannel.open(path, Files.isDirectory(path)
? StandardOpenOption.READ : StandardOpenOption.WRITE)) {
@@ -38,6 +38,7 @@ final class OfflineRegionWriter {
}
Result write(HeadlessRegionTerrain.Result generated) throws Exception {
options.workers().requireOpen();
try (OpenRegions regions = new OpenRegions(options)) {
List<ChunkWrite> pending = new ArrayList<>(generated.updated().size());
for (ProtoChunk chunk : generated.updated()) {
@@ -62,8 +63,8 @@ final class OfflineRegionWriter {
return new Result(0, 0);
}
int[] counts = new int[2];
RegionGenerationWindow.process(new RegionGenerationWindow.Request<>(pending.size(),
Math.min(options.parallelism(), MAXIMUM_PREPARATION_WORKERS),
options.workers().process(new RegionGenerationWindow.Request<>(pending.size(),
Math.min(options.workers().parallelism(), MAXIMUM_PREPARATION_WORKERS),
index -> prepare(pending.get(index)),
(index, prepared) -> {
ChunkWrite chunk = prepared.chunk();
@@ -73,7 +74,7 @@ final class OfflineRegionWriter {
if (chunk.chunk().getPersistedStatus().isOrAfter(ChunkStatus.TERRAIN)) {
counts[1]++;
}
}), options.policy());
}));
return new Result(counts[0], counts[1]);
}
}
@@ -100,14 +101,11 @@ final class OfflineRegionWriter {
}
}
record Options(HeadlessTerrainContext context, Path directory, int parallelism, RegionGenerationWindow.Policy policy) {
record Options(HeadlessTerrainContext context, Path directory, RegionGenerationWindow workers) {
Options {
Objects.requireNonNull(context, "context");
Objects.requireNonNull(directory, "directory");
Objects.requireNonNull(policy, "policy");
if (parallelism < 1 || parallelism > 32) {
throw new IllegalArgumentException("Offline writer parallelism must be 1..32");
}
Objects.requireNonNull(workers, "workers");
}
}
@@ -11,20 +11,55 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
final class RegionGenerationWindow {
final class RegionGenerationWindow implements AutoCloseable {
private static final int UNRESPONSIVE_WORKER_EXIT = 70;
private RegionGenerationWindow() {
private final int parallelism;
private final Policy policy;
private final ExecutorService workers;
private boolean closed;
private boolean processing;
RegionGenerationWindow(int parallelism, Policy policy) {
if (parallelism < 1 || parallelism > 32) {
throw new IllegalArgumentException("Generation workers must be 1..32");
}
this.parallelism = parallelism;
this.policy = Objects.requireNonNull(policy, "policy");
workers = Executors.newFixedThreadPool(parallelism);
}
static <T> void process(Request<T> request) throws Exception {
process(request, Policy.EMBEDDED);
@Override
public synchronized void close() {
if (processing) {
throw new IllegalStateException("Cannot close generation workers from an active batch");
}
if (closed) {
return;
}
closed = true;
workers.shutdown();
drain(workers, policy, null);
}
static <T> void process(Request<T> request, Policy policy) throws Exception {
ExecutorService workers = Executors.newFixedThreadPool(request.parallelism());
int parallelism() {
return parallelism;
}
synchronized void requireOpen() {
if (closed) {
throw new IllegalStateException("Generation workers are closed");
}
}
synchronized <T> void process(Request<T> request) throws Exception {
requireOpen();
if (processing || request.parallelism() > parallelism) {
throw new IllegalArgumentException("Generation batch requires idle workers within the session limit");
}
ExecutorCompletionService<Completed<T>> completions = new ExecutorCompletionService<>(workers);
Set<Future<Completed<T>>> active = new HashSet<>(request.parallelism());
processing = true;
try {
int submitted = 0;
for (; submitted < Math.min(request.count(), request.parallelism()); submitted++) {
@@ -43,15 +78,16 @@ final class RegionGenerationWindow {
request.sink().accept(result.index(), result.value());
}
} catch (Exception | Error failure) {
closed = true;
for (Future<Completed<T>> future : active) {
future.cancel(true);
}
workers.shutdownNow();
drain(workers, policy, failure);
throw failure;
} finally {
processing = false;
}
workers.shutdown();
drain(workers, policy, null);
}
private static void drain(ExecutorService workers, Policy policy, Throwable failure) {
@@ -55,15 +55,17 @@ final class DetachedNativeTerrain {
request.level(), request.level().palettedContainerFactory(), null)));
}
long started = System.nanoTime();
for (ChunkStatus status : ChunkStatus.getStatusList()) {
if (status == ChunkStatus.EMPTY) {
continue;
try (RegionGenerationWindow workers = new RegionGenerationWindow(request.parallelism(), RegionGenerationWindow.Policy.EMBEDDED)) {
for (ChunkStatus status : ChunkStatus.getStatusList()) {
if (status == ChunkStatus.EMPTY) {
continue;
}
if (status.isAfter(ChunkStatus.TERRAIN)) {
break;
}
ChunkStep step = ChunkPyramid.GENERATION_PYRAMID.getStepTo(status);
executeStage(new Stage(context, step, plan, holders, request.parallelism()), workers, progress);
}
if (status.isAfter(ChunkStatus.TERRAIN)) {
break;
}
ChunkStep step = ChunkPyramid.GENERATION_PYRAMID.getStepTo(status);
executeStage(new Stage(context, step, plan, holders, request.parallelism()), progress);
}
long serializationStarted = System.nanoTime();
write(request.level(), output, plan, holders);
@@ -72,7 +74,7 @@ final class DetachedNativeTerrain {
return new Result(request.width() * request.width(), plan.size(), seconds);
}
private static void executeStage(Stage stage, Consumer<String> progress) throws Exception {
private static void executeStage(Stage stage, RegionGenerationWindow workers, Consumer<String> progress) throws Exception {
long started = System.nanoTime();
List<PlannedChunk> eligible = new ArrayList<>();
for (PlannedChunk planned : stage.plan()) {
@@ -82,7 +84,7 @@ final class DetachedNativeTerrain {
}
AtomicInteger completed = new AtomicInteger();
AtomicInteger targetCompleted = new AtomicInteger();
RegionGenerationWindow.process(new RegionGenerationWindow.Request<>(eligible.size(), stage.parallelism(),
workers.process(new RegionGenerationWindow.Request<>(eligible.size(), stage.parallelism(),
index -> {
PlannedChunk planned = eligible.get(index);
apply(stage, planned);
@@ -5,14 +5,22 @@ import art.arcane.volmlib.nativelib.minecraft26_2.modded.NativeBlockResolver;
import art.arcane.volmlib.nativelib.terrain.NativeBlockState;
import net.minecraft.core.registries.BuiltInRegistries;
import net.minecraft.world.level.block.Block;
import net.minecraft.world.level.block.Blocks;
import net.minecraft.world.level.block.state.BlockState;
import org.junit.BeforeClass;
import org.junit.Test;
import org.mockito.MockedStatic;
import java.io.File;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.CALLS_REAL_METHODS;
import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.times;
public final class HeadlessNativeStateTest {
@BeforeClass
@@ -20,6 +28,21 @@ public final class HeadlessNativeStateTest {
HeadlessNativeBootstrap.initialize();
}
@Test
public void platformAirResolvesOnceAndRetainsCanonicalNativeState() {
BlockState air = Blocks.AIR.defaultBlockState();
try (MockedStatic<ModdedBlockState> states = mockStatic(ModdedBlockState.class, CALLS_REAL_METHODS)) {
HeadlessNativePlatform platform = new HeadlessNativePlatform(new File("unused-native-platform"),
HeadlessNativeTestRegistries::get);
NativeBlockState wrapped = platform.air();
for (int iteration = 0; iteration < 1024; iteration++) {
assertSame(wrapped, platform.air());
assertSame(air, ((ModdedBlockState) platform.air()).handle());
}
states.verify(() -> ModdedBlockState.of(air, null), times(1));
}
}
@Test
public void nativeMaterialsDistinguishDecorationsFluidsGlassAndBlockEntities() {
assertFalse(NativeBlockResolver.strictParse("minecraft:short_grass").isSolid());
@@ -56,8 +56,9 @@ public final class HeadlessRegionTerrainTest {
configuration, "probe", Level.OVERWORLD, root.resolve("configuration")));
try (HeadlessTerrainContext context = HeadlessTerrainContext.create(new HeadlessTerrainContext.Options(
session.engine(), runtime, root.resolve("templates"), Level.OVERWORLD,
NoiseGeneratorSettings.OVERWORLD, spigot))) {
HeadlessRegionTerrain.Session rolling = new HeadlessRegionTerrain.Session(RegionGenerationWindow.Policy.EMBEDDED);
NoiseGeneratorSettings.OVERWORLD, spigot));
RegionGenerationWindow workers = new RegionGenerationWindow(4, RegionGenerationWindow.Policy.EMBEDDED)) {
HeadlessRegionTerrain.Session rolling = new HeadlessRegionTerrain.Session(workers);
HeadlessRegionTerrain.Result result = rolling.generate(
new HeadlessRegionTerrain.Request(context, -3, 2, 1, 4), System.out::println);
OfflineRegionGenerator.Configuration export = new OfflineRegionGenerator.Configuration(
@@ -100,14 +101,14 @@ public final class HeadlessRegionTerrainTest {
assertEquals(ChunkStatus.TERRAIN, retained.getPersistedStatus());
Path regionDirectory = Files.createDirectory(root.resolve("region"));
OfflineRegionWriter writer = new OfflineRegionWriter(new OfflineRegionWriter.Options(
context, regionDirectory, 4, RegionGenerationWindow.Policy.EMBEDDED));
context, regionDirectory, workers));
OfflineRegionWriter.Result firstWrite = writer.write(result);
assertEquals(result.updated().size(), firstWrite.chunks());
assertEquals(tag, read(regionDirectory, target.getPos()));
OfflineRegionWriter.Result duplicateWrite = writer.write(result);
assertEquals(0, duplicateWrite.chunks());
OfflineRegionWriter reopened = new OfflineRegionWriter(new OfflineRegionWriter.Options(
context, regionDirectory, 4, RegionGenerationWindow.Policy.EMBEDDED));
context, regionDirectory, workers));
assertEquals(0, reopened.write(result).chunks());
target.setPersistedStatus(ChunkStatus.STRUCTURE_STARTS);
try {
@@ -138,7 +139,7 @@ public final class HeadlessRegionTerrainTest {
assertEquals(0, writer.write(back).chunks());
Path oversizedDirectory = Files.createDirectory(root.resolve("oversized"));
OfflineRegionWriter oversizedWriter = new OfflineRegionWriter(new OfflineRegionWriter.Options(
context, oversizedDirectory, 4, RegionGenerationWindow.Policy.EMBEDDED));
context, oversizedDirectory, workers));
NamespacedKey payloadKey = new NamespacedKey("iris", "serialization_test_payload");
byte[] payload = new byte[2 * 1024 * 1024];
new Random(69420L).nextBytes(payload);
@@ -151,14 +152,19 @@ public final class HeadlessRegionTerrainTest {
regenerated.persistentDataContainer.remove(payloadKey);
Path failedDirectory = Files.createDirectory(root.resolve("failed"));
OfflineRegionWriter failedWriter = new OfflineRegionWriter(new OfflineRegionWriter.Options(
context, failedDirectory, 4, RegionGenerationWindow.Policy.EMBEDDED));
context, failedDirectory, workers));
regenerated.setPersistedStatus(ChunkStatus.FULL);
try {
assertThrows(Exception.class, () -> failedWriter.write(singleton));
} finally {
regenerated.setPersistedStatus(ChunkStatus.TERRAIN);
}
assertEquals(1, failedWriter.write(singleton).chunks());
assertThrows(IllegalStateException.class, () -> failedWriter.write(singleton));
try (RegionGenerationWindow replacement = new RegionGenerationWindow(4, RegionGenerationWindow.Policy.EMBEDDED)) {
OfflineRegionWriter retried = new OfflineRegionWriter(new OfflineRegionWriter.Options(
context, failedDirectory, replacement));
assertEquals(1, retried.write(singleton).chunks());
}
assertEquals("minecraft:terrain", read(failedDirectory, regenerated.getPos()).getStringOr("Status", ""));
}
}
@@ -0,0 +1,83 @@
package art.arcane.iris.probe;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.mockito.MockedStatic;
import java.io.IOException;
import java.nio.channels.FileChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import static org.mockito.AdditionalAnswers.delegatesTo;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
public final class OfflineRegionPublicationTest {
@Rule
public TemporaryFolder temporary = new TemporaryFolder();
@Test
public void unpublishedRuntimeRejectsWorldRootsOutsideItsStagingDirectory() throws Exception {
Path staging = temporary.newFolder("staging").toPath();
Path outside = temporary.getRoot().toPath().resolve("outside");
try (HeadlessNativeRuntime runtime = new HeadlessNativeRuntime(staging)) {
assertThrows(IOException.class, () -> runtime.createHistory(
new RealPackProbeSupport.HistoryRequest(outside, null, null, 1L)));
}
assertFalse(Files.exists(outside));
}
@Test
public void publishesOnlyAfterEveryFileAndDirectoryIsForced() throws Exception {
for (boolean fail : List.of(true, false)) {
Path output = temporary.newFolder().toPath();
Path staged = Files.createDirectory(output.resolve("incomplete"));
Path nested = Files.createDirectory(staged.resolve("world"));
Path data = Files.writeString(nested.resolve("data"), "checkpoint");
Path target = output.resolve("native-terrain-checkpoint");
List<Path> forced = new ArrayList<>();
try (MockedStatic<FileChannel> ignored = mockStatic(FileChannel.class, invocation -> {
if (invocation.getMethod().getParameterCount() != 2) {
return invocation.callRealMethod();
}
FileChannel actual = (FileChannel) invocation.callRealMethod();
Path path = invocation.getArgument(0);
FileChannel channel = mock(FileChannel.class, delegatesTo(actual));
doAnswer(force -> {
if (path.equals(output)) {
assertTrue(Files.isDirectory(target));
assertFalse(Files.exists(staged));
} else {
assertFalse(Files.exists(target));
}
if (fail && path.equals(data)) {
throw new IOException("Expected publication sync failure");
}
actual.force(true);
forced.add(path);
return null;
}).when(channel).force(true);
return channel;
})) {
if (fail) {
assertThrows(IOException.class, () -> OfflineRegionGenerator.publish(staged, target));
assertTrue(Files.isRegularFile(data));
assertFalse(Files.exists(target));
} else {
OfflineRegionGenerator.publish(staged, target);
assertEquals(List.of(data, nested, staged, output), forced);
assertEquals("checkpoint", Files.readString(target.resolve("world/data")));
}
}
}
}
}
@@ -9,6 +9,8 @@ import java.nio.file.Path;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -20,27 +22,96 @@ import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
public final class RegionGenerationWindowTest {
@Test
public void retainsWorkersAndThreadLocalScratchAcrossBoundedBatches() throws Exception {
Set<Thread> firstWorkers = ConcurrentHashMap.newKeySet();
AtomicInteger allocations = new AtomicInteger();
ThreadLocal<Integer> scratch = ThreadLocal.withInitial(allocations::incrementAndGet);
CountDownLatch bothStarted = new CountDownLatch(2);
AtomicInteger active = new AtomicInteger();
AtomicInteger maximum = new AtomicInteger();
RegionGenerationWindow workers = new RegionGenerationWindow(2, RegionGenerationWindow.Policy.EMBEDDED);
try (workers) {
workers.process(new RegionGenerationWindow.Request<>(2, 2, index -> {
firstWorkers.add(Thread.currentThread());
scratch.get();
bothStarted.countDown();
assertTrue(bothStarted.await(5, TimeUnit.SECONDS));
return index;
}, (index, value) -> { }));
assertEquals(2, allocations.get());
workers.process(new RegionGenerationWindow.Request<>(16, 1, index -> {
maximum.accumulateAndGet(active.incrementAndGet(), Math::max);
try {
assertTrue(firstWorkers.contains(Thread.currentThread()));
return scratch.get();
} finally {
active.decrementAndGet();
}
}, (index, value) -> assertTrue(value > 0)));
assertEquals(2, allocations.get());
assertEquals(1, maximum.get());
}
for (Thread worker : firstWorkers) {
worker.join(1000L);
assertFalse(worker.isAlive());
}
workers.close();
assertThrows(IllegalStateException.class, () -> workers.process(
new RegionGenerationWindow.Request<>(1, 1, index -> index, (index, value) -> { })));
}
@Test(timeout = 5000)
public void sinkFailureDrainsWorkersAndPermanentlyClosesSession() {
CountDownLatch entered = new CountDownLatch(1);
AtomicInteger exited = new AtomicInteger();
try (RegionGenerationWindow workers = new RegionGenerationWindow(2, RegionGenerationWindow.Policy.EMBEDDED)) {
assertThrows(IllegalStateException.class, () -> workers.process(
new RegionGenerationWindow.Request<>(2, 2, index -> {
if (index == 0) {
assertTrue(entered.await(2, TimeUnit.SECONDS));
return index;
}
entered.countDown();
try {
new CountDownLatch(1).await();
} finally {
exited.incrementAndGet();
}
return index;
}, (index, value) -> {
throw new IllegalStateException("Expected sink failure");
})));
assertEquals(1, exited.get());
assertThrows(IllegalStateException.class, workers::requireOpen);
assertThrows(IllegalStateException.class, () -> workers.process(
new RegionGenerationWindow.Request<>(1, 1, index -> index, (index, value) -> { })));
}
}
@Test
public void replenishesWorkersWhileEarlierChunksRemainBlocked() throws Exception {
CountDownLatch replacementStarted = new CountDownLatch(1);
AtomicInteger active = new AtomicInteger();
AtomicInteger maximum = new AtomicInteger();
List<Integer> completed = new ArrayList<>();
RegionGenerationWindow.process(new RegionGenerationWindow.Request<>(16, 2, index -> {
int running = active.incrementAndGet();
maximum.accumulateAndGet(running, Math::max);
try {
if (index == 0 && !replacementStarted.await(5, TimeUnit.SECONDS)) {
throw new IllegalStateException("A completed worker was not replenished");
try (RegionGenerationWindow workers = new RegionGenerationWindow(2, RegionGenerationWindow.Policy.EMBEDDED)) {
workers.process(new RegionGenerationWindow.Request<>(16, 2, index -> {
int running = active.incrementAndGet();
maximum.accumulateAndGet(running, Math::max);
try {
if (index == 0 && !replacementStarted.await(5, TimeUnit.SECONDS)) {
throw new IllegalStateException("A completed worker was not replenished");
}
if (index == 2) {
replacementStarted.countDown();
}
return index;
} finally {
active.decrementAndGet();
}
if (index == 2) {
replacementStarted.countDown();
}
return index;
} finally {
active.decrementAndGet();
}
}, (index, value) -> completed.add(value)));
}, (index, value) -> completed.add(value)));
}
assertEquals(16, completed.size());
assertTrue(completed.indexOf(1) < completed.indexOf(0));
assertTrue(maximum.get() <= 2);
@@ -53,18 +124,20 @@ public final class RegionGenerationWindowTest {
AtomicInteger active = new AtomicInteger();
AtomicInteger maximum = new AtomicInteger();
AtomicInteger completed = new AtomicInteger();
RegionGenerationWindow.process(new RegionGenerationWindow.Request<>(97, 32, index -> {
maximum.accumulateAndGet(active.incrementAndGet(), Math::max);
try {
if (index < 32) {
initialWorkers.countDown();
assertTrue(initialWorkers.await(5, TimeUnit.SECONDS));
try (RegionGenerationWindow workers = new RegionGenerationWindow(32, RegionGenerationWindow.Policy.EMBEDDED)) {
workers.process(new RegionGenerationWindow.Request<>(97, 32, index -> {
maximum.accumulateAndGet(active.incrementAndGet(), Math::max);
try {
if (index < 32) {
initialWorkers.countDown();
assertTrue(initialWorkers.await(5, TimeUnit.SECONDS));
}
return index;
} finally {
active.decrementAndGet();
}
return index;
} finally {
active.decrementAndGet();
}
}, (index, value) -> completed.incrementAndGet()));
}, (index, value) -> completed.incrementAndGet()));
}
assertEquals(32, maximum.get());
assertEquals(97, completed.get());
assertEquals(0, active.get());
@@ -76,21 +149,23 @@ public final class RegionGenerationWindowTest {
public void interruptsAndDrainsOtherWorkersAfterFailure() {
CountDownLatch blocked = new CountDownLatch(1);
AtomicInteger interrupted = new AtomicInteger();
assertThrows(Exception.class, () -> RegionGenerationWindow.process(
new RegionGenerationWindow.Request<>(2, 2, index -> {
if (index == 0) {
assertTrue(blocked.await(5, TimeUnit.SECONDS));
throw new IllegalStateException("Expected fixture failure");
}
blocked.countDown();
try {
new CountDownLatch(1).await();
} catch (InterruptedException failure) {
interrupted.incrementAndGet();
Thread.currentThread().interrupt();
}
return index;
}, (index, value) -> { })));
try (RegionGenerationWindow workers = new RegionGenerationWindow(2, RegionGenerationWindow.Policy.EMBEDDED)) {
assertThrows(Exception.class, () -> workers.process(
new RegionGenerationWindow.Request<>(2, 2, index -> {
if (index == 0) {
assertTrue(blocked.await(5, TimeUnit.SECONDS));
throw new IllegalStateException("Expected fixture failure");
}
blocked.countDown();
try {
new CountDownLatch(1).await();
} catch (InterruptedException failure) {
interrupted.incrementAndGet();
Thread.currentThread().interrupt();
}
return index;
}, (index, value) -> { })));
}
assertEquals(1, interrupted.get());
}
@@ -98,16 +173,18 @@ public final class RegionGenerationWindowTest {
public void drainsStartedWorkBeforeReportingFailure() {
CountDownLatch secondStarted = new CountDownLatch(1);
AtomicInteger finished = new AtomicInteger();
assertThrows(Exception.class, () -> RegionGenerationWindow.process(
new RegionGenerationWindow.Request<>(4, 2, index -> {
if (index == 0) {
assertTrue(secondStarted.await(5, TimeUnit.SECONDS));
throw new IllegalStateException("Expected fixture failure");
}
secondStarted.countDown();
finished.incrementAndGet();
return index;
}, (index, value) -> { })));
try (RegionGenerationWindow workers = new RegionGenerationWindow(2, RegionGenerationWindow.Policy.EMBEDDED)) {
assertThrows(Exception.class, () -> workers.process(
new RegionGenerationWindow.Request<>(4, 2, index -> {
if (index == 0) {
assertTrue(secondStarted.await(5, TimeUnit.SECONDS));
throw new IllegalStateException("Expected fixture failure");
}
secondStarted.countDown();
finished.incrementAndGet();
return index;
}, (index, value) -> { })));
}
assertTrue(finished.get() >= 1);
}
@@ -116,15 +193,17 @@ public final class RegionGenerationWindowTest {
AtomicInteger completed = new AtomicInteger();
RegionGenerationWindow.Policy policy = new RegionGenerationWindow.Policy(
Duration.ofMillis(50), Duration.ofSeconds(1), false);
assertThrows(TimeoutException.class, () -> RegionGenerationWindow.process(
new RegionGenerationWindow.Request<>(1, 1, index -> {
try {
new CountDownLatch(1).await();
} finally {
completed.incrementAndGet();
}
return index;
}, (index, value) -> { }), policy));
try (RegionGenerationWindow workers = new RegionGenerationWindow(1, policy)) {
assertThrows(TimeoutException.class, () -> workers.process(
new RegionGenerationWindow.Request<>(1, 1, index -> {
try {
new CountDownLatch(1).await();
} finally {
completed.incrementAndGet();
}
return index;
}, (index, value) -> { })));
}
assertEquals(1, completed.get());
}
@@ -160,8 +239,8 @@ public final class RegionGenerationWindowTest {
CountDownLatch never = new CountDownLatch(1);
RegionGenerationWindow.Policy policy = new RegionGenerationWindow.Policy(
Duration.ofSeconds(1), Duration.ofMillis(25), true);
try {
RegionGenerationWindow.process(new RegionGenerationWindow.Request<>(2, 2, index -> {
try (RegionGenerationWindow workers = new RegionGenerationWindow(2, policy)) {
workers.process(new RegionGenerationWindow.Request<>(2, 2, index -> {
if (index == 0) {
if (!entered.await(1, TimeUnit.SECONDS)) {
throw new IllegalStateException("Worker did not start");
@@ -177,7 +256,7 @@ public final class RegionGenerationWindowTest {
}
return index;
}
}, (index, value) -> { }), policy);
}, (index, value) -> { }));
} finally {
Files.writeString(Path.of(arguments[0]), "closed");
}
@@ -125,8 +125,8 @@ public final class RegionNativeFixture extends JavaPlugin {
int chunks = request.width() * request.width();
getLogger().info("NATIVE_LOAD_STARTED chunks=" + chunks + " chunkX=" + request.chunkX()
+ " chunkZ=" + request.chunkZ() + " workers=" + request.parallelism());
try {
RegionGenerationWindow.process(new RegionGenerationWindow.Request<>(chunks, request.parallelism(),
try (RegionGenerationWindow workers = new RegionGenerationWindow(request.parallelism(), RegionGenerationWindow.Policy.EMBEDDED)) {
workers.process(new RegionGenerationWindow.Request<>(chunks, request.parallelism(),
index -> {
int x = request.chunkX() + index % request.width();
int z = request.chunkZ() + index / request.width();