mirror of
https://github.com/VolmitSoftware/Iris.git
synced 2026-10-26 14:00:26 +00:00
Group commit generation history
- Semantic journal and saved-biome appends write without forcing; a shared flusher forces touched files within 2s outside every store lock - GenerationHistory.sync forces pending history at world save, engine shutdown, pregen completion and before activation promotion - A failed deferred force poisons the store the same way an uncertain append did - Cache the active epoch pack root instead of regex-validating it on every stage open - Tests pin append-without-force, sync barriers, failure poisoning and the save, shutdown and pregen hooks
This commit is contained in:
1 parent
241260a762
commit
a35ecca7bd
24 files changed
+877
-136
No files matched your search
+5
@@ -116,6 +116,11 @@ public final class ModdedPregenMethod implements PregeneratorMethod {
|
|||||||
}
|
}
|
||||||
ModdedIrisLog.info("Iris modded pregen done: dim={} completed={} peakInFlight={} finalLimit={}",
|
ModdedIrisLog.info("Iris modded pregen done: dim={} completed={} peakInFlight={} finalLimit={}",
|
||||||
runtime.worldIdentity(), completed.get(), inFlightPeak.get(), adaptiveLimit.get());
|
runtime.worldIdentity(), completed.get(), inFlightPeak.get(), adaptiveLimit.get());
|
||||||
|
try {
|
||||||
|
engine.syncGenerationHistory();
|
||||||
|
} catch (RuntimeException e) {
|
||||||
|
ModdedIrisLog.error("Iris modded pregen could not force generation history to stable storage", e);
|
||||||
|
}
|
||||||
if (deferFinalSaveIfRequested()) {
|
if (deferFinalSaveIfRequested()) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -250,6 +250,7 @@ public interface Engine extends DataProvider, Fallible, BlockUpdater, Renderer,
|
|||||||
|
|
||||||
default void save() {
|
default void save() {
|
||||||
NativeStructureOwnershipStore.flush(this);
|
NativeStructureOwnershipStore.flush(this);
|
||||||
|
syncGenerationHistory();
|
||||||
getMantle().save();
|
getMantle().save();
|
||||||
getWorldManager().onSave();
|
getWorldManager().onSave();
|
||||||
saveEngineData();
|
saveEngineData();
|
||||||
@@ -259,6 +260,7 @@ public interface Engine extends DataProvider, Fallible, BlockUpdater, Renderer,
|
|||||||
|
|
||||||
default void saveNow() {
|
default void saveNow() {
|
||||||
NativeStructureOwnershipStore.flush(this);
|
NativeStructureOwnershipStore.flush(this);
|
||||||
|
syncGenerationHistory();
|
||||||
getMantle().saveAllNow();
|
getMantle().saveAllNow();
|
||||||
saveEngineData();
|
saveEngineData();
|
||||||
}
|
}
|
||||||
@@ -267,6 +269,8 @@ public interface Engine extends DataProvider, Fallible, BlockUpdater, Renderer,
|
|||||||
|
|
||||||
void saveEngineData();
|
void saveEngineData();
|
||||||
|
|
||||||
|
void syncGenerationHistory();
|
||||||
|
|
||||||
default String getName() {
|
default String getName() {
|
||||||
return getDimension().getName();
|
return getDimension().getName();
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1529,6 +1529,7 @@ public class IrisEngine implements Engine {
|
|||||||
|
|
||||||
private void scheduleWorldSave() {
|
private void scheduleWorldSave() {
|
||||||
backgroundTasks.scheduleAdmittedTask(this, () -> NativeStructureOwnershipStore.flush(this));
|
backgroundTasks.scheduleAdmittedTask(this, () -> NativeStructureOwnershipStore.flush(this));
|
||||||
|
backgroundTasks.scheduleAdmittedTask(this, this::syncGenerationHistory);
|
||||||
getMantle().save();
|
getMantle().save();
|
||||||
getWorldManager().onSave();
|
getWorldManager().onSave();
|
||||||
saveEngineData();
|
saveEngineData();
|
||||||
@@ -1539,6 +1540,20 @@ public class IrisEngine implements Engine {
|
|||||||
engineDataStore.saveEngineData();
|
engineDataStore.saveEngineData();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void syncGenerationHistory() {
|
||||||
|
GenerationHistoryRuntimeRouter router = getGenerationHistoryRuntimeRouter().orElse(null);
|
||||||
|
if (router == null) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
router.history().sync();
|
||||||
|
} catch (IOException failure) {
|
||||||
|
throw new IllegalStateException("Unable to force Iris generation history to stable storage for "
|
||||||
|
+ getWorld().name() + ".", failure);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void blockUpdatedMetric() {
|
public void blockUpdatedMetric() {
|
||||||
bud.incrementAndGet();
|
bud.incrementAndGet();
|
||||||
|
|||||||
@@ -82,6 +82,27 @@ public final class GenerationHistory {
|
|||||||
return savedBiomes;
|
return savedBiomes;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public void sync() throws IOException {
|
||||||
|
IOException failure = null;
|
||||||
|
try {
|
||||||
|
semantics.sync();
|
||||||
|
} catch (IOException semanticFailure) {
|
||||||
|
failure = semanticFailure;
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
savedBiomes.sync();
|
||||||
|
} catch (IOException biomeFailure) {
|
||||||
|
if (failure == null) {
|
||||||
|
failure = biomeFailure;
|
||||||
|
} else {
|
||||||
|
failure.addSuppressed(biomeFailure);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (failure != null) {
|
||||||
|
throw failure;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
public static GenerationHistory create(
|
public static GenerationHistory create(
|
||||||
Path dimensionRoot,
|
Path dimensionRoot,
|
||||||
Path packSource,
|
Path packSource,
|
||||||
@@ -478,6 +499,7 @@ public final class GenerationHistory {
|
|||||||
private GenerationActivation promotePendingLocked(
|
private GenerationActivation promotePendingLocked(
|
||||||
BoundarySignatureCapture signatureCapture, boolean startupRecovery, boolean validateSemantics
|
BoundarySignatureCapture signatureCapture, boolean startupRecovery, boolean validateSemantics
|
||||||
) throws IOException {
|
) throws IOException {
|
||||||
|
sync();
|
||||||
WorldChunkInventory inventory = recoverUnstoredClaims(startupRecovery);
|
WorldChunkInventory inventory = recoverUnstoredClaims(startupRecovery);
|
||||||
validateReferencedState(Optional.empty(), validateSemantics);
|
validateReferencedState(Optional.empty(), validateSemantics);
|
||||||
Optional<GenerationActivation> pending = store.pendingActivation();
|
Optional<GenerationActivation> pending = store.pendingActivation();
|
||||||
|
|||||||
@@ -26,6 +26,7 @@ public final class GenerationHistoryPaths {
|
|||||||
private final Path legacyPackRoot;
|
private final Path legacyPackRoot;
|
||||||
private final Path legacyMantleRoot;
|
private final Path legacyMantleRoot;
|
||||||
private final Path regionRoot;
|
private final Path regionRoot;
|
||||||
|
private volatile PackRoot lastPackRoot;
|
||||||
|
|
||||||
private GenerationHistoryPaths(Path dimensionRoot) {
|
private GenerationHistoryPaths(Path dimensionRoot) {
|
||||||
this.dimensionRoot = Objects.requireNonNull(dimensionRoot, "dimensionRoot")
|
this.dimensionRoot = Objects.requireNonNull(dimensionRoot, "dimensionRoot")
|
||||||
@@ -71,7 +72,13 @@ public final class GenerationHistoryPaths {
|
|||||||
}
|
}
|
||||||
|
|
||||||
public Path packRoot(String epochId) {
|
public Path packRoot(String epochId) {
|
||||||
return epochRoot(epochId).resolve(LEGACY_PACK_DIRECTORY_NAME);
|
PackRoot cached = lastPackRoot;
|
||||||
|
if (cached != null && cached.epochId().equals(epochId)) {
|
||||||
|
return cached.path();
|
||||||
|
}
|
||||||
|
Path resolved = epochRoot(epochId).resolve(LEGACY_PACK_DIRECTORY_NAME);
|
||||||
|
lastPackRoot = new PackRoot(epochId, resolved);
|
||||||
|
return resolved;
|
||||||
}
|
}
|
||||||
|
|
||||||
public Path activationsRoot() {
|
public Path activationsRoot() {
|
||||||
@@ -108,4 +115,7 @@ public final class GenerationHistoryPaths {
|
|||||||
public Path regionRoot() {
|
public Path regionRoot() {
|
||||||
return regionRoot;
|
return regionRoot;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private record PackRoot(String epochId, Path path) {
|
||||||
|
}
|
||||||
}
|
}
|
||||||
@@ -500,6 +500,11 @@ public final class GenerationHistoryRuntimeRouter implements AutoCloseable {
|
|||||||
} catch (Throwable detachFailure) {
|
} catch (Throwable detachFailure) {
|
||||||
failure = detachFailure;
|
failure = detachFailure;
|
||||||
}
|
}
|
||||||
|
try {
|
||||||
|
history.sync();
|
||||||
|
} catch (Throwable syncFailure) {
|
||||||
|
failure = appendFailure(failure, syncFailure);
|
||||||
|
}
|
||||||
try {
|
try {
|
||||||
biomes.close();
|
biomes.close();
|
||||||
} catch (Throwable biomeFailure) {
|
} catch (Throwable biomeFailure) {
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package art.arcane.iris.world.history;
|
|||||||
|
|
||||||
import art.arcane.iris.generation.hydrology.HydrologyFeatureType;
|
import art.arcane.iris.generation.hydrology.HydrologyFeatureType;
|
||||||
import art.arcane.iris.spi.IrisLogging;
|
import art.arcane.iris.spi.IrisLogging;
|
||||||
|
import art.arcane.iris.world.storage.DeferredDurability;
|
||||||
import art.arcane.iris.world.storage.Durability;
|
import art.arcane.iris.world.storage.Durability;
|
||||||
import it.unimi.dsi.fastutil.longs.Long2IntOpenHashMap;
|
import it.unimi.dsi.fastutil.longs.Long2IntOpenHashMap;
|
||||||
import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
|
import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
|
||||||
@@ -93,6 +94,7 @@ public final class GenerationSemanticIndex {
|
|||||||
private final PointerPublisher pointerPublisher;
|
private final PointerPublisher pointerPublisher;
|
||||||
private final CatalogPublisher catalogPublisher;
|
private final CatalogPublisher catalogPublisher;
|
||||||
private final boolean unpublished;
|
private final boolean unpublished;
|
||||||
|
private final DeferredDurability durability;
|
||||||
private long regionDecodeCount;
|
private long regionDecodeCount;
|
||||||
private volatile IOException journalFailure;
|
private volatile IOException journalFailure;
|
||||||
private long appendingRegionKey;
|
private long appendingRegionKey;
|
||||||
@@ -118,6 +120,8 @@ public final class GenerationSemanticIndex {
|
|||||||
this.pointerPublisher = pointerPublisher;
|
this.pointerPublisher = pointerPublisher;
|
||||||
this.catalogPublisher = catalogPublisher;
|
this.catalogPublisher = catalogPublisher;
|
||||||
this.unpublished = unpublished;
|
this.unpublished = unpublished;
|
||||||
|
this.durability = new DeferredDurability(
|
||||||
|
DeferredDurability.DEFAULT_FLUSH_DELAY_MILLIS, failure -> journalFailure = failure);
|
||||||
}
|
}
|
||||||
|
|
||||||
public static GenerationSemanticIndex load(Path dimensionRoot) throws IOException {
|
public static GenerationSemanticIndex load(Path dimensionRoot) throws IOException {
|
||||||
@@ -265,7 +269,7 @@ public final class GenerationSemanticIndex {
|
|||||||
appendingBase = baseRegion;
|
appendingBase = baseRegion;
|
||||||
publicationLock.unlock();
|
publicationLock.unlock();
|
||||||
try {
|
try {
|
||||||
SemanticJournal.append(directory, regionX, regionZ, updates, unpublished);
|
SemanticJournal.append(directory, regionX, regionZ, updates, unpublished, durability);
|
||||||
} catch (JournalAppendFailure failure) {
|
} catch (JournalAppendFailure failure) {
|
||||||
journalFailure = failure;
|
journalFailure = failure;
|
||||||
throw failure;
|
throw failure;
|
||||||
@@ -289,6 +293,15 @@ public final class GenerationSemanticIndex {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public void sync() throws IOException {
|
||||||
|
try {
|
||||||
|
durability.sync();
|
||||||
|
} catch (IOException failure) {
|
||||||
|
journalFailure = failure;
|
||||||
|
throw failure;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
public int discardUnstoredClaims(WorldChunkInventory stored, Set<Long> activationIds) throws IOException {
|
public int discardUnstoredClaims(WorldChunkInventory stored, Set<Long> activationIds) throws IOException {
|
||||||
Objects.requireNonNull(stored, "stored chunks");
|
Objects.requireNonNull(stored, "stored chunks");
|
||||||
Set<Long> selected = Set.copyOf(activationIds);
|
Set<Long> selected = Set.copyOf(activationIds);
|
||||||
@@ -1940,7 +1953,8 @@ public final class GenerationSemanticIndex {
|
|||||||
int regionX,
|
int regionX,
|
||||||
int regionZ,
|
int regionZ,
|
||||||
List<ChunkGenerationSemantics> claims,
|
List<ChunkGenerationSemantics> claims,
|
||||||
boolean unpublished
|
boolean unpublished,
|
||||||
|
DeferredDurability durability
|
||||||
) throws IOException {
|
) throws IOException {
|
||||||
Path file = directory.resolve(journalFileName(regionX, regionZ));
|
Path file = directory.resolve(journalFileName(regionX, regionZ));
|
||||||
boolean created = !Files.exists(file, LinkOption.NOFOLLOW_LINKS);
|
boolean created = !Files.exists(file, LinkOption.NOFOLLOW_LINKS);
|
||||||
@@ -1974,10 +1988,10 @@ public final class GenerationSemanticIndex {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
if (!unpublished) {
|
if (!unpublished) {
|
||||||
Durability.force(channel);
|
durability.written(file);
|
||||||
}
|
}
|
||||||
if (created) {
|
if (created) {
|
||||||
RegionShard.forceDirectory(directory);
|
durability.linked(directory);
|
||||||
}
|
}
|
||||||
} catch (IOException failure) {
|
} catch (IOException failure) {
|
||||||
try {
|
try {
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
package art.arcane.iris.world.history;
|
package art.arcane.iris.world.history;
|
||||||
|
|
||||||
|
import art.arcane.iris.world.storage.DeferredDurability;
|
||||||
import art.arcane.iris.world.storage.Durability;
|
import art.arcane.iris.world.storage.Durability;
|
||||||
|
|
||||||
import java.io.BufferedInputStream;
|
import java.io.BufferedInputStream;
|
||||||
@@ -52,6 +53,7 @@ public final class SavedBiomeStore {
|
|||||||
private final Path dimensionRoot;
|
private final Path dimensionRoot;
|
||||||
private final Path directory;
|
private final Path directory;
|
||||||
private final boolean unpublished;
|
private final boolean unpublished;
|
||||||
|
private final DeferredDurability durability;
|
||||||
private final WriteStripe[] regionLocks = new WriteStripe[64];
|
private final WriteStripe[] regionLocks = new WriteStripe[64];
|
||||||
private final LinkedHashMap<Long, RegionIndex> regions = new LinkedHashMap<>(64, 0.75F, true);
|
private final LinkedHashMap<Long, RegionIndex> regions = new LinkedHashMap<>(64, 0.75F, true);
|
||||||
private final LinkedHashMap<Long, Optional<SavedBiomeChunk>> chunks = new LinkedHashMap<>(128, 0.75F, true);
|
private final LinkedHashMap<Long, Optional<SavedBiomeChunk>> chunks = new LinkedHashMap<>(128, 0.75F, true);
|
||||||
@@ -61,6 +63,8 @@ public final class SavedBiomeStore {
|
|||||||
private SavedBiomeStore(Path dimensionRoot, boolean unpublished) throws IOException {
|
private SavedBiomeStore(Path dimensionRoot, boolean unpublished) throws IOException {
|
||||||
this.dimensionRoot = Objects.requireNonNull(dimensionRoot, "dimensionRoot").toAbsolutePath().normalize();
|
this.dimensionRoot = Objects.requireNonNull(dimensionRoot, "dimensionRoot").toAbsolutePath().normalize();
|
||||||
this.unpublished = unpublished;
|
this.unpublished = unpublished;
|
||||||
|
this.durability = new DeferredDurability(
|
||||||
|
DeferredDurability.DEFAULT_FLUSH_DELAY_MILLIS, failure -> writeFailure = failure);
|
||||||
directory = this.dimensionRoot.resolve("iris/generation/biomes");
|
directory = this.dimensionRoot.resolve("iris/generation/biomes");
|
||||||
requireSafeAncestors();
|
requireSafeAncestors();
|
||||||
for (int index = 0; index < regionLocks.length; index++) {
|
for (int index = 0; index < regionLocks.length; index++) {
|
||||||
@@ -227,7 +231,7 @@ public final class SavedBiomeStore {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
if (!unpublished) {
|
if (!unpublished) {
|
||||||
Durability.force(output.getChannel());
|
durability.written(region.path);
|
||||||
}
|
}
|
||||||
appendCompleted = true;
|
appendCompleted = true;
|
||||||
} catch (IOException failure) {
|
} catch (IOException failure) {
|
||||||
@@ -269,6 +273,15 @@ public final class SavedBiomeStore {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public void sync() throws IOException {
|
||||||
|
try {
|
||||||
|
durability.sync();
|
||||||
|
} catch (IOException failure) {
|
||||||
|
writeFailure = failure;
|
||||||
|
throw failure;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
public int discardUnstoredClaims(WorldChunkInventory inventory, Set<Long> selectedActivations) throws IOException {
|
public int discardUnstoredClaims(WorldChunkInventory inventory, Set<Long> selectedActivations) throws IOException {
|
||||||
Objects.requireNonNull(inventory, "inventory");
|
Objects.requireNonNull(inventory, "inventory");
|
||||||
Set<Long> selected = Set.copyOf(selectedActivations);
|
Set<Long> selected = Set.copyOf(selectedActivations);
|
||||||
|
|||||||
@@ -779,6 +779,7 @@ public class AsyncPregenMethod implements PregeneratorMethod {
|
|||||||
TimeUnit.SECONDS,
|
TimeUnit.SECONDS,
|
||||||
() -> IrisLogging.warn("Async pregen is still draining outstanding chunks. " + metricsSnapshot())
|
() -> IrisLogging.warn("Async pregen is still draining outstanding chunks. " + metricsSnapshot())
|
||||||
);
|
);
|
||||||
|
syncGenerationHistory();
|
||||||
|
|
||||||
mantleCleanup.close(MANTLE_CLEANUP_DRAIN_SECONDS, TimeUnit.SECONDS);
|
mantleCleanup.close(MANTLE_CLEANUP_DRAIN_SECONDS, TimeUnit.SECONDS);
|
||||||
flushAllRemainingChunks();
|
flushAllRemainingChunks();
|
||||||
@@ -807,6 +808,19 @@ public class AsyncPregenMethod implements PregeneratorMethod {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private void syncGenerationHistory() {
|
||||||
|
Engine engine = resolveMetricsEngine();
|
||||||
|
if (engine == null) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
engine.syncGenerationHistory();
|
||||||
|
} catch (Throwable e) {
|
||||||
|
IrisLogging.reportError("Async pregen could not force generation history to stable storage for world "
|
||||||
|
+ world.getName() + ".", e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private boolean isCancelled() {
|
private boolean isCancelled() {
|
||||||
return closing.get() || Thread.currentThread().isInterrupted();
|
return closing.get() || Thread.currentThread().isInterrupted();
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -112,6 +112,19 @@ public class MedievalPregenMethod implements PregeneratorMethod {
|
|||||||
return cached;
|
return cached;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private void syncGenerationHistory() {
|
||||||
|
Engine engine = resolveEngine();
|
||||||
|
if (engine == null) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
engine.syncGenerationHistory();
|
||||||
|
} catch (Throwable e) {
|
||||||
|
IrisLogging.reportError("Medieval pregen could not force generation history to stable storage for world "
|
||||||
|
+ world.getName() + ".", e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private void waitForChunks() {
|
private void waitForChunks() {
|
||||||
for (CompletableFuture<?> i : futures) {
|
for (CompletableFuture<?> i : futures) {
|
||||||
try {
|
try {
|
||||||
@@ -179,6 +192,7 @@ public class MedievalPregenMethod implements PregeneratorMethod {
|
|||||||
if (prefetchPool != null) {
|
if (prefetchPool != null) {
|
||||||
prefetchPool.shutdownNow();
|
prefetchPool.shutdownNow();
|
||||||
}
|
}
|
||||||
|
syncGenerationHistory();
|
||||||
unloadAndSaveAllChunks(true);
|
unloadAndSaveAllChunks(true);
|
||||||
chunkFlush.close(UNLOAD_TIMEOUT_SECONDS, TimeUnit.SECONDS);
|
chunkFlush.close(UNLOAD_TIMEOUT_SECONDS, TimeUnit.SECONDS);
|
||||||
} finally {
|
} finally {
|
||||||
|
|||||||
@@ -0,0 +1,138 @@
|
|||||||
|
package art.arcane.iris.world.storage;
|
||||||
|
|
||||||
|
import java.io.File;
|
||||||
|
import java.io.IOException;
|
||||||
|
import java.nio.channels.FileChannel;
|
||||||
|
import java.nio.file.NoSuchFileException;
|
||||||
|
import java.nio.file.Path;
|
||||||
|
import java.nio.file.StandardOpenOption;
|
||||||
|
import java.util.Objects;
|
||||||
|
import java.util.Set;
|
||||||
|
import java.util.concurrent.ConcurrentHashMap;
|
||||||
|
import java.util.concurrent.ScheduledThreadPoolExecutor;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
import java.util.concurrent.atomic.AtomicBoolean;
|
||||||
|
import java.util.concurrent.locks.ReentrantLock;
|
||||||
|
import java.util.function.Consumer;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Group commit for append-only stores: writers record which files and directory entries they
|
||||||
|
* changed, and one background pass forces them shortly afterwards, outside every store lock.
|
||||||
|
* {@link #sync()} is the durability barrier for saves, shutdown and pregeneration completion.
|
||||||
|
*/
|
||||||
|
public final class DeferredDurability {
|
||||||
|
public static final long DEFAULT_FLUSH_DELAY_MILLIS = 2_000L;
|
||||||
|
private static final ScheduledThreadPoolExecutor FLUSHER = createFlusher();
|
||||||
|
|
||||||
|
private final Set<Path> files = ConcurrentHashMap.newKeySet();
|
||||||
|
private final Set<Path> directories = ConcurrentHashMap.newKeySet();
|
||||||
|
private final ReentrantLock flushLock = new ReentrantLock();
|
||||||
|
private final AtomicBoolean scheduled = new AtomicBoolean();
|
||||||
|
private final long flushDelayMillis;
|
||||||
|
private final Consumer<IOException> backgroundFailure;
|
||||||
|
|
||||||
|
public DeferredDurability(long flushDelayMillis, Consumer<IOException> backgroundFailure) {
|
||||||
|
if (flushDelayMillis < 0L) {
|
||||||
|
throw new IllegalArgumentException("Flush delay must not be negative.");
|
||||||
|
}
|
||||||
|
this.flushDelayMillis = flushDelayMillis;
|
||||||
|
this.backgroundFailure = Objects.requireNonNull(backgroundFailure, "backgroundFailure");
|
||||||
|
}
|
||||||
|
|
||||||
|
public void written(Path file) {
|
||||||
|
if (!Durability.enabled()) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
files.add(file);
|
||||||
|
schedule();
|
||||||
|
}
|
||||||
|
|
||||||
|
public void linked(Path directory) {
|
||||||
|
if (!Durability.enabled() || File.separatorChar == '\\') {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
directories.add(directory);
|
||||||
|
schedule();
|
||||||
|
}
|
||||||
|
|
||||||
|
public void sync() throws IOException {
|
||||||
|
flushLock.lock();
|
||||||
|
try {
|
||||||
|
IOException failure = null;
|
||||||
|
for (Path file : files) {
|
||||||
|
files.remove(file);
|
||||||
|
try {
|
||||||
|
forceFile(file);
|
||||||
|
} catch (IOException fileFailure) {
|
||||||
|
failure = append(failure, fileFailure);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for (Path directory : directories) {
|
||||||
|
directories.remove(directory);
|
||||||
|
try {
|
||||||
|
forceDirectory(directory);
|
||||||
|
} catch (IOException directoryFailure) {
|
||||||
|
failure = append(failure, directoryFailure);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (failure != null) {
|
||||||
|
throw failure;
|
||||||
|
}
|
||||||
|
} finally {
|
||||||
|
flushLock.unlock();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private void schedule() {
|
||||||
|
if (scheduled.compareAndSet(false, true)) {
|
||||||
|
FLUSHER.schedule(this::flushInBackground, flushDelayMillis, TimeUnit.MILLISECONDS);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private void flushInBackground() {
|
||||||
|
scheduled.set(false);
|
||||||
|
try {
|
||||||
|
sync();
|
||||||
|
} catch (IOException failure) {
|
||||||
|
backgroundFailure.accept(failure);
|
||||||
|
} catch (RuntimeException failure) {
|
||||||
|
backgroundFailure.accept(new IOException("Deferred durability sync failed unexpectedly.", failure));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void forceFile(Path file) throws IOException {
|
||||||
|
try (FileChannel channel = FileChannel.open(file, StandardOpenOption.WRITE)) {
|
||||||
|
Durability.force(channel);
|
||||||
|
} catch (NoSuchFileException replaced) {
|
||||||
|
// Compaction or rewrite already published this file's contents durably under another name.
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void forceDirectory(Path directory) throws IOException {
|
||||||
|
try (FileChannel channel = FileChannel.open(directory, StandardOpenOption.READ)) {
|
||||||
|
Durability.force(channel);
|
||||||
|
} catch (UnsupportedOperationException failure) {
|
||||||
|
throw new IOException("Directory cannot be durability-synced: " + directory, failure);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static IOException append(IOException failure, IOException additional) {
|
||||||
|
if (failure == null) {
|
||||||
|
return additional;
|
||||||
|
}
|
||||||
|
failure.addSuppressed(additional);
|
||||||
|
return failure;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static ScheduledThreadPoolExecutor createFlusher() {
|
||||||
|
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1, runnable -> {
|
||||||
|
Thread thread = new Thread(runnable, "Iris Durability Flush");
|
||||||
|
thread.setDaemon(true);
|
||||||
|
return thread;
|
||||||
|
});
|
||||||
|
executor.setKeepAliveTime(30L, TimeUnit.SECONDS);
|
||||||
|
executor.allowCoreThreadTimeOut(true);
|
||||||
|
executor.setRemoveOnCancelPolicy(true);
|
||||||
|
return executor;
|
||||||
|
}
|
||||||
|
}
|
||||||
+54
@@ -0,0 +1,54 @@
|
|||||||
|
package art.arcane.iris.generation.runtime;
|
||||||
|
|
||||||
|
import art.arcane.iris.world.history.GenerationHistory;
|
||||||
|
import art.arcane.iris.world.history.GenerationHistoryRuntimeRouter;
|
||||||
|
import art.arcane.iris.world.IrisWorld;
|
||||||
|
import org.junit.Test;
|
||||||
|
|
||||||
|
import java.io.IOException;
|
||||||
|
import java.util.Optional;
|
||||||
|
|
||||||
|
import static org.junit.Assert.assertSame;
|
||||||
|
import static org.junit.Assert.assertThrows;
|
||||||
|
import static org.mockito.Mockito.doCallRealMethod;
|
||||||
|
import static org.mockito.Mockito.doThrow;
|
||||||
|
import static org.mockito.Mockito.mock;
|
||||||
|
import static org.mockito.Mockito.verify;
|
||||||
|
import static org.mockito.Mockito.when;
|
||||||
|
|
||||||
|
public class IrisEngineGenerationHistorySyncTest {
|
||||||
|
@Test
|
||||||
|
public void syncForcesTheAttachedHistory() throws Exception {
|
||||||
|
GenerationHistory history = mock(GenerationHistory.class);
|
||||||
|
IrisEngine engine = engineWith(history);
|
||||||
|
engine.syncGenerationHistory();
|
||||||
|
verify(history).sync();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void syncWithoutAnAttachedHistoryDoesNothing() {
|
||||||
|
IrisEngine engine = mock(IrisEngine.class);
|
||||||
|
when(engine.getGenerationHistoryRuntimeRouter()).thenReturn(Optional.empty());
|
||||||
|
doCallRealMethod().when(engine).syncGenerationHistory();
|
||||||
|
engine.syncGenerationHistory();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void failedSyncSurfacesTheStorageFailure() throws Exception {
|
||||||
|
GenerationHistory history = mock(GenerationHistory.class);
|
||||||
|
IOException failure = new IOException("history force failed");
|
||||||
|
doThrow(failure).when(history).sync();
|
||||||
|
IrisEngine engine = engineWith(history);
|
||||||
|
when(engine.getWorld()).thenReturn(IrisWorld.builder().name("world").build());
|
||||||
|
assertSame(failure, assertThrows(IllegalStateException.class, engine::syncGenerationHistory).getCause());
|
||||||
|
}
|
||||||
|
|
||||||
|
private static IrisEngine engineWith(GenerationHistory history) {
|
||||||
|
GenerationHistoryRuntimeRouter router = mock(GenerationHistoryRuntimeRouter.class);
|
||||||
|
when(router.history()).thenReturn(history);
|
||||||
|
IrisEngine engine = mock(IrisEngine.class);
|
||||||
|
when(engine.getGenerationHistoryRuntimeRouter()).thenReturn(Optional.of(router));
|
||||||
|
doCallRealMethod().when(engine).syncGenerationHistory();
|
||||||
|
return engine;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -68,6 +68,7 @@ public class IrisEngineWorldSaveTest {
|
|||||||
fixture.release().countDown();
|
fixture.release().countDown();
|
||||||
close.get(5, TimeUnit.SECONDS).requireComplete("close");
|
close.get(5, TimeUnit.SECONDS).requireComplete("close");
|
||||||
assertEquals(0, fixture.sessions().activeLeases());
|
assertEquals(0, fixture.sessions().activeLeases());
|
||||||
|
verify(fixture.engine()).syncGenerationHistory();
|
||||||
} finally {
|
} finally {
|
||||||
fixture.release().countDown();
|
fixture.release().countDown();
|
||||||
}
|
}
|
||||||
|
|||||||
+12
-9
@@ -193,7 +193,7 @@ public class GenerationHistoryAdmissionTest {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void queuedClaimsShareOneForceAndRetainIndividualValidation() throws Exception {
|
public void queuedClaimsShareOneAppendPerRegionWithoutForcingAndRetainIndividualValidation() throws Exception {
|
||||||
GenerationHistory history = history();
|
GenerationHistory history = history();
|
||||||
GenerationAdmission.RuntimeLease runtime = history.retainRuntime();
|
GenerationAdmission.RuntimeLease runtime = history.retainRuntime();
|
||||||
List<GenerationHistory.GenerationStage> stages = new ArrayList<>();
|
List<GenerationHistory.GenerationStage> stages = new ArrayList<>();
|
||||||
@@ -207,6 +207,7 @@ public class GenerationHistoryAdmissionTest {
|
|||||||
workers.add(thread);
|
workers.add(thread);
|
||||||
return thread;
|
return thread;
|
||||||
});
|
});
|
||||||
|
AtomicInteger appends = new AtomicInteger();
|
||||||
AtomicInteger forces = new AtomicInteger();
|
AtomicInteger forces = new AtomicInteger();
|
||||||
List<Future<Boolean>> results = new ArrayList<>();
|
List<Future<Boolean>> results = new ArrayList<>();
|
||||||
try {
|
try {
|
||||||
@@ -217,7 +218,7 @@ public class GenerationHistoryAdmissionTest {
|
|||||||
? ChunkGenerationSemantics.builder(1, 0, stage.activation().activationId())
|
? ChunkGenerationSemantics.builder(1, 0, stage.activation().activationId())
|
||||||
.addSurfaceBiome("iris:conflict").seal().build()
|
.addSurfaceBiome("iris:conflict").seal().build()
|
||||||
: claim(stage);
|
: claim(stage);
|
||||||
results.add(executor.submit(() -> persistCountingForces(history, stage, update, forces)));
|
results.add(executor.submit(() -> persistCounting(history, stage, update, appends, forces)));
|
||||||
awaitBlockedOnHistory(history, workers, index + 1);
|
awaitBlockedOnHistory(history, workers, index + 1);
|
||||||
}
|
}
|
||||||
stages.get(5).close();
|
stages.get(5).close();
|
||||||
@@ -232,7 +233,8 @@ public class GenerationHistoryAdmissionTest {
|
|||||||
.getCause() instanceof IllegalStateException);
|
.getCause() instanceof IllegalStateException);
|
||||||
assertTrue(results.get(6).get(5, TimeUnit.SECONDS));
|
assertTrue(results.get(6).get(5, TimeUnit.SECONDS));
|
||||||
assertTrue(results.get(7).get(5, TimeUnit.SECONDS));
|
assertTrue(results.get(7).get(5, TimeUnit.SECONDS));
|
||||||
assertEquals(2, forces.get());
|
assertEquals(2, appends.get());
|
||||||
|
assertEquals(0, forces.get());
|
||||||
GenerationSemanticIndex loaded = GenerationSemanticIndex.loadRequired(history.paths().dimensionRoot());
|
GenerationSemanticIndex loaded = GenerationSemanticIndex.loadRequired(history.paths().dimensionRoot());
|
||||||
assertEquals(5, loaded.recordCount());
|
assertEquals(5, loaded.recordCount());
|
||||||
assertEquals(claim(stages.getFirst()), loaded.get(1, 0).orElseThrow());
|
assertEquals(claim(stages.getFirst()), loaded.get(1, 0).orElseThrow());
|
||||||
@@ -348,8 +350,9 @@ public class GenerationHistoryAdmissionTest {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private static boolean persistCountingForces(GenerationHistory history, GenerationHistory.GenerationStage stage,
|
private static boolean persistCounting(GenerationHistory history, GenerationHistory.GenerationStage stage,
|
||||||
ChunkGenerationSemantics claim, AtomicInteger forces) throws Exception {
|
ChunkGenerationSemantics claim, AtomicInteger appends,
|
||||||
|
AtomicInteger forces) throws Exception {
|
||||||
try (MockedStatic<FileChannel> ignored = mockStatic(FileChannel.class, invocation -> {
|
try (MockedStatic<FileChannel> ignored = mockStatic(FileChannel.class, invocation -> {
|
||||||
FileChannel source = (FileChannel) invocation.callRealMethod();
|
FileChannel source = (FileChannel) invocation.callRealMethod();
|
||||||
Path path = invocation.getArgument(0);
|
Path path = invocation.getArgument(0);
|
||||||
@@ -357,6 +360,7 @@ public class GenerationHistoryAdmissionTest {
|
|||||||
|| !path.getFileName().toString().endsWith(".iswal")) {
|
|| !path.getFileName().toString().endsWith(".iswal")) {
|
||||||
return source;
|
return source;
|
||||||
}
|
}
|
||||||
|
appends.incrementAndGet();
|
||||||
FileChannel intercepted = mock(FileChannel.class, delegatesTo(source));
|
FileChannel intercepted = mock(FileChannel.class, delegatesTo(source));
|
||||||
doAnswer(force -> {
|
doAnswer(force -> {
|
||||||
forces.incrementAndGet();
|
forces.incrementAndGet();
|
||||||
@@ -396,14 +400,13 @@ public class GenerationHistoryAdmissionTest {
|
|||||||
return channel;
|
return channel;
|
||||||
}
|
}
|
||||||
FileChannel intercepted = mock(FileChannel.class, delegatesTo(channel));
|
FileChannel intercepted = mock(FileChannel.class, delegatesTo(channel));
|
||||||
doAnswer(force -> {
|
doAnswer(write -> {
|
||||||
entered.countDown();
|
entered.countDown();
|
||||||
if (!release.await(5, TimeUnit.SECONDS)) {
|
if (!release.await(5, TimeUnit.SECONDS)) {
|
||||||
throw new IOException("Timed out waiting to resume journal persistence");
|
throw new IOException("Timed out waiting to resume journal persistence");
|
||||||
}
|
}
|
||||||
channel.force(true);
|
return channel.write((ByteBuffer) write.getArgument(0));
|
||||||
return null;
|
}).when(intercepted).write(any(ByteBuffer.class));
|
||||||
}).when(intercepted).force(true);
|
|
||||||
return intercepted;
|
return intercepted;
|
||||||
})) {
|
})) {
|
||||||
return history.claimGeneratedSemantics(stage, claim);
|
return history.claimGeneratedSemantics(stage, claim);
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import org.junit.Test;
|
|||||||
import java.nio.file.Path;
|
import java.nio.file.Path;
|
||||||
|
|
||||||
import static org.junit.Assert.assertEquals;
|
import static org.junit.Assert.assertEquals;
|
||||||
|
import static org.junit.Assert.assertSame;
|
||||||
import static org.junit.Assert.assertThrows;
|
import static org.junit.Assert.assertThrows;
|
||||||
|
|
||||||
public final class GenerationHistoryPathsTest {
|
public final class GenerationHistoryPathsTest {
|
||||||
@@ -34,4 +35,18 @@ public final class GenerationHistoryPathsTest {
|
|||||||
assertThrows(IllegalArgumentException.class, () -> paths.epochRoot("../escape"));
|
assertThrows(IllegalArgumentException.class, () -> paths.epochRoot("../escape"));
|
||||||
assertThrows(IllegalArgumentException.class, () -> paths.activationRoot(0L));
|
assertThrows(IllegalArgumentException.class, () -> paths.activationRoot(0L));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void repeatedPackRootsStayValidatedAndFollowTheRequestedEpoch() {
|
||||||
|
GenerationHistoryPaths paths = GenerationHistoryPaths.forDimension(Path.of("build", "world"));
|
||||||
|
String first = "a".repeat(64);
|
||||||
|
String second = "b".repeat(64);
|
||||||
|
|
||||||
|
Path firstRoot = paths.packRoot(first);
|
||||||
|
assertSame(firstRoot, paths.packRoot(first));
|
||||||
|
assertEquals(paths.epochRoot(second).resolve("pack"), paths.packRoot(second));
|
||||||
|
assertEquals(firstRoot, paths.packRoot(first));
|
||||||
|
assertThrows(IllegalArgumentException.class, () -> paths.packRoot("../escape"));
|
||||||
|
assertThrows(NullPointerException.class, () -> paths.packRoot(null));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
+78
@@ -1,17 +1,24 @@
|
|||||||
package art.arcane.iris.world.history;
|
package art.arcane.iris.world.history;
|
||||||
|
|
||||||
import art.arcane.iris.generation.runtime.IrisEngine;
|
import art.arcane.iris.generation.runtime.IrisEngine;
|
||||||
|
import art.arcane.iris.world.storage.Durability;
|
||||||
import org.junit.Test;
|
import org.junit.Test;
|
||||||
|
import org.mockito.MockedStatic;
|
||||||
|
|
||||||
|
import java.nio.channels.FileChannel;
|
||||||
import java.nio.file.Files;
|
import java.nio.file.Files;
|
||||||
import java.nio.file.Path;
|
import java.nio.file.Path;
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.Map;
|
||||||
import java.util.Set;
|
import java.util.Set;
|
||||||
|
import java.util.concurrent.ConcurrentHashMap;
|
||||||
import java.util.concurrent.CompletableFuture;
|
import java.util.concurrent.CompletableFuture;
|
||||||
import java.util.concurrent.CountDownLatch;
|
import java.util.concurrent.CountDownLatch;
|
||||||
import java.util.concurrent.ExecutorService;
|
import java.util.concurrent.ExecutorService;
|
||||||
import java.util.concurrent.Executors;
|
import java.util.concurrent.Executors;
|
||||||
import java.util.concurrent.TimeUnit;
|
import java.util.concurrent.TimeUnit;
|
||||||
import java.util.concurrent.TimeoutException;
|
import java.util.concurrent.TimeoutException;
|
||||||
|
import java.util.concurrent.atomic.AtomicInteger;
|
||||||
import java.util.concurrent.atomic.AtomicReference;
|
import java.util.concurrent.atomic.AtomicReference;
|
||||||
|
|
||||||
import static org.junit.Assert.assertEquals;
|
import static org.junit.Assert.assertEquals;
|
||||||
@@ -21,8 +28,12 @@ import static org.junit.Assert.assertNotSame;
|
|||||||
import static org.junit.Assert.assertSame;
|
import static org.junit.Assert.assertSame;
|
||||||
import static org.junit.Assert.assertThrows;
|
import static org.junit.Assert.assertThrows;
|
||||||
import static org.junit.Assert.assertTrue;
|
import static org.junit.Assert.assertTrue;
|
||||||
|
import static org.mockito.AdditionalAnswers.delegatesTo;
|
||||||
import static org.mockito.ArgumentMatchers.any;
|
import static org.mockito.ArgumentMatchers.any;
|
||||||
|
import static org.mockito.Mockito.CALLS_REAL_METHODS;
|
||||||
|
import static org.mockito.Mockito.doAnswer;
|
||||||
import static org.mockito.Mockito.mock;
|
import static org.mockito.Mockito.mock;
|
||||||
|
import static org.mockito.Mockito.mockStatic;
|
||||||
import static org.mockito.Mockito.never;
|
import static org.mockito.Mockito.never;
|
||||||
import static org.mockito.Mockito.verify;
|
import static org.mockito.Mockito.verify;
|
||||||
import static org.mockito.Mockito.when;
|
import static org.mockito.Mockito.when;
|
||||||
@@ -67,6 +78,73 @@ public final class GenerationHistoryRuntimeRouterLifecycleTest extends Generatio
|
|||||||
router.close();
|
router.close();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void closeForcesDeferredSemanticAndBiomeWrites() throws Exception {
|
||||||
|
Path world = temporaryFolder.newFolder("router-close-sync-world").toPath();
|
||||||
|
GenerationHistory history = createHistory(world, createPack("router-close-sync-pack", "alpha"));
|
||||||
|
IrisEngine engine = mock(IrisEngine.class);
|
||||||
|
FakeRuntimeFactory runtimes = new FakeRuntimeFactory();
|
||||||
|
IrisEngine.GenerationRuntimeBinding active = runtimes.binding(history, history.activeActivation());
|
||||||
|
when(engine.getActiveGenerationRuntimeBinding()).thenReturn(active);
|
||||||
|
GenerationHistoryRuntimeRouter router = GenerationHistoryRuntimeRouter.attach(
|
||||||
|
engine, history, (ignored, x, z) -> signature(x, z), runtimes);
|
||||||
|
Path journal = history.paths().generationRoot().resolve("semantics/r.0.0.iswal");
|
||||||
|
Path biomes = history.paths().generationRoot().resolve("biomes/r.0.0.ibio");
|
||||||
|
Map<Path, AtomicInteger> forces = new ConcurrentHashMap<>();
|
||||||
|
try (MockedStatic<Durability> durability = mockStatic(Durability.class, CALLS_REAL_METHODS);
|
||||||
|
MockedStatic<FileChannel> ignored = countingForces(Set.of(journal, biomes), forces)) {
|
||||||
|
durability.when(Durability::enabled).thenReturn(true);
|
||||||
|
try (GenerationAdmission.RuntimeLease runtime = history.retainRuntime();
|
||||||
|
GenerationHistory.GenerationStage stage = history.openStage(1, 0)) {
|
||||||
|
long activation = stage.activation().activationId();
|
||||||
|
assertTrue(history.claimGeneratedSemantics(stage, ChunkGenerationSemantics.builder(1, 0, activation)
|
||||||
|
.addSurfaceBiome("iris:forest").seal().build()));
|
||||||
|
assertTrue(history.savedBiomes().claimAndPersist(savedBiomes(1, activation)));
|
||||||
|
}
|
||||||
|
assertEquals(0, forced(forces, journal));
|
||||||
|
assertEquals(0, forced(forces, biomes));
|
||||||
|
router.close();
|
||||||
|
assertEquals(1, forced(forces, journal));
|
||||||
|
assertEquals(1, forced(forces, biomes));
|
||||||
|
}
|
||||||
|
verify(engine).detachGenerationHistoryRuntimeRouter(router);
|
||||||
|
}
|
||||||
|
|
||||||
|
private static int forced(Map<Path, AtomicInteger> forces, Path path) {
|
||||||
|
AtomicInteger count = forces.get(path);
|
||||||
|
return count == null ? 0 : count.get();
|
||||||
|
}
|
||||||
|
|
||||||
|
private static MockedStatic<FileChannel> countingForces(Set<Path> watched, Map<Path, AtomicInteger> forces) {
|
||||||
|
return mockStatic(FileChannel.class, invocation -> {
|
||||||
|
FileChannel source = (FileChannel) invocation.callRealMethod();
|
||||||
|
if (invocation.getMethod().getParameterCount() != 2 || !watched.contains(invocation.getArgument(0))) {
|
||||||
|
return source;
|
||||||
|
}
|
||||||
|
Path path = invocation.getArgument(0);
|
||||||
|
FileChannel intercepted = mock(FileChannel.class, delegatesTo(source));
|
||||||
|
doAnswer(force -> {
|
||||||
|
forces.computeIfAbsent(path, unused -> new AtomicInteger()).incrementAndGet();
|
||||||
|
source.force(true);
|
||||||
|
return null;
|
||||||
|
}).when(intercepted).force(true);
|
||||||
|
return intercepted;
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
private static SavedBiomeChunk savedBiomes(int chunkX, long activation) {
|
||||||
|
SavedBiomeChunk.Cell cell = new SavedBiomeChunk.Cell(activation, "biome/main", "region/main");
|
||||||
|
SavedBiomeChunk.Column column = new SavedBiomeChunk.Column(cell, cell,
|
||||||
|
List.of(new SavedBiomeChunk.Span(-64, 320, cell)));
|
||||||
|
SavedBiomeChunk.Builder builder = SavedBiomeChunk.builder(new SavedBiomeChunk.Header(chunkX, 0, activation, -64, 384));
|
||||||
|
for (int z = 0; z < 16; z++) {
|
||||||
|
for (int x = 0; x < 16; x++) {
|
||||||
|
builder.column(x, z, column);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return builder.build();
|
||||||
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void closeRejectsNewRoutesAndAwaitsExistingLeases() throws Exception {
|
public void closeRejectsNewRoutesAndAwaitsExistingLeases() throws Exception {
|
||||||
Path world = temporaryFolder.newFolder("router-close-await-world").toPath();
|
Path world = temporaryFolder.newFolder("router-close-await-world").toPath();
|
||||||
|
|||||||
+5
-6
@@ -95,19 +95,18 @@ public class GenerationSemanticBatchWriteTest {
|
|||||||
GenerationSemanticIndex index = GenerationSemanticIndex.initialize(root);
|
GenerationSemanticIndex index = GenerationSemanticIndex.initialize(root);
|
||||||
List<GenerationSemanticIndex.Claim> claims = List.of(
|
List<GenerationSemanticIndex.Claim> claims = List.of(
|
||||||
pending(0, "first"), pending(32, "failed"), pending(64, "last"));
|
pending(0, "first"), pending(32, "failed"), pending(64, "last"));
|
||||||
IOException failure = new IOException("Middle region force failed");
|
IOException failure = new IOException("Middle region write failed");
|
||||||
AtomicInteger forces = new AtomicInteger();
|
AtomicInteger writes = new AtomicInteger();
|
||||||
try (MockedStatic<FileChannel> ignored = channels((path, source, intercepted) -> {
|
try (MockedStatic<FileChannel> ignored = channels((path, source, intercepted) -> {
|
||||||
if (!path.getFileName().toString().equals("r.1.0.iswal")) {
|
if (!path.getFileName().toString().equals("r.1.0.iswal")) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
doAnswer(invocation -> {
|
doAnswer(invocation -> {
|
||||||
if (forces.incrementAndGet() == 1) {
|
if (writes.incrementAndGet() == 1) {
|
||||||
throw failure;
|
throw failure;
|
||||||
}
|
}
|
||||||
source.force(true);
|
return source.write((ByteBuffer) invocation.getArgument(0));
|
||||||
return null;
|
}).when(intercepted).write(any(ByteBuffer.class));
|
||||||
}).when(intercepted).force(true);
|
|
||||||
})) {
|
})) {
|
||||||
index.claimAndPersistBatch(claims);
|
index.claimAndPersistBatch(claims);
|
||||||
}
|
}
|
||||||
|
|||||||
+46
-29
@@ -92,44 +92,64 @@ public class GenerationSemanticJournalWriteTest {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void failedForceRestoresBytesAndLeavesTheClaimRetryable() throws Exception {
|
public void claimReturnsBeforeTheJournalIsForcedAndSyncForcesIt() throws Exception {
|
||||||
Path root = temporaryFolder.newFolder().toPath();
|
Path root = temporaryFolder.newFolder().toPath();
|
||||||
GenerationSemanticIndex index = GenerationSemanticIndex.initialize(root);
|
GenerationSemanticIndex index = GenerationSemanticIndex.initialize(root);
|
||||||
index.claimAndPersist(claim(0, "first"));
|
index.claimAndPersist(claim(0, "first"));
|
||||||
Path journal = index.storageDirectory().resolve("r.0.0.iswal");
|
index.sync();
|
||||||
byte[] original = Files.readAllBytes(journal);
|
|
||||||
IOException failure = new IOException("Journal force failed");
|
|
||||||
AtomicInteger forces = new AtomicInteger();
|
AtomicInteger forces = new AtomicInteger();
|
||||||
try (MockedStatic<FileChannel> ignored = journalChannels((source, intercepted) ->
|
try (MockedStatic<FileChannel> ignored = journalChannels((source, intercepted) ->
|
||||||
doAnswer(invocation -> {
|
doAnswer(invocation -> {
|
||||||
if (forces.incrementAndGet() == 1) {
|
forces.incrementAndGet();
|
||||||
throw failure;
|
|
||||||
}
|
|
||||||
source.force(true);
|
source.force(true);
|
||||||
return null;
|
return null;
|
||||||
}).when(intercepted).force(true))) {
|
}).when(intercepted).force(true))) {
|
||||||
assertSame(failure, assertThrows(IOException.class, () -> index.claimAndPersist(claim(1, "second"))));
|
assertTrue(index.claimAndPersist(claim(1, "second")));
|
||||||
|
assertEquals(0, forces.get());
|
||||||
|
assertEquals(Optional.of(claim(1, "second")), index.get(1, 0));
|
||||||
|
index.sync();
|
||||||
|
assertEquals(1, forces.get());
|
||||||
|
index.sync();
|
||||||
|
assertEquals(1, forces.get());
|
||||||
}
|
}
|
||||||
assertEquals(2, forces.get());
|
|
||||||
assertArrayEquals(original, Files.readAllBytes(journal));
|
|
||||||
assertTrue(index.get(1, 0).isEmpty());
|
|
||||||
assertTrue(index.claimAndPersist(claim(1, "second")));
|
|
||||||
assertEquals(Optional.of(claim(1, "second")), GenerationSemanticIndex.loadRequired(root).get(1, 0));
|
assertEquals(Optional.of(claim(1, "second")), GenerationSemanticIndex.loadRequired(root).get(1, 0));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void failedSyncRejectsFurtherMutationButKeepsWrittenClaims() throws Exception {
|
||||||
|
Path root = temporaryFolder.newFolder().toPath();
|
||||||
|
GenerationSemanticIndex index = GenerationSemanticIndex.initialize(root);
|
||||||
|
index.claimAndPersist(claim(0, "first"));
|
||||||
|
IOException failure = new IOException("Deferred journal force failed");
|
||||||
|
try (MockedStatic<FileChannel> ignored = journalChannels((source, intercepted) ->
|
||||||
|
doAnswer(invocation -> {
|
||||||
|
throw failure;
|
||||||
|
}).when(intercepted).force(true))) {
|
||||||
|
assertSame(failure, assertThrows(IOException.class, index::sync));
|
||||||
|
}
|
||||||
|
assertSame(failure, assertThrows(IOException.class, () -> index.claimAndPersist(claim(1, "second"))).getCause());
|
||||||
|
assertThrows(IllegalStateException.class, () -> index.get(0, 0));
|
||||||
|
GenerationSemanticIndex replayed = GenerationSemanticIndex.loadRequired(root);
|
||||||
|
assertEquals(Optional.of(claim(0, "first")), replayed.get(0, 0));
|
||||||
|
assertTrue(replayed.get(1, 0).isEmpty());
|
||||||
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void rollbackFailurePreservesCausesAndRejectsFurtherMutation() throws Exception {
|
public void rollbackFailurePreservesCausesAndRejectsFurtherMutation() throws Exception {
|
||||||
Path root = temporaryFolder.newFolder().toPath();
|
Path root = temporaryFolder.newFolder().toPath();
|
||||||
GenerationSemanticIndex index = GenerationSemanticIndex.initialize(root);
|
GenerationSemanticIndex index = GenerationSemanticIndex.initialize(root);
|
||||||
index.claimAndPersist(claim(0, "first"));
|
index.claimAndPersist(claim(0, "first"));
|
||||||
IOException failure = new IOException("Journal force failed");
|
IOException failure = new IOException("Journal write failed");
|
||||||
IOException rollback = new IOException("Rollback force failed");
|
IOException rollback = new IOException("Rollback force failed");
|
||||||
AtomicInteger forces = new AtomicInteger();
|
|
||||||
IOException reported;
|
IOException reported;
|
||||||
try (MockedStatic<FileChannel> ignored = journalChannels((source, intercepted) ->
|
try (MockedStatic<FileChannel> ignored = journalChannels((source, intercepted) -> {
|
||||||
doAnswer(invocation -> {
|
doAnswer(invocation -> {
|
||||||
throw forces.incrementAndGet() == 1 ? failure : rollback;
|
throw failure;
|
||||||
}).when(intercepted).force(true))) {
|
}).when(intercepted).write(any(ByteBuffer.class));
|
||||||
|
doAnswer(invocation -> {
|
||||||
|
throw rollback;
|
||||||
|
}).when(intercepted).force(true);
|
||||||
|
})) {
|
||||||
reported = assertThrows(IOException.class, () -> index.claimAndPersist(claim(1, "failed")));
|
reported = assertThrows(IOException.class, () -> index.claimAndPersist(claim(1, "failed")));
|
||||||
}
|
}
|
||||||
assertSame(failure, reported.getCause());
|
assertSame(failure, reported.getCause());
|
||||||
@@ -176,7 +196,7 @@ public class GenerationSemanticJournalWriteTest {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void newJournalDirectoryFailureRollsBackBeforeRetry() throws Exception {
|
public void newJournalDirectoryIsForcedBySyncAndItsFailureRejectsFurtherMutation() throws Exception {
|
||||||
assumeTrue(File.separatorChar != '\\');
|
assumeTrue(File.separatorChar != '\\');
|
||||||
Path root = temporaryFolder.newFolder().toPath();
|
Path root = temporaryFolder.newFolder().toPath();
|
||||||
GenerationSemanticIndex index = GenerationSemanticIndex.initialize(root);
|
GenerationSemanticIndex index = GenerationSemanticIndex.initialize(root);
|
||||||
@@ -190,20 +210,17 @@ public class GenerationSemanticJournalWriteTest {
|
|||||||
}
|
}
|
||||||
FileChannel intercepted = mock(FileChannel.class, delegatesTo(source));
|
FileChannel intercepted = mock(FileChannel.class, delegatesTo(source));
|
||||||
doAnswer(force -> {
|
doAnswer(force -> {
|
||||||
if (directoryForces.incrementAndGet() == 1) {
|
directoryForces.incrementAndGet();
|
||||||
throw failure;
|
throw failure;
|
||||||
}
|
|
||||||
source.force(true);
|
|
||||||
return null;
|
|
||||||
}).when(intercepted).force(true);
|
}).when(intercepted).force(true);
|
||||||
return intercepted;
|
return intercepted;
|
||||||
})) {
|
})) {
|
||||||
assertSame(failure, assertThrows(IOException.class, () -> index.claimAndPersist(claim(0, "first"))));
|
assertTrue(index.claimAndPersist(claim(0, "first")));
|
||||||
|
assertEquals(0, directoryForces.get());
|
||||||
|
assertSame(failure, assertThrows(IOException.class, index::sync));
|
||||||
|
assertEquals(1, directoryForces.get());
|
||||||
}
|
}
|
||||||
assertEquals(2, directoryForces.get());
|
assertThrows(IOException.class, () -> index.claimAndPersist(claim(1, "second")));
|
||||||
assertEquals(0L, Files.size(index.storageDirectory().resolve("r.0.0.iswal")));
|
|
||||||
assertTrue(index.get(0, 0).isEmpty());
|
|
||||||
assertTrue(index.claimAndPersist(claim(0, "first")));
|
|
||||||
assertEquals(Optional.of(claim(0, "first")), GenerationSemanticIndex.loadRequired(root).get(0, 0));
|
assertEquals(Optional.of(claim(0, "first")), GenerationSemanticIndex.loadRequired(root).get(0, 0));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+15
-13
@@ -6,6 +6,7 @@ import org.junit.rules.TemporaryFolder;
|
|||||||
import org.mockito.MockedStatic;
|
import org.mockito.MockedStatic;
|
||||||
|
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
|
import java.nio.ByteBuffer;
|
||||||
import java.nio.channels.FileChannel;
|
import java.nio.channels.FileChannel;
|
||||||
import java.nio.file.Files;
|
import java.nio.file.Files;
|
||||||
import java.nio.file.Path;
|
import java.nio.file.Path;
|
||||||
@@ -29,6 +30,7 @@ import static org.junit.Assert.assertSame;
|
|||||||
import static org.junit.Assert.assertThrows;
|
import static org.junit.Assert.assertThrows;
|
||||||
import static org.junit.Assert.assertTrue;
|
import static org.junit.Assert.assertTrue;
|
||||||
import static org.mockito.AdditionalAnswers.delegatesTo;
|
import static org.mockito.AdditionalAnswers.delegatesTo;
|
||||||
|
import static org.mockito.ArgumentMatchers.any;
|
||||||
import static org.mockito.Mockito.doAnswer;
|
import static org.mockito.Mockito.doAnswer;
|
||||||
import static org.mockito.Mockito.mock;
|
import static org.mockito.Mockito.mock;
|
||||||
import static org.mockito.Mockito.mockStatic;
|
import static org.mockito.Mockito.mockStatic;
|
||||||
@@ -69,7 +71,7 @@ public class GenerationSemanticPointReadTest {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void newRegionRemainsAbsentUntilItsFirstForceCompletes() throws Exception {
|
public void newRegionRemainsAbsentUntilItsFirstAppendCompletes() throws Exception {
|
||||||
GenerationSemanticIndex index = index();
|
GenerationSemanticIndex index = index();
|
||||||
ExecutorService reader = Executors.newSingleThreadExecutor();
|
ExecutorService reader = Executors.newSingleThreadExecutor();
|
||||||
try (AppendGate gate = append(() -> index.claimAndPersist(claim(-32, 1)), Failure.NONE)) {
|
try (AppendGate gate = append(() -> index.claimAndPersist(claim(-32, 1)), Failure.NONE)) {
|
||||||
@@ -233,7 +235,7 @@ public class GenerationSemanticPointReadTest {
|
|||||||
private static AppendGate append(Callable<?> operation, Failure mode) throws InterruptedException {
|
private static AppendGate append(Callable<?> operation, Failure mode) throws InterruptedException {
|
||||||
AppendGate gate = new AppendGate(operation, mode);
|
AppendGate gate = new AppendGate(operation, mode);
|
||||||
gate.worker.start();
|
gate.worker.start();
|
||||||
assertTrue(gate.forcing.await(5, TimeUnit.SECONDS));
|
assertTrue(gate.appending.await(5, TimeUnit.SECONDS));
|
||||||
return gate;
|
return gate;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -250,7 +252,7 @@ public class GenerationSemanticPointReadTest {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private static final class AppendGate implements AutoCloseable {
|
private static final class AppendGate implements AutoCloseable {
|
||||||
private final CountDownLatch forcing = new CountDownLatch(1);
|
private final CountDownLatch appending = new CountDownLatch(1);
|
||||||
private final CountDownLatch release = new CountDownLatch(1);
|
private final CountDownLatch release = new CountDownLatch(1);
|
||||||
private final AtomicReference<Throwable> failure = new AtomicReference<>();
|
private final AtomicReference<Throwable> failure = new AtomicReference<>();
|
||||||
private final IOException originalFailure = new IOException("Semantic journal failed");
|
private final IOException originalFailure = new IOException("Semantic journal failed");
|
||||||
@@ -266,24 +268,24 @@ public class GenerationSemanticPointReadTest {
|
|||||||
return source;
|
return source;
|
||||||
}
|
}
|
||||||
FileChannel intercepted = mock(FileChannel.class, delegatesTo(source));
|
FileChannel intercepted = mock(FileChannel.class, delegatesTo(source));
|
||||||
AtomicInteger forces = new AtomicInteger();
|
AtomicInteger writes = new AtomicInteger();
|
||||||
doAnswer(call -> {
|
doAnswer(call -> {
|
||||||
int count = forces.incrementAndGet();
|
if (writes.incrementAndGet() == 1) {
|
||||||
if (count == 1) {
|
appending.countDown();
|
||||||
forcing.countDown();
|
|
||||||
if (!release.await(10, TimeUnit.SECONDS)) {
|
if (!release.await(10, TimeUnit.SECONDS)) {
|
||||||
throw new IOException("Timed out waiting to resume semantic force");
|
throw new IOException("Timed out waiting to resume semantic append");
|
||||||
}
|
}
|
||||||
if (mode == Failure.ROLLBACK_CONFIRMED || mode == Failure.ROLLBACK_UNCERTAIN) {
|
if (mode == Failure.ROLLBACK_CONFIRMED || mode == Failure.ROLLBACK_UNCERTAIN) {
|
||||||
throw originalFailure;
|
throw originalFailure;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if (count == 2 && mode == Failure.ROLLBACK_UNCERTAIN) {
|
return source.write((ByteBuffer) call.getArgument(0));
|
||||||
|
}).when(intercepted).write(any(ByteBuffer.class));
|
||||||
|
if (mode == Failure.ROLLBACK_UNCERTAIN) {
|
||||||
|
doAnswer(call -> {
|
||||||
throw new IOException("Semantic rollback failed");
|
throw new IOException("Semantic rollback failed");
|
||||||
}
|
}).when(intercepted).force(true);
|
||||||
source.force(true);
|
}
|
||||||
return null;
|
|
||||||
}).when(intercepted).force(true);
|
|
||||||
if (mode == Failure.CLOSE_UNCERTAIN) {
|
if (mode == Failure.CLOSE_UNCERTAIN) {
|
||||||
doAnswer(call -> {
|
doAnswer(call -> {
|
||||||
source.close();
|
source.close();
|
||||||
|
|||||||
+8
-2
@@ -46,6 +46,7 @@ public final class GenerationUnpublishedHistoryTest extends GenerationHistorySup
|
|||||||
GenerationHistory reopened = GenerationHistory.open(root);
|
GenerationHistory reopened = GenerationHistory.open(root);
|
||||||
assertEquals(biomes(1), reopened.savedBiomes().get(1, 0).orElseThrow());
|
assertEquals(biomes(1), reopened.savedBiomes().get(1, 0).orElseThrow());
|
||||||
assertBiomeAppendForces(reopened.savedBiomes(), root, 2, 1);
|
assertBiomeAppendForces(reopened.savedBiomes(), root, 2, 1);
|
||||||
|
assertEquals(biomes(2), GenerationHistory.open(root).savedBiomes().get(2, 0).orElseThrow());
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
@@ -61,7 +62,10 @@ public final class GenerationUnpublishedHistoryTest extends GenerationHistorySup
|
|||||||
GenerationSemanticIndex reopened = GenerationSemanticIndex.loadRequired(root);
|
GenerationSemanticIndex reopened = GenerationSemanticIndex.loadRequired(root);
|
||||||
assertEquals(semantics(1), reopened.get(1, 0).orElseThrow());
|
assertEquals(semantics(1), reopened.get(1, 0).orElseThrow());
|
||||||
try (MockedStatic<Durability> forces = mockStatic(Durability.class, CALLS_REAL_METHODS)) {
|
try (MockedStatic<Durability> forces = mockStatic(Durability.class, CALLS_REAL_METHODS)) {
|
||||||
|
forces.when(Durability::enabled).thenReturn(true);
|
||||||
assertTrue(reopened.claimAndPersist(semantics(2)));
|
assertTrue(reopened.claimAndPersist(semantics(2)));
|
||||||
|
forces.verify(() -> Durability.force(any(FileChannel.class)), times(0));
|
||||||
|
reopened.sync();
|
||||||
forces.verify(() -> Durability.force(any(FileChannel.class)), times(1));
|
forces.verify(() -> Durability.force(any(FileChannel.class)), times(1));
|
||||||
}
|
}
|
||||||
reopened.compactJournals();
|
reopened.compactJournals();
|
||||||
@@ -91,7 +95,7 @@ public final class GenerationUnpublishedHistoryTest extends GenerationHistorySup
|
|||||||
assertTrue(store.claimAndPersist(biomes(1)));
|
assertTrue(store.claimAndPersist(biomes(1)));
|
||||||
}
|
}
|
||||||
|
|
||||||
private static void assertBiomeAppendForces(SavedBiomeStore store, Path root, int x, int count) throws Exception {
|
private static void assertBiomeAppendForces(SavedBiomeStore store, Path root, int x, int syncForces) throws Exception {
|
||||||
try (MockedStatic<Durability> durability = mockStatic(Durability.class, CALLS_REAL_METHODS);
|
try (MockedStatic<Durability> durability = mockStatic(Durability.class, CALLS_REAL_METHODS);
|
||||||
RandomAccessFile actual = new RandomAccessFile(region(root).toFile(), "rw")) {
|
RandomAccessFile actual = new RandomAccessFile(region(root).toFile(), "rw")) {
|
||||||
durability.when(Durability::enabled).thenReturn(true);
|
durability.when(Durability::enabled).thenReturn(true);
|
||||||
@@ -101,7 +105,9 @@ public final class GenerationUnpublishedHistoryTest extends GenerationHistorySup
|
|||||||
(file, context) -> doReturn(channel).when(file).getChannel())) {
|
(file, context) -> doReturn(channel).when(file).getChannel())) {
|
||||||
assertTrue(store.claimAndPersist(biomes(x)));
|
assertTrue(store.claimAndPersist(biomes(x)));
|
||||||
}
|
}
|
||||||
verify(channel, times(count)).force(true);
|
verify(channel, times(0)).force(true);
|
||||||
|
store.sync();
|
||||||
|
durability.verify(() -> Durability.force(any(FileChannel.class)), times(syncForces));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -60,9 +60,7 @@ public class SavedBiomeStoreBatchTest {
|
|||||||
List<SavedBiomeChunk> claims = List.of(chunk(1, 3L), chunk(2, 3L), chunk(3, 3L));
|
List<SavedBiomeChunk> claims = List.of(chunk(1, 3L), chunk(2, 3L), chunk(3, 3L));
|
||||||
try {
|
try {
|
||||||
List<Future<ClaimResult>> results = enqueue(store, root, control, workers, claims);
|
List<Future<ClaimResult>> results = enqueue(store, root, control, workers, claims);
|
||||||
assertTrue(control.forcing.await(5, TimeUnit.SECONDS));
|
assertTrue(control.writing.await(5, TimeUnit.SECONDS));
|
||||||
assertEquals(3, control.maximumGathered.get());
|
|
||||||
assertTrue(control.writes.get() > claims.size());
|
|
||||||
for (SavedBiomeChunk claim : claims) {
|
for (SavedBiomeChunk claim : claims) {
|
||||||
assertTrue(store.cached(claim.chunkX(), claim.chunkZ()).isEmpty());
|
assertTrue(store.cached(claim.chunkX(), claim.chunkZ()).isEmpty());
|
||||||
}
|
}
|
||||||
@@ -72,7 +70,9 @@ public class SavedBiomeStoreBatchTest {
|
|||||||
assertTrue(completed.persisted());
|
assertTrue(completed.persisted());
|
||||||
assertNull(completed.failure());
|
assertNull(completed.failure());
|
||||||
}
|
}
|
||||||
assertEquals(1, control.forces.get());
|
assertEquals(3, control.maximumGathered.get());
|
||||||
|
assertTrue(control.writes.get() > claims.size());
|
||||||
|
assertEquals(0, control.forces.get());
|
||||||
SavedBiomeStore reopened = SavedBiomeStore.open(root);
|
SavedBiomeStore reopened = SavedBiomeStore.open(root);
|
||||||
for (SavedBiomeChunk claim : claims) {
|
for (SavedBiomeChunk claim : claims) {
|
||||||
assertEquals(claim, reopened.get(claim.chunkX(), claim.chunkZ()).orElseThrow());
|
assertEquals(claim, reopened.get(claim.chunkX(), claim.chunkZ()).orElseThrow());
|
||||||
@@ -94,6 +94,7 @@ public class SavedBiomeStoreBatchTest {
|
|||||||
BatchControl control = new BatchControl();
|
BatchControl control = new BatchControl();
|
||||||
control.maximumWriteBytes = 11;
|
control.maximumWriteBytes = 11;
|
||||||
control.writeFailure = new IOException("Partial gathered append failed");
|
control.writeFailure = new IOException("Partial gathered append failed");
|
||||||
|
control.failingWrite = 2;
|
||||||
control.release.countDown();
|
control.release.countDown();
|
||||||
ExecutorService workers = Executors.newFixedThreadPool(4);
|
ExecutorService workers = Executors.newFixedThreadPool(4);
|
||||||
List<SavedBiomeChunk> claims = List.of(chunk(1, 3L), chunk(2, 3L), chunk(3, 3L));
|
List<SavedBiomeChunk> claims = List.of(chunk(1, 3L), chunk(2, 3L), chunk(3, 3L));
|
||||||
@@ -167,7 +168,7 @@ public class SavedBiomeStoreBatchTest {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void queuedClaimsShareOneForceAndPublishAfterItCompletes() throws Exception {
|
public void queuedClaimsShareOneAppendAndPublishAfterItCompletes() throws Exception {
|
||||||
Path root = temporaryFolder.newFolder().toPath();
|
Path root = temporaryFolder.newFolder().toPath();
|
||||||
SavedBiomeStore store = SavedBiomeStore.open(root);
|
SavedBiomeStore store = SavedBiomeStore.open(root);
|
||||||
store.claimAndPersist(chunk(0, 3L));
|
store.claimAndPersist(chunk(0, 3L));
|
||||||
@@ -177,7 +178,7 @@ public class SavedBiomeStoreBatchTest {
|
|||||||
List<Future<ClaimResult>> results = enqueue(store, root, control, workers,
|
List<Future<ClaimResult>> results = enqueue(store, root, control, workers,
|
||||||
List.of(chunk(1, 3L), chunk(2, 3L), chunk(3, 3L), chunk(4, 3L),
|
List.of(chunk(1, 3L), chunk(2, 3L), chunk(3, 3L), chunk(4, 3L),
|
||||||
chunk(5, 3L), chunk(6, 3L), chunk(7, 3L), chunk(8, 3L)));
|
chunk(5, 3L), chunk(6, 3L), chunk(7, 3L), chunk(8, 3L)));
|
||||||
assertTrue(control.forcing.await(5, TimeUnit.SECONDS));
|
assertTrue(control.writing.await(5, TimeUnit.SECONDS));
|
||||||
for (int index = 1; index <= 8; index++) {
|
for (int index = 1; index <= 8; index++) {
|
||||||
assertTrue(store.cached(index, 0).isEmpty());
|
assertTrue(store.cached(index, 0).isEmpty());
|
||||||
assertFalse(results.get(index - 1).isDone());
|
assertFalse(results.get(index - 1).isDone());
|
||||||
@@ -188,7 +189,8 @@ public class SavedBiomeStoreBatchTest {
|
|||||||
assertNull(claim.failure);
|
assertNull(claim.failure);
|
||||||
assertTrue(claim.persisted);
|
assertTrue(claim.persisted);
|
||||||
}
|
}
|
||||||
assertEquals(1, control.forces.get());
|
assertEquals(1, control.writes.get());
|
||||||
|
assertEquals(0, control.forces.get());
|
||||||
SavedBiomeStore reopened = SavedBiomeStore.open(root);
|
SavedBiomeStore reopened = SavedBiomeStore.open(root);
|
||||||
for (int index = 1; index <= 8; index++) {
|
for (int index = 1; index <= 8; index++) {
|
||||||
assertEquals(chunk(index, 3L), reopened.get(index, 0).orElseThrow());
|
assertEquals(chunk(index, 3L), reopened.get(index, 0).orElseThrow());
|
||||||
@@ -201,19 +203,20 @@ public class SavedBiomeStoreBatchTest {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void batchForceFailureRollsBackEveryNewClaimButPreservesExistingDuplicates() throws Exception {
|
public void batchWriteFailureRollsBackEveryNewClaimButPreservesExistingDuplicates() throws Exception {
|
||||||
Path root = temporaryFolder.newFolder().toPath();
|
Path root = temporaryFolder.newFolder().toPath();
|
||||||
SavedBiomeStore store = SavedBiomeStore.open(root);
|
SavedBiomeStore store = SavedBiomeStore.open(root);
|
||||||
store.claimAndPersist(chunk(0, 3L));
|
store.claimAndPersist(chunk(0, 3L));
|
||||||
Path region = root.resolve("iris/generation/biomes/r.0.0.ibio");
|
Path region = root.resolve("iris/generation/biomes/r.0.0.ibio");
|
||||||
byte[] original = Files.readAllBytes(region);
|
byte[] original = Files.readAllBytes(region);
|
||||||
BatchControl control = new BatchControl();
|
BatchControl control = new BatchControl();
|
||||||
control.failure = new IOException("Batch force failed");
|
control.writeFailure = new IOException("Batch write failed");
|
||||||
|
control.failingWrite = 1;
|
||||||
ExecutorService workers = Executors.newFixedThreadPool(4);
|
ExecutorService workers = Executors.newFixedThreadPool(4);
|
||||||
try {
|
try {
|
||||||
List<Future<ClaimResult>> results = enqueue(store, root, control, workers,
|
List<Future<ClaimResult>> results = enqueue(store, root, control, workers,
|
||||||
List.of(chunk(0, 3L), chunk(1, 3L), chunk(1, 3L), chunk(2, 3L)));
|
List.of(chunk(0, 3L), chunk(1, 3L), chunk(1, 3L), chunk(2, 3L)));
|
||||||
assertTrue(control.forcing.await(5, TimeUnit.SECONDS));
|
assertTrue(control.writing.await(5, TimeUnit.SECONDS));
|
||||||
control.release.countDown();
|
control.release.countDown();
|
||||||
ClaimResult existing = results.getFirst().get(5, TimeUnit.SECONDS);
|
ClaimResult existing = results.getFirst().get(5, TimeUnit.SECONDS);
|
||||||
assertFalse(existing.persisted);
|
assertFalse(existing.persisted);
|
||||||
@@ -221,9 +224,9 @@ public class SavedBiomeStoreBatchTest {
|
|||||||
for (int index = 1; index < results.size(); index++) {
|
for (int index = 1; index < results.size(); index++) {
|
||||||
ClaimResult result = results.get(index).get(5, TimeUnit.SECONDS);
|
ClaimResult result = results.get(index).get(5, TimeUnit.SECONDS);
|
||||||
assertFalse(result.persisted);
|
assertFalse(result.persisted);
|
||||||
assertSame(control.failure, result.failure);
|
assertSame(control.writeFailure, result.failure);
|
||||||
}
|
}
|
||||||
assertEquals(2, control.forces.get());
|
assertEquals(1, control.forces.get());
|
||||||
assertArrayEquals(original, Files.readAllBytes(region));
|
assertArrayEquals(original, Files.readAllBytes(region));
|
||||||
assertTrue(store.cached(1, 0).isEmpty());
|
assertTrue(store.cached(1, 0).isEmpty());
|
||||||
assertTrue(store.claimAndPersist(chunk(1, 3L)));
|
assertTrue(store.claimAndPersist(chunk(1, 3L)));
|
||||||
@@ -245,7 +248,7 @@ public class SavedBiomeStoreBatchTest {
|
|||||||
try {
|
try {
|
||||||
List<Future<ClaimResult>> results = enqueue(store, root, control, workers,
|
List<Future<ClaimResult>> results = enqueue(store, root, control, workers,
|
||||||
List.of(chunk(1, 3L), chunk(1, 3L), chunk(1, 4L)));
|
List.of(chunk(1, 3L), chunk(1, 3L), chunk(1, 4L)));
|
||||||
assertTrue(control.forcing.await(5, TimeUnit.SECONDS));
|
assertTrue(control.writing.await(5, TimeUnit.SECONDS));
|
||||||
control.release.countDown();
|
control.release.countDown();
|
||||||
int winners = 0;
|
int winners = 0;
|
||||||
SavedBiomeChunk actual = null;
|
SavedBiomeChunk actual = null;
|
||||||
@@ -257,7 +260,8 @@ public class SavedBiomeStoreBatchTest {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
assertEquals(1, winners);
|
assertEquals(1, winners);
|
||||||
assertEquals(1, control.forces.get());
|
assertEquals(1, control.writes.get());
|
||||||
|
assertEquals(0, control.forces.get());
|
||||||
assertEquals(actual, SavedBiomeStore.open(root).get(1, 0).orElseThrow());
|
assertEquals(actual, SavedBiomeStore.open(root).get(1, 0).orElseThrow());
|
||||||
for (int index = 0; index < results.size(); index++) {
|
for (int index = 0; index < results.size(); index++) {
|
||||||
ClaimResult result = results.get(index).get(5, TimeUnit.SECONDS);
|
ClaimResult result = results.get(index).get(5, TimeUnit.SECONDS);
|
||||||
@@ -277,12 +281,13 @@ public class SavedBiomeStoreBatchTest {
|
|||||||
SavedBiomeStore store = SavedBiomeStore.open(root);
|
SavedBiomeStore store = SavedBiomeStore.open(root);
|
||||||
store.claimAndPersist(chunk(0, 3L));
|
store.claimAndPersist(chunk(0, 3L));
|
||||||
BatchControl control = new BatchControl();
|
BatchControl control = new BatchControl();
|
||||||
control.uncheckedFailure = new IllegalStateException("Batch force failed unexpectedly");
|
control.uncheckedFailure = new IllegalStateException("Batch write failed unexpectedly");
|
||||||
|
control.failingWrite = 1;
|
||||||
ExecutorService workers = Executors.newFixedThreadPool(4);
|
ExecutorService workers = Executors.newFixedThreadPool(4);
|
||||||
try {
|
try {
|
||||||
List<Future<ClaimResult>> results = enqueue(store, root, control, workers,
|
List<Future<ClaimResult>> results = enqueue(store, root, control, workers,
|
||||||
List.of(chunk(1, 3L), chunk(2, 3L), chunk(3, 3L), chunk(4, 3L)));
|
List.of(chunk(1, 3L), chunk(2, 3L), chunk(3, 3L), chunk(4, 3L)));
|
||||||
assertTrue(control.forcing.await(5, TimeUnit.SECONDS));
|
assertTrue(control.writing.await(5, TimeUnit.SECONDS));
|
||||||
Object writingRegion = regionLock(store, 0, 0);
|
Object writingRegion = regionLock(store, 0, 0);
|
||||||
for (int region = 1; region <= 132; region++) {
|
for (int region = 1; region <= 132; region++) {
|
||||||
if (regionLock(store, region << 5, 0) != writingRegion) {
|
if (regionLock(store, region << 5, 0) != writingRegion) {
|
||||||
@@ -295,7 +300,8 @@ public class SavedBiomeStoreBatchTest {
|
|||||||
assertFalse(result.persisted);
|
assertFalse(result.persisted);
|
||||||
assertSame(control.uncheckedFailure, rootCause(result.failure));
|
assertSame(control.uncheckedFailure, rootCause(result.failure));
|
||||||
}
|
}
|
||||||
assertEquals(1, control.forces.get());
|
assertEquals(1, control.writes.get());
|
||||||
|
assertEquals(0, control.forces.get());
|
||||||
IOException retry = assertThrows(IOException.class, () -> store.claimAndPersist(chunk(1, 3L)));
|
IOException retry = assertThrows(IOException.class, () -> store.claimAndPersist(chunk(1, 3L)));
|
||||||
assertSame(control.uncheckedFailure, rootCause(retry));
|
assertSame(control.uncheckedFailure, rootCause(retry));
|
||||||
IOException read = assertThrows(IOException.class, () -> store.get(1, 0));
|
IOException read = assertThrows(IOException.class, () -> store.get(1, 0));
|
||||||
@@ -348,45 +354,43 @@ public class SavedBiomeStoreBatchTest {
|
|||||||
try (RandomAccessFile actual = new RandomAccessFile(
|
try (RandomAccessFile actual = new RandomAccessFile(
|
||||||
root.resolve("iris/generation/biomes/r.0.0.ibio").toFile(), "rw")) {
|
root.resolve("iris/generation/biomes/r.0.0.ibio").toFile(), "rw")) {
|
||||||
FileChannel channel = mock(FileChannel.class, delegatesTo(actual.getChannel()));
|
FileChannel channel = mock(FileChannel.class, delegatesTo(actual.getChannel()));
|
||||||
if (control.maximumWriteBytes > 0) {
|
|
||||||
doAnswer(invocation -> {
|
|
||||||
ByteBuffer[] records = invocation.getArgument(0);
|
|
||||||
int offset = invocation.getArgument(1);
|
|
||||||
int count = invocation.getArgument(2);
|
|
||||||
control.maximumGathered.accumulateAndGet(count, Math::max);
|
|
||||||
int write = control.writes.incrementAndGet();
|
|
||||||
if (control.zeroFirstWrite && write == 1) {
|
|
||||||
return 0L;
|
|
||||||
}
|
|
||||||
if (control.writeFailure != null && write == 2) {
|
|
||||||
throw control.writeFailure;
|
|
||||||
}
|
|
||||||
long written = 0L;
|
|
||||||
for (int index = offset; index < offset + count && written < control.maximumWriteBytes; index++) {
|
|
||||||
ByteBuffer record = records[index];
|
|
||||||
int previousLimit = record.limit();
|
|
||||||
record.limit(record.position() + Math.min(record.remaining(), control.maximumWriteBytes - (int) written));
|
|
||||||
try {
|
|
||||||
written += actual.getChannel().write(record);
|
|
||||||
} finally {
|
|
||||||
record.limit(previousLimit);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return written;
|
|
||||||
}).when(channel).write(any(ByteBuffer[].class), anyInt(), anyInt());
|
|
||||||
}
|
|
||||||
doAnswer(invocation -> {
|
doAnswer(invocation -> {
|
||||||
int force = control.forces.incrementAndGet();
|
ByteBuffer[] records = invocation.getArgument(0);
|
||||||
if (force == 1) {
|
int offset = invocation.getArgument(1);
|
||||||
control.forcing.countDown();
|
int count = invocation.getArgument(2);
|
||||||
|
control.maximumGathered.accumulateAndGet(count, Math::max);
|
||||||
|
int write = control.writes.incrementAndGet();
|
||||||
|
if (write == 1) {
|
||||||
|
control.writing.countDown();
|
||||||
assertTrue(control.release.await(5, TimeUnit.SECONDS));
|
assertTrue(control.release.await(5, TimeUnit.SECONDS));
|
||||||
if (control.failure != null) {
|
}
|
||||||
throw control.failure;
|
if (write == control.failingWrite && control.uncheckedFailure != null) {
|
||||||
}
|
throw control.uncheckedFailure;
|
||||||
if (control.uncheckedFailure != null) {
|
}
|
||||||
throw control.uncheckedFailure;
|
if (write == control.failingWrite && control.writeFailure != null) {
|
||||||
|
throw control.writeFailure;
|
||||||
|
}
|
||||||
|
if (control.zeroFirstWrite && write == 1) {
|
||||||
|
return 0L;
|
||||||
|
}
|
||||||
|
if (control.maximumWriteBytes <= 0) {
|
||||||
|
return actual.getChannel().write(records, offset, count);
|
||||||
|
}
|
||||||
|
long written = 0L;
|
||||||
|
for (int index = offset; index < offset + count && written < control.maximumWriteBytes; index++) {
|
||||||
|
ByteBuffer record = records[index];
|
||||||
|
int previousLimit = record.limit();
|
||||||
|
record.limit(record.position() + Math.min(record.remaining(), control.maximumWriteBytes - (int) written));
|
||||||
|
try {
|
||||||
|
written += actual.getChannel().write(record);
|
||||||
|
} finally {
|
||||||
|
record.limit(previousLimit);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
return written;
|
||||||
|
}).when(channel).write(any(ByteBuffer[].class), anyInt(), anyInt());
|
||||||
|
doAnswer(invocation -> {
|
||||||
|
control.forces.incrementAndGet();
|
||||||
actual.getChannel().force(true);
|
actual.getChannel().force(true);
|
||||||
return null;
|
return null;
|
||||||
}).when(channel).force(true);
|
}).when(channel).force(true);
|
||||||
@@ -426,15 +430,15 @@ public class SavedBiomeStoreBatchTest {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private static final class BatchControl {
|
private static final class BatchControl {
|
||||||
private final CountDownLatch forcing = new CountDownLatch(1);
|
private final CountDownLatch writing = new CountDownLatch(1);
|
||||||
private final CountDownLatch release = new CountDownLatch(1);
|
private final CountDownLatch release = new CountDownLatch(1);
|
||||||
private final AtomicInteger forces = new AtomicInteger();
|
private final AtomicInteger forces = new AtomicInteger();
|
||||||
private int maximumWriteBytes;
|
private int maximumWriteBytes;
|
||||||
private boolean zeroFirstWrite;
|
private boolean zeroFirstWrite;
|
||||||
|
private int failingWrite;
|
||||||
private IOException writeFailure;
|
private IOException writeFailure;
|
||||||
private final AtomicInteger writes = new AtomicInteger();
|
private final AtomicInteger writes = new AtomicInteger();
|
||||||
private final AtomicInteger maximumGathered = new AtomicInteger();
|
private final AtomicInteger maximumGathered = new AtomicInteger();
|
||||||
private IOException failure;
|
|
||||||
private RuntimeException uncheckedFailure;
|
private RuntimeException uncheckedFailure;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import org.junit.Rule;
|
|||||||
import org.junit.Test;
|
import org.junit.Test;
|
||||||
import org.junit.rules.TemporaryFolder;
|
import org.junit.rules.TemporaryFolder;
|
||||||
import org.mockito.MockedConstruction;
|
import org.mockito.MockedConstruction;
|
||||||
|
import org.mockito.MockedStatic;
|
||||||
|
|
||||||
import java.io.ByteArrayOutputStream;
|
import java.io.ByteArrayOutputStream;
|
||||||
import java.io.DataOutputStream;
|
import java.io.DataOutputStream;
|
||||||
@@ -21,7 +22,7 @@ import java.util.concurrent.ExecutorService;
|
|||||||
import java.util.concurrent.Executors;
|
import java.util.concurrent.Executors;
|
||||||
import java.util.concurrent.Future;
|
import java.util.concurrent.Future;
|
||||||
import java.util.concurrent.TimeUnit;
|
import java.util.concurrent.TimeUnit;
|
||||||
import java.util.concurrent.atomic.AtomicBoolean;
|
import java.util.concurrent.atomic.AtomicInteger;
|
||||||
import java.util.zip.CRC32;
|
import java.util.zip.CRC32;
|
||||||
import java.util.zip.Deflater;
|
import java.util.zip.Deflater;
|
||||||
import java.util.zip.DeflaterOutputStream;
|
import java.util.zip.DeflaterOutputStream;
|
||||||
@@ -41,6 +42,7 @@ import static org.mockito.Mockito.doAnswer;
|
|||||||
import static org.mockito.Mockito.doReturn;
|
import static org.mockito.Mockito.doReturn;
|
||||||
import static org.mockito.Mockito.mock;
|
import static org.mockito.Mockito.mock;
|
||||||
import static org.mockito.Mockito.mockConstruction;
|
import static org.mockito.Mockito.mockConstruction;
|
||||||
|
import static org.mockito.Mockito.mockStatic;
|
||||||
import static org.mockito.Mockito.never;
|
import static org.mockito.Mockito.never;
|
||||||
import static org.mockito.Mockito.spy;
|
import static org.mockito.Mockito.spy;
|
||||||
import static org.mockito.Mockito.verify;
|
import static org.mockito.Mockito.verify;
|
||||||
@@ -112,40 +114,41 @@ public class SavedBiomeStoreWriteTest {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void appendWritesOneFrameAndPublishesOnlyAfterForce() throws Exception {
|
public void appendWritesOneFrameAndPublishesOnceTheWriteCompletes() throws Exception {
|
||||||
Path root = temporaryFolder.newFolder().toPath();
|
Path root = temporaryFolder.newFolder().toPath();
|
||||||
SavedBiomeStore store = SavedBiomeStore.open(root);
|
SavedBiomeStore store = SavedBiomeStore.open(root);
|
||||||
store.claimAndPersist(chunk(0, 3L));
|
store.claimAndPersist(chunk(0, 3L));
|
||||||
CountDownLatch forcing = new CountDownLatch(1);
|
CountDownLatch writing = new CountDownLatch(1);
|
||||||
CountDownLatch release = new CountDownLatch(1);
|
CountDownLatch release = new CountDownLatch(1);
|
||||||
ExecutorService workers = Executors.newSingleThreadExecutor();
|
ExecutorService workers = Executors.newSingleThreadExecutor();
|
||||||
try {
|
try {
|
||||||
Future<Boolean> writing = workers.submit(() -> {
|
Future<Boolean> claimed = workers.submit(() -> {
|
||||||
try (RandomAccessFile actual = new RandomAccessFile(regionPath(root).toFile(), "rw")) {
|
try (RandomAccessFile actual = new RandomAccessFile(regionPath(root).toFile(), "rw")) {
|
||||||
FileChannel channel = mock(FileChannel.class, delegatesTo(actual.getChannel()));
|
FileChannel channel = mock(FileChannel.class, delegatesTo(actual.getChannel()));
|
||||||
doAnswer(invocation -> {
|
doAnswer(invocation -> {
|
||||||
forcing.countDown();
|
writing.countDown();
|
||||||
await(release);
|
await(release);
|
||||||
actual.getChannel().force(true);
|
return actual.getChannel().write((ByteBuffer[]) invocation.getArgument(0),
|
||||||
return null;
|
(int) invocation.getArgument(1), (int) invocation.getArgument(2));
|
||||||
}).when(channel).force(true);
|
}).when(channel).write(any(ByteBuffer[].class), anyInt(), anyInt());
|
||||||
try (MockedConstruction<RandomAccessFile> files = mockConstruction(RandomAccessFile.class,
|
try (MockedConstruction<RandomAccessFile> files = mockConstruction(RandomAccessFile.class,
|
||||||
withSettings().defaultAnswer(delegatesTo(actual)),
|
withSettings().defaultAnswer(delegatesTo(actual)),
|
||||||
(file, context) -> doReturn(channel).when(file).getChannel())) {
|
(file, context) -> doReturn(channel).when(file).getChannel())) {
|
||||||
boolean result = store.claimAndPersist(chunk(1, 3L));
|
boolean result = store.claimAndPersist(chunk(1, 3L));
|
||||||
assertEquals(1, files.constructed().size());
|
assertEquals(1, files.constructed().size());
|
||||||
verify(channel).write(any(ByteBuffer[].class), anyInt(), anyInt());
|
verify(channel).write(any(ByteBuffer[].class), anyInt(), anyInt());
|
||||||
|
verify(channel, never()).force(true);
|
||||||
verify(files.constructed().getFirst(), never()).write(any(byte[].class));
|
verify(files.constructed().getFirst(), never()).write(any(byte[].class));
|
||||||
verify(files.constructed().getFirst(), never()).writeInt(anyInt());
|
verify(files.constructed().getFirst(), never()).writeInt(anyInt());
|
||||||
return result;
|
return result;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
assertTrue(forcing.await(5, TimeUnit.SECONDS));
|
assertTrue(writing.await(5, TimeUnit.SECONDS));
|
||||||
assertTrue(store.cached(1, 0).isEmpty());
|
assertTrue(store.cached(1, 0).isEmpty());
|
||||||
assertFalse(writing.isDone());
|
assertFalse(claimed.isDone());
|
||||||
release.countDown();
|
release.countDown();
|
||||||
assertTrue(writing.get(5, TimeUnit.SECONDS));
|
assertTrue(claimed.get(5, TimeUnit.SECONDS));
|
||||||
assertEquals(chunk(1, 3L), store.cached(1, 0).orElseThrow());
|
assertEquals(chunk(1, 3L), store.cached(1, 0).orElseThrow());
|
||||||
assertEquals(chunk(1, 3L), SavedBiomeStore.open(root).get(1, 0).orElseThrow());
|
assertEquals(chunk(1, 3L), SavedBiomeStore.open(root).get(1, 0).orElseThrow());
|
||||||
} finally {
|
} finally {
|
||||||
@@ -155,21 +158,63 @@ public class SavedBiomeStoreWriteTest {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void failedForceRollsBackWithoutPublishingAndRetainsRollbackFailure() throws Exception {
|
public void syncForcesAppendedRegionsOnce() throws Exception {
|
||||||
|
Path root = temporaryFolder.newFolder().toPath();
|
||||||
|
SavedBiomeStore store = SavedBiomeStore.open(root);
|
||||||
|
store.claimAndPersist(chunk(0, 3L));
|
||||||
|
store.sync();
|
||||||
|
AtomicInteger forces = new AtomicInteger();
|
||||||
|
try (MockedStatic<FileChannel> ignored = regionChannels(root, (source, intercepted) ->
|
||||||
|
doAnswer(invocation -> {
|
||||||
|
forces.incrementAndGet();
|
||||||
|
source.force(true);
|
||||||
|
return null;
|
||||||
|
}).when(intercepted).force(true))) {
|
||||||
|
assertTrue(store.claimAndPersist(chunk(1, 3L)));
|
||||||
|
assertTrue(store.claimAndPersist(chunk(2, 3L)));
|
||||||
|
assertEquals(0, forces.get());
|
||||||
|
store.sync();
|
||||||
|
assertEquals(1, forces.get());
|
||||||
|
store.sync();
|
||||||
|
assertEquals(1, forces.get());
|
||||||
|
}
|
||||||
|
assertEquals(chunk(2, 3L), SavedBiomeStore.open(root).get(2, 0).orElseThrow());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void failedSyncRejectsFurtherClaimsButKeepsWrittenRecords() throws Exception {
|
||||||
|
Path root = temporaryFolder.newFolder().toPath();
|
||||||
|
SavedBiomeStore store = SavedBiomeStore.open(root);
|
||||||
|
store.claimAndPersist(chunk(0, 3L));
|
||||||
|
IOException failure = new IOException("Deferred saved biome force failed");
|
||||||
|
try (MockedStatic<FileChannel> ignored = regionChannels(root, (source, intercepted) ->
|
||||||
|
doAnswer(invocation -> {
|
||||||
|
throw failure;
|
||||||
|
}).when(intercepted).force(true))) {
|
||||||
|
assertSame(failure, assertThrows(IOException.class, store::sync));
|
||||||
|
}
|
||||||
|
assertSame(failure, assertThrows(IOException.class, () -> store.claimAndPersist(chunk(1, 3L))).getCause());
|
||||||
|
assertSame(failure, assertThrows(IOException.class, () -> store.get(5, 0)).getCause());
|
||||||
|
SavedBiomeStore reopened = SavedBiomeStore.open(root);
|
||||||
|
assertEquals(chunk(0, 3L), reopened.get(0, 0).orElseThrow());
|
||||||
|
assertTrue(reopened.get(1, 0).isEmpty());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void failedWriteRollsBackWithoutPublishingAndRetainsRollbackFailure() throws Exception {
|
||||||
for (boolean failRollback : List.of(false, true)) {
|
for (boolean failRollback : List.of(false, true)) {
|
||||||
Path root = temporaryFolder.newFolder().toPath();
|
Path root = temporaryFolder.newFolder().toPath();
|
||||||
SavedBiomeStore store = SavedBiomeStore.open(root);
|
SavedBiomeStore store = SavedBiomeStore.open(root);
|
||||||
store.claimAndPersist(chunk(0, 3L));
|
store.claimAndPersist(chunk(0, 3L));
|
||||||
byte[] original = Files.readAllBytes(regionPath(root));
|
byte[] original = Files.readAllBytes(regionPath(root));
|
||||||
IOException failure = new IOException("Saved biome force failed");
|
IOException failure = new IOException("Saved biome write failed");
|
||||||
IOException rollback = new IOException("Saved biome rollback force failed");
|
IOException rollback = new IOException("Saved biome rollback force failed");
|
||||||
AtomicBoolean first = new AtomicBoolean(true);
|
|
||||||
try (RandomAccessFile actual = new RandomAccessFile(regionPath(root).toFile(), "rw")) {
|
try (RandomAccessFile actual = new RandomAccessFile(regionPath(root).toFile(), "rw")) {
|
||||||
FileChannel channel = mock(FileChannel.class, delegatesTo(actual.getChannel()));
|
FileChannel channel = mock(FileChannel.class, delegatesTo(actual.getChannel()));
|
||||||
doAnswer(invocation -> {
|
doAnswer(invocation -> {
|
||||||
if (first.getAndSet(false)) {
|
throw failure;
|
||||||
throw failure;
|
}).when(channel).write(any(ByteBuffer[].class), anyInt(), anyInt());
|
||||||
}
|
doAnswer(invocation -> {
|
||||||
if (failRollback) {
|
if (failRollback) {
|
||||||
throw rollback;
|
throw rollback;
|
||||||
}
|
}
|
||||||
@@ -262,6 +307,24 @@ public class SavedBiomeStoreWriteTest {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private static MockedStatic<FileChannel> regionChannels(Path root, ChannelMutation mutation) {
|
||||||
|
Path region = regionPath(root);
|
||||||
|
return mockStatic(FileChannel.class, invocation -> {
|
||||||
|
FileChannel source = (FileChannel) invocation.callRealMethod();
|
||||||
|
if (invocation.getMethod().getParameterCount() != 2 || !region.equals(invocation.getArgument(0))) {
|
||||||
|
return source;
|
||||||
|
}
|
||||||
|
FileChannel intercepted = mock(FileChannel.class, delegatesTo(source));
|
||||||
|
mutation.apply(source, intercepted);
|
||||||
|
return intercepted;
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
@FunctionalInterface
|
||||||
|
private interface ChannelMutation {
|
||||||
|
void apply(FileChannel source, FileChannel intercepted) throws IOException;
|
||||||
|
}
|
||||||
|
|
||||||
private static ClaimResult claimResult(SavedBiomeStore store, SavedBiomeChunk chunk) {
|
private static ClaimResult claimResult(SavedBiomeStore store, SavedBiomeChunk chunk) {
|
||||||
try {
|
try {
|
||||||
return new ClaimResult(store.claimAndPersist(chunk), null);
|
return new ClaimResult(store.claimAndPersist(chunk), null);
|
||||||
|
|||||||
@@ -0,0 +1,84 @@
|
|||||||
|
package art.arcane.iris.world.pregen;
|
||||||
|
|
||||||
|
import art.arcane.iris.generation.runtime.Engine;
|
||||||
|
import art.arcane.iris.spi.IrisLogging;
|
||||||
|
import art.arcane.iris.world.IrisToolbelt;
|
||||||
|
import org.bukkit.World;
|
||||||
|
import org.junit.Test;
|
||||||
|
import org.mockito.InOrder;
|
||||||
|
import org.mockito.MockedStatic;
|
||||||
|
|
||||||
|
import java.lang.reflect.Constructor;
|
||||||
|
import java.lang.reflect.Field;
|
||||||
|
import java.util.concurrent.ConcurrentHashMap;
|
||||||
|
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||||
|
import java.util.concurrent.atomic.AtomicBoolean;
|
||||||
|
|
||||||
|
import static org.mockito.ArgumentMatchers.any;
|
||||||
|
import static org.mockito.ArgumentMatchers.anyLong;
|
||||||
|
import static org.mockito.ArgumentMatchers.anyString;
|
||||||
|
import static org.mockito.ArgumentMatchers.same;
|
||||||
|
import static org.mockito.Mockito.CALLS_REAL_METHODS;
|
||||||
|
import static org.mockito.Mockito.doThrow;
|
||||||
|
import static org.mockito.Mockito.inOrder;
|
||||||
|
import static org.mockito.Mockito.mock;
|
||||||
|
import static org.mockito.Mockito.mockStatic;
|
||||||
|
import static org.mockito.Mockito.verify;
|
||||||
|
import static org.mockito.Mockito.when;
|
||||||
|
|
||||||
|
public class AsyncPregenHistorySyncTest {
|
||||||
|
@Test
|
||||||
|
public void closeForcesGenerationHistoryOnceOutstandingChunksDrain() throws Exception {
|
||||||
|
Engine engine = mock(Engine.class);
|
||||||
|
PregenAdmissionGate admission = mock(PregenAdmissionGate.class);
|
||||||
|
AsyncPregenMethod method = closable(engine, admission);
|
||||||
|
try (MockedStatic<IrisToolbelt> toolbelt = mockStatic(IrisToolbelt.class)) {
|
||||||
|
toolbelt.when(IrisToolbelt::isServerStopping).thenReturn(true);
|
||||||
|
method.close();
|
||||||
|
}
|
||||||
|
InOrder order = inOrder(admission, engine);
|
||||||
|
order.verify(admission).awaitDrain(anyLong(), any(), any());
|
||||||
|
order.verify(engine).syncGenerationHistory();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void closeReportsAHistorySyncFailureAndStillFinishes() throws Exception {
|
||||||
|
Engine engine = mock(Engine.class);
|
||||||
|
IllegalStateException failure = new IllegalStateException("history force failed");
|
||||||
|
doThrow(failure).when(engine).syncGenerationHistory();
|
||||||
|
AsyncPregenMethod method = closable(engine, mock(PregenAdmissionGate.class));
|
||||||
|
try (MockedStatic<IrisToolbelt> toolbelt = mockStatic(IrisToolbelt.class);
|
||||||
|
MockedStatic<IrisLogging> logging = mockStatic(IrisLogging.class)) {
|
||||||
|
toolbelt.when(IrisToolbelt::isServerStopping).thenReturn(true);
|
||||||
|
method.close();
|
||||||
|
logging.verify(() -> IrisLogging.reportError(anyString(), same(failure)));
|
||||||
|
}
|
||||||
|
verify(engine).syncGenerationHistory();
|
||||||
|
}
|
||||||
|
|
||||||
|
private static AsyncPregenMethod closable(Engine engine, PregenAdmissionGate admission) throws Exception {
|
||||||
|
AsyncPregenMethod method = mock(AsyncPregenMethod.class, CALLS_REAL_METHODS);
|
||||||
|
World world = mock(World.class);
|
||||||
|
when(world.getName()).thenReturn("world");
|
||||||
|
set(method, "world", world);
|
||||||
|
set(method, "closing", new AtomicBoolean());
|
||||||
|
set(method, "holdsWorkerBoost", new AtomicBoolean());
|
||||||
|
set(method, "admission", admission);
|
||||||
|
set(method, "mantleCleanup", new PregenSerialWorker("Test Mantle Cleanup", "world"));
|
||||||
|
set(method, "chunkFlush", new PregenSerialWorker("Test Chunk Flush", "world"));
|
||||||
|
set(method, "regionChunks", new ConcurrentHashMap<>());
|
||||||
|
set(method, "pendingEvictions", new ConcurrentLinkedQueue<>());
|
||||||
|
set(method, "metricsEngine", engine);
|
||||||
|
Class<?> ticketExecutor = Class.forName(AsyncPregenMethod.class.getName() + "$TicketExecutor");
|
||||||
|
Constructor<?> constructor = ticketExecutor.getDeclaredConstructor(AsyncPregenMethod.class);
|
||||||
|
constructor.setAccessible(true);
|
||||||
|
set(method, "executor", constructor.newInstance(method));
|
||||||
|
return method;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static void set(AsyncPregenMethod method, String name, Object value) throws Exception {
|
||||||
|
Field field = AsyncPregenMethod.class.getDeclaredField(name);
|
||||||
|
field.setAccessible(true);
|
||||||
|
field.set(method, value);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,161 @@
|
|||||||
|
package art.arcane.iris.world.storage;
|
||||||
|
|
||||||
|
import org.junit.Rule;
|
||||||
|
import org.junit.Test;
|
||||||
|
import org.junit.rules.TemporaryFolder;
|
||||||
|
import org.mockito.MockedStatic;
|
||||||
|
|
||||||
|
import java.io.File;
|
||||||
|
import java.io.IOException;
|
||||||
|
import java.nio.channels.FileChannel;
|
||||||
|
import java.nio.file.Files;
|
||||||
|
import java.nio.file.Path;
|
||||||
|
import java.util.Map;
|
||||||
|
import java.util.concurrent.ConcurrentHashMap;
|
||||||
|
import java.util.concurrent.CountDownLatch;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
import java.util.concurrent.atomic.AtomicInteger;
|
||||||
|
import java.util.concurrent.atomic.AtomicReference;
|
||||||
|
|
||||||
|
import static org.junit.Assert.assertEquals;
|
||||||
|
import static org.junit.Assert.assertNotNull;
|
||||||
|
import static org.junit.Assert.assertSame;
|
||||||
|
import static org.junit.Assert.assertThrows;
|
||||||
|
import static org.junit.Assert.assertTrue;
|
||||||
|
import static org.junit.Assume.assumeTrue;
|
||||||
|
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 class DeferredDurabilityTest {
|
||||||
|
private static final long NEVER_MILLIS = TimeUnit.HOURS.toMillis(1L);
|
||||||
|
|
||||||
|
@Rule
|
||||||
|
public final TemporaryFolder temporaryFolder = new TemporaryFolder();
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void syncForcesEachWrittenFileOnceAndThenNothing() throws Exception {
|
||||||
|
Path first = temporaryFolder.newFile().toPath();
|
||||||
|
Path second = temporaryFolder.newFile().toPath();
|
||||||
|
DeferredDurability durability = new DeferredDurability(NEVER_MILLIS, failure -> {
|
||||||
|
throw new AssertionError(failure);
|
||||||
|
});
|
||||||
|
durability.written(first);
|
||||||
|
durability.written(second);
|
||||||
|
durability.written(first);
|
||||||
|
Map<Path, AtomicInteger> forces = new ConcurrentHashMap<>();
|
||||||
|
try (MockedStatic<FileChannel> ignored = countingForces(forces, null)) {
|
||||||
|
durability.sync();
|
||||||
|
assertEquals(Map.of(first, 1, second, 1), snapshot(forces));
|
||||||
|
durability.sync();
|
||||||
|
assertEquals(Map.of(first, 1, second, 1), snapshot(forces));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void syncForcesLinkedDirectories() throws Exception {
|
||||||
|
assumeTrue(File.separatorChar != '\\');
|
||||||
|
Path directory = temporaryFolder.newFolder().toPath();
|
||||||
|
DeferredDurability durability = new DeferredDurability(NEVER_MILLIS, failure -> {
|
||||||
|
throw new AssertionError(failure);
|
||||||
|
});
|
||||||
|
durability.linked(directory);
|
||||||
|
Map<Path, AtomicInteger> forces = new ConcurrentHashMap<>();
|
||||||
|
try (MockedStatic<FileChannel> ignored = countingForces(forces, null)) {
|
||||||
|
durability.sync();
|
||||||
|
}
|
||||||
|
assertEquals(Map.of(directory, 1), snapshot(forces));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void relaxedModeTracksNothing() throws Exception {
|
||||||
|
Path file = temporaryFolder.newFile().toPath();
|
||||||
|
String previous = System.getProperty(Durability.MODE_PROPERTY);
|
||||||
|
System.setProperty(Durability.MODE_PROPERTY, "relaxed");
|
||||||
|
try {
|
||||||
|
DeferredDurability durability = new DeferredDurability(NEVER_MILLIS, failure -> {
|
||||||
|
throw new AssertionError(failure);
|
||||||
|
});
|
||||||
|
durability.written(file);
|
||||||
|
durability.linked(file.getParent());
|
||||||
|
Map<Path, AtomicInteger> forces = new ConcurrentHashMap<>();
|
||||||
|
try (MockedStatic<FileChannel> ignored = countingForces(forces, null)) {
|
||||||
|
durability.sync();
|
||||||
|
}
|
||||||
|
assertTrue(forces.isEmpty());
|
||||||
|
} finally {
|
||||||
|
if (previous == null) {
|
||||||
|
System.clearProperty(Durability.MODE_PROPERTY);
|
||||||
|
} else {
|
||||||
|
System.setProperty(Durability.MODE_PROPERTY, previous);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void replacedFilesAreSkippedAndFailuresReachTheSyncCaller() throws Exception {
|
||||||
|
Path removed = temporaryFolder.newFile().toPath();
|
||||||
|
Path failing = temporaryFolder.newFile().toPath();
|
||||||
|
DeferredDurability durability = new DeferredDurability(NEVER_MILLIS, failure -> {
|
||||||
|
throw new AssertionError(failure);
|
||||||
|
});
|
||||||
|
durability.written(removed);
|
||||||
|
durability.written(failing);
|
||||||
|
Files.delete(removed);
|
||||||
|
IOException failure = new IOException("force failed");
|
||||||
|
Map<Path, AtomicInteger> forces = new ConcurrentHashMap<>();
|
||||||
|
try (MockedStatic<FileChannel> ignored = countingForces(forces, path -> path.equals(failing) ? failure : null)) {
|
||||||
|
assertSame(failure, assertThrows(IOException.class, durability::sync));
|
||||||
|
durability.sync();
|
||||||
|
}
|
||||||
|
assertEquals(Map.of(failing, 1), snapshot(forces));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void backgroundFlushRunsAfterTheDelayAndReportsFailures() throws Exception {
|
||||||
|
Path directory = temporaryFolder.newFolder().toPath();
|
||||||
|
CountDownLatch reported = new CountDownLatch(1);
|
||||||
|
AtomicReference<IOException> failure = new AtomicReference<>();
|
||||||
|
DeferredDurability durability = new DeferredDurability(10L, background -> {
|
||||||
|
failure.set(background);
|
||||||
|
reported.countDown();
|
||||||
|
});
|
||||||
|
durability.written(directory);
|
||||||
|
assertTrue(reported.await(5L, TimeUnit.SECONDS));
|
||||||
|
assertNotNull(failure.get());
|
||||||
|
durability.sync();
|
||||||
|
}
|
||||||
|
|
||||||
|
private static Map<Path, Integer> snapshot(Map<Path, AtomicInteger> forces) {
|
||||||
|
Map<Path, Integer> values = new ConcurrentHashMap<>();
|
||||||
|
forces.forEach((path, count) -> values.put(path, count.get()));
|
||||||
|
return values;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static MockedStatic<FileChannel> countingForces(Map<Path, AtomicInteger> forces, FailurePlan failures) {
|
||||||
|
return mockStatic(FileChannel.class, invocation -> {
|
||||||
|
FileChannel source = (FileChannel) invocation.callRealMethod();
|
||||||
|
if (invocation.getMethod().getParameterCount() != 2) {
|
||||||
|
return source;
|
||||||
|
}
|
||||||
|
Path path = invocation.getArgument(0);
|
||||||
|
FileChannel intercepted = mock(FileChannel.class, delegatesTo(source));
|
||||||
|
doAnswer(force -> {
|
||||||
|
forces.computeIfAbsent(path, ignored -> new AtomicInteger()).incrementAndGet();
|
||||||
|
IOException planned = failures == null ? null : failures.failureFor(path);
|
||||||
|
if (planned != null) {
|
||||||
|
throw planned;
|
||||||
|
}
|
||||||
|
source.force(true);
|
||||||
|
return null;
|
||||||
|
}).when(intercepted).force(true);
|
||||||
|
return intercepted;
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
@FunctionalInterface
|
||||||
|
private interface FailurePlan {
|
||||||
|
IOException failureFor(Path path);
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in new issue
Block a user