diff --git a/adapters/bukkit/plugin/src/main/java/art/arcane/iris/generation/runtime/GenerationRateWindow.java b/adapters/bukkit/plugin/src/main/java/art/arcane/iris/generation/runtime/GenerationRateWindow.java new file mode 100644 index 000000000..a2b9c1776 --- /dev/null +++ b/adapters/bukkit/plugin/src/main/java/art/arcane/iris/generation/runtime/GenerationRateWindow.java @@ -0,0 +1,35 @@ +package art.arcane.iris.generation.runtime; + +final class GenerationRateWindow { + static final long TIMING_IDLE_MILLIS = 10_000L; + + private long lastSampleAtMs; + private int lastGeneratedCount; + private long lastGenerationAtMs; + + double sampleChunksPerSecond(int generatedCount, long sampledAtMs) { + int safeGeneratedCount = Math.max(0, generatedCount); + long previousAt = lastSampleAtMs; + int previousCount = lastGeneratedCount; + lastSampleAtMs = sampledAtMs; + lastGeneratedCount = safeGeneratedCount; + if (previousAt <= 0L) { + if (safeGeneratedCount > 0) { + lastGenerationAtMs = sampledAtMs; + } + return 0D; + } + if (sampledAtMs <= previousAt) { + return 0D; + } + int generatedDelta = Math.max(0, safeGeneratedCount - previousCount); + if (generatedDelta > 0) { + lastGenerationAtMs = sampledAtMs; + } + return generatedDelta * 1000D / (sampledAtMs - previousAt); + } + + boolean generationTimingsFresh(long sampledAtMs) { + return lastGenerationAtMs > 0L && sampledAtMs - lastGenerationAtMs <= TIMING_IDLE_MILLIS; + } +} diff --git a/adapters/bukkit/plugin/src/main/java/art/arcane/iris/generation/runtime/IrisEngineSVC.java b/adapters/bukkit/plugin/src/main/java/art/arcane/iris/generation/runtime/IrisEngineSVC.java index e0815a177..619b049a6 100644 --- a/adapters/bukkit/plugin/src/main/java/art/arcane/iris/generation/runtime/IrisEngineSVC.java +++ b/adapters/bukkit/plugin/src/main/java/art/arcane/iris/generation/runtime/IrisEngineSVC.java @@ -460,8 +460,12 @@ public final class IrisEngineSVC implements IrisService { continue; } - double chunksPerSecond = registered.sampleChunksPerSecond(engine.getGenerated(), now); - worldSnapshots.add(EngineTelemetrySnapshot.capture(engine, chunksPerSecond, now)); + double chunksPerSecond = registered.generationRate.sampleChunksPerSecond(engine.getGenerated(), now); + worldSnapshots.add(EngineTelemetrySnapshot.capture( + engine, + chunksPerSecond, + registered.generationRate.generationTimingsFresh(now), + now)); registered.clearMetricsFailure(); } catch (Throwable exception) { if (EngineMaintenance.isMantleClosed(exception)) { @@ -552,6 +556,7 @@ public final class IrisEngineSVC implements IrisService { progress.generated(), progress.totalChunks(), progress.chunksRemaining(), + progress.chunksInFlight(), progress.chunksPerSecond(), progress.eta(), progress.elapsed(), @@ -612,8 +617,7 @@ public final class IrisEngineSVC implements IrisService { private final AtomicReference> activeMaintenance = new AtomicReference<>(); private volatile ScheduledFuture maintenance; private volatile boolean closed; - private long lastGeneratedSampleAtMs; - private int lastGeneratedCount; + private final GenerationRateWindow generationRate = new GenerationRateWindow(); private Registered(Registration registration, ScheduledThreadPoolExecutor executor) { name = registration.name(); @@ -696,19 +700,6 @@ public final class IrisEngineSVC implements IrisService { return activeMaintenance.get() != null; } - private double sampleChunksPerSecond(int generatedCount, long sampledAtMs) { - int safeGeneratedCount = Math.max(0, generatedCount); - long previousAt = lastGeneratedSampleAtMs; - int previousCount = lastGeneratedCount; - lastGeneratedSampleAtMs = sampledAtMs; - lastGeneratedCount = safeGeneratedCount; - if (previousAt <= 0L || sampledAtMs <= previousAt) { - return 0D; - } - int generatedDelta = Math.max(0, safeGeneratedCount - previousCount); - return generatedDelta * 1000D / (sampledAtMs - previousAt); - } - private void close() { if (closed) { return; diff --git a/adapters/bukkit/plugin/src/main/java/art/arcane/iris/generation/runtime/IrisTelemetrySnapshot.java b/adapters/bukkit/plugin/src/main/java/art/arcane/iris/generation/runtime/IrisTelemetrySnapshot.java index b6cc78da1..b29c71b6b 100644 --- a/adapters/bukkit/plugin/src/main/java/art/arcane/iris/generation/runtime/IrisTelemetrySnapshot.java +++ b/adapters/bukkit/plugin/src/main/java/art/arcane/iris/generation/runtime/IrisTelemetrySnapshot.java @@ -101,6 +101,7 @@ public record IrisTelemetrySnapshot( long generated, long total, long remaining, + long inFlight, double chunksPerSecond, long etaMs, long elapsedMs, @@ -115,6 +116,7 @@ public record IrisTelemetrySnapshot( 0L, 0L, 0L, + 0L, 0D, 0L, 0L, @@ -130,6 +132,7 @@ public record IrisTelemetrySnapshot( generated = Math.max(0L, generated); total = Math.max(0L, total); remaining = Math.max(0L, remaining); + inFlight = Math.max(0L, inFlight); chunksPerSecond = Double.isFinite(chunksPerSecond) ? Math.max(0D, chunksPerSecond) : 0D; etaMs = Math.max(0L, etaMs); elapsedMs = Math.max(0L, elapsedMs); diff --git a/adapters/bukkit/plugin/src/main/java/art/arcane/iris/integration/IrisIntegrationService.java b/adapters/bukkit/plugin/src/main/java/art/arcane/iris/integration/IrisIntegrationService.java index aabbeb53c..e78bbb780 100644 --- a/adapters/bukkit/plugin/src/main/java/art/arcane/iris/integration/IrisIntegrationService.java +++ b/adapters/bukkit/plugin/src/main/java/art/arcane/iris/integration/IrisIntegrationService.java @@ -319,6 +319,7 @@ public class IrisIntegrationService implements IrisService, IntegrationServiceCo IntegrationMetricSchema.IRIS_PREGEN_GENERATED, IntegrationMetricSchema.IRIS_PREGEN_TOTAL, IntegrationMetricSchema.IRIS_PREGEN_QUEUE, + IntegrationMetricSchema.IRIS_PREGEN_REMAINING, IntegrationMetricSchema.IRIS_PREGEN_THROUGHPUT, IntegrationMetricSchema.IRIS_PREGEN_ETA_MS, IntegrationMetricSchema.IRIS_PREGEN_ELAPSED_MS, @@ -332,7 +333,8 @@ public class IrisIntegrationService implements IrisService, IntegrationServiceCo put(samples, IntegrationMetricSchema.IRIS_PREGEN_PROGRESS, pregenerator.progressPercent(), sampledAtMs); put(samples, IntegrationMetricSchema.IRIS_PREGEN_GENERATED, pregenerator.generated(), sampledAtMs); put(samples, IntegrationMetricSchema.IRIS_PREGEN_TOTAL, pregenerator.total(), sampledAtMs); - put(samples, IntegrationMetricSchema.IRIS_PREGEN_QUEUE, pregenerator.remaining(), sampledAtMs); + put(samples, IntegrationMetricSchema.IRIS_PREGEN_QUEUE, pregenerator.inFlight(), sampledAtMs); + put(samples, IntegrationMetricSchema.IRIS_PREGEN_REMAINING, pregenerator.remaining(), sampledAtMs); put(samples, IntegrationMetricSchema.IRIS_PREGEN_THROUGHPUT, pregenerator.chunksPerSecond(), sampledAtMs); put(samples, IntegrationMetricSchema.IRIS_PREGEN_ETA_MS, pregenerator.etaMs(), sampledAtMs); put(samples, IntegrationMetricSchema.IRIS_PREGEN_ELAPSED_MS, pregenerator.elapsedMs(), sampledAtMs); diff --git a/adapters/bukkit/plugin/src/test/java/art/arcane/iris/generation/runtime/GenerationRateWindowTest.java b/adapters/bukkit/plugin/src/test/java/art/arcane/iris/generation/runtime/GenerationRateWindowTest.java new file mode 100644 index 000000000..d6fb63562 --- /dev/null +++ b/adapters/bukkit/plugin/src/test/java/art/arcane/iris/generation/runtime/GenerationRateWindowTest.java @@ -0,0 +1,46 @@ +package art.arcane.iris.generation.runtime; + +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class GenerationRateWindowTest { + @Test + public void generationTimingsGoStaleOnceChunksStopGenerating() { + GenerationRateWindow window = new GenerationRateWindow(); + + window.sampleChunksPerSecond(0, 1_000L); + assertEquals(20D, window.sampleChunksPerSecond(20, 2_000L), 0D); + assertTrue(window.generationTimingsFresh(2_000L)); + + assertEquals(0D, window.sampleChunksPerSecond(20, 7_000L), 0D); + assertTrue(window.generationTimingsFresh(7_000L)); + assertEquals(0D, window.sampleChunksPerSecond(20, 12_001L), 0D); + assertFalse(window.generationTimingsFresh(12_001L)); + + window.sampleChunksPerSecond(21, 13_001L); + assertTrue(window.generationTimingsFresh(13_001L)); + } + + @Test + public void anIdleWorldNeverPublishesTimings() { + GenerationRateWindow window = new GenerationRateWindow(); + + window.sampleChunksPerSecond(0, 1_000L); + window.sampleChunksPerSecond(0, 2_000L); + + assertFalse(window.generationTimingsFresh(2_000L)); + } + + @Test + public void chunksGeneratedBeforeTheFirstSampleCountAsRecentGeneration() { + GenerationRateWindow window = new GenerationRateWindow(); + + assertEquals(0D, window.sampleChunksPerSecond(400, 5_000L), 0D); + + assertTrue(window.generationTimingsFresh(5_000L)); + assertFalse(window.generationTimingsFresh(5_000L + GenerationRateWindow.TIMING_IDLE_MILLIS + 1L)); + } +} diff --git a/adapters/bukkit/plugin/src/test/java/art/arcane/iris/integration/IrisIntegrationServiceTest.java b/adapters/bukkit/plugin/src/test/java/art/arcane/iris/integration/IrisIntegrationServiceTest.java index 7d3c6314e..4e4600e6e 100644 --- a/adapters/bukkit/plugin/src/test/java/art/arcane/iris/integration/IrisIntegrationServiceTest.java +++ b/adapters/bukkit/plugin/src/test/java/art/arcane/iris/integration/IrisIntegrationServiceTest.java @@ -49,6 +49,7 @@ public class IrisIntegrationServiceTest { 100L, 200L, 100L, + 48L, 8D, 2_000L, 5_000L, @@ -64,6 +65,8 @@ public class IrisIntegrationServiceTest { IntegrationMetricSchema.IRIS_LOADED_CHUNKS, IntegrationMetricSchema.IRIS_CHUNKS_GENERATED_TOTAL, IntegrationMetricSchema.IRIS_PREGEN_THROUGHPUT, + IntegrationMetricSchema.IRIS_PREGEN_QUEUE, + IntegrationMetricSchema.IRIS_PREGEN_REMAINING, IntegrationMetricSchema.IRIS_GENERATION_TOTAL_MS, "iris.unsupported" )); @@ -73,6 +76,8 @@ public class IrisIntegrationServiceTest { assertEquals(20D, value(samples, IntegrationMetricSchema.IRIS_LOADED_CHUNKS), 0D); assertEquals(50D, value(samples, IntegrationMetricSchema.IRIS_CHUNKS_GENERATED_TOTAL), 0D); assertEquals(8D, value(samples, IntegrationMetricSchema.IRIS_PREGEN_THROUGHPUT), 0D); + assertEquals(48D, value(samples, IntegrationMetricSchema.IRIS_PREGEN_QUEUE), 0D); + assertEquals(100D, value(samples, IntegrationMetricSchema.IRIS_PREGEN_REMAINING), 0D); assertEquals(4D, value(samples, IntegrationMetricSchema.IRIS_GENERATION_TOTAL_MS), 0D); assertFalse(samples.get("iris.unsupported").available()); assertEquals("unsupported-key", samples.get("iris.unsupported").message()); @@ -84,9 +89,32 @@ public class IrisIntegrationServiceTest { assertEquals("minecraft:overworld", overworld.scopeId()); assertEquals(1D, value(overworld.samples(), IntegrationMetricSchema.IRIS_PREGEN_ACTIVE), 0D); assertEquals(8D, value(overworld.samples(), IntegrationMetricSchema.IRIS_PREGEN_THROUGHPUT), 0D); + assertEquals(48D, value(overworld.samples(), IntegrationMetricSchema.IRIS_PREGEN_QUEUE), 0D); + assertEquals(100D, value(overworld.samples(), IntegrationMetricSchema.IRIS_PREGEN_REMAINING), 0D); assertEquals("minecraft:the_nether", nether.scopeId()); assertEquals(0D, value(nether.samples(), IntegrationMetricSchema.IRIS_PREGEN_ACTIVE), 0D); assertFalse(nether.samples().get(IntegrationMetricSchema.IRIS_PREGEN_QUEUE).available()); + assertFalse(nether.samples().get(IntegrationMetricSchema.IRIS_PREGEN_REMAINING).available()); + } + + @Test + public void idleWorldsPublishGenerationTimingsAsUnavailable() { + long now = System.currentTimeMillis(); + EngineTelemetrySnapshot idle = new EngineTelemetrySnapshot( + now, "minecraft:overworld", "world", "dimension", true, false, false, false, + 12L, 3L, 0.25D, 20L, 20L, 0D, 0L, 2, 0, 1L, 4L, 1L, 10D, Map.of()); + IrisIntegrationService service = new IrisIntegrationService( + () -> telemetry(List.of(idle), IrisTelemetrySnapshot.PregenSnapshot.INACTIVE, now) + ); + + IntegrationMetricSample total = service.sampleMetrics(Set.of(IntegrationMetricSchema.IRIS_GENERATION_TOTAL_MS)) + .get(IntegrationMetricSchema.IRIS_GENERATION_TOTAL_MS); + IntegrationMetricSample worldTotal = service.metricGroups().get(0).samples() + .get(IntegrationMetricSchema.IRIS_GENERATION_TOTAL_MS); + + assertFalse(total.available()); + assertEquals("timing-not-available", total.message()); + assertFalse(worldTotal.available()); } @Test diff --git a/core/src/main/java/art/arcane/iris/generation/runtime/EngineTelemetrySnapshot.java b/core/src/main/java/art/arcane/iris/generation/runtime/EngineTelemetrySnapshot.java index 6912fcf5c..3d5371d80 100644 --- a/core/src/main/java/art/arcane/iris/generation/runtime/EngineTelemetrySnapshot.java +++ b/core/src/main/java/art/arcane/iris/generation/runtime/EngineTelemetrySnapshot.java @@ -52,7 +52,12 @@ public record EngineTelemetrySnapshot( generationTimingsMs = sanitizeTimings(generationTimingsMs); } - public static EngineTelemetrySnapshot capture(Engine engine, double chunksPerSecond, long sampledAtMs) { + public static EngineTelemetrySnapshot capture( + Engine engine, + double chunksPerSecond, + boolean generationTimingsFresh, + long sampledAtMs + ) { if (engine == null) { throw new IllegalArgumentException("Engine cannot be null"); } @@ -85,7 +90,7 @@ public record EngineTelemetrySnapshot( engine.getMantle().getLoadedRegionCount(), engine.getMantle().getUnloadRegionCount(), engine.getMantle().getAdjustedIdleDuration(), - engine.getMetrics().telemetryAverages() + generationTimingsFresh ? engine.getMetrics().telemetryAverages() : Map.of() ); } diff --git a/core/src/main/java/art/arcane/iris/studio/view/PregeneratorJob.java b/core/src/main/java/art/arcane/iris/studio/view/PregeneratorJob.java index afe412d8f..c90ed45fd 100644 --- a/core/src/main/java/art/arcane/iris/studio/view/PregeneratorJob.java +++ b/core/src/main/java/art/arcane/iris/studio/view/PregeneratorJob.java @@ -302,7 +302,8 @@ public class PregeneratorJob implements PregenListener, PregenRenderSource { public record PregenProgress(double percent, long generated, long totalChunks, double chunksPerSecond, double overallChunksPerSecond, double thirtySecondChunksPerSecond, - double sixtySecondChunksPerSecond, long chunksRemaining, long eta, long elapsed, + double sixtySecondChunksPerSecond, long chunksRemaining, long chunksInFlight, + long eta, long elapsed, String method, boolean paused, long failed, String worldName, String worldIdentity) { } @@ -322,6 +323,7 @@ public class PregeneratorJob implements PregenListener, PregenRenderSource { Math.max(0D, lastThirtySecondChunksPerSecond), Math.max(0D, lastSixtySecondChunksPerSecond), Math.max(0L, lastChunksRemaining), + pregenerator.getInFlightChunks(), lastEta, lastElapsed, lastMethod, diff --git a/core/src/main/java/art/arcane/iris/world/pregen/AsyncOrMedievalPregenMethod.java b/core/src/main/java/art/arcane/iris/world/pregen/AsyncOrMedievalPregenMethod.java index e4f50bd83..30f4cbc89 100644 --- a/core/src/main/java/art/arcane/iris/world/pregen/AsyncOrMedievalPregenMethod.java +++ b/core/src/main/java/art/arcane/iris/world/pregen/AsyncOrMedievalPregenMethod.java @@ -105,6 +105,11 @@ public class AsyncOrMedievalPregenMethod implements PregeneratorMethod { method.reclaimMemory(); } + @Override + public int inFlightChunks() { + return method.inFlightChunks(); + } + @Override public Mantle getMantle() { return method.getMantle(); diff --git a/core/src/main/java/art/arcane/iris/world/pregen/AsyncPregenMethod.java b/core/src/main/java/art/arcane/iris/world/pregen/AsyncPregenMethod.java index 09d5e0954..78045cf87 100644 --- a/core/src/main/java/art/arcane/iris/world/pregen/AsyncPregenMethod.java +++ b/core/src/main/java/art/arcane/iris/world/pregen/AsyncPregenMethod.java @@ -1127,6 +1127,11 @@ public class AsyncPregenMethod implements PregeneratorMethod { return null; } + @Override + public int inFlightChunks() { + return Math.max(0, inFlight.get()); + } + @Override public Mantle getMantle() { if (IrisToolbelt.isIrisWorld(world)) { diff --git a/core/src/main/java/art/arcane/iris/world/pregen/CachedPregenMethod.java b/core/src/main/java/art/arcane/iris/world/pregen/CachedPregenMethod.java index ecd559cd7..71d49725b 100644 --- a/core/src/main/java/art/arcane/iris/world/pregen/CachedPregenMethod.java +++ b/core/src/main/java/art/arcane/iris/world/pregen/CachedPregenMethod.java @@ -213,6 +213,11 @@ public class CachedPregenMethod implements PregeneratorMethod { method.reclaimMemory(); } + @Override + public int inFlightChunks() { + return method.inFlightChunks(); + } + @Override public Mantle getMantle() { return method.getMantle(); diff --git a/core/src/main/java/art/arcane/iris/world/pregen/HybridPregenMethod.java b/core/src/main/java/art/arcane/iris/world/pregen/HybridPregenMethod.java index e1855d7f9..a7576d8ef 100644 --- a/core/src/main/java/art/arcane/iris/world/pregen/HybridPregenMethod.java +++ b/core/src/main/java/art/arcane/iris/world/pregen/HybridPregenMethod.java @@ -98,6 +98,11 @@ public class HybridPregenMethod implements PregeneratorMethod { inWorld.reclaimMemory(); } + @Override + public int inFlightChunks() { + return inWorld.inFlightChunks(); + } + @Override public Mantle getMantle() { return inWorld.getMantle(); diff --git a/core/src/main/java/art/arcane/iris/world/pregen/IrisPregenerator.java b/core/src/main/java/art/arcane/iris/world/pregen/IrisPregenerator.java index 1b56eaf85..74e3b251c 100644 --- a/core/src/main/java/art/arcane/iris/world/pregen/IrisPregenerator.java +++ b/core/src/main/java/art/arcane/iris/world/pregen/IrisPregenerator.java @@ -464,6 +464,10 @@ public class IrisPregenerator { return failed.get(); } + public int getInFlightChunks() { + return generator.inFlightChunks(); + } + public PregenRates getRates() { return rates; } diff --git a/core/src/main/java/art/arcane/iris/world/pregen/MedievalPregenMethod.java b/core/src/main/java/art/arcane/iris/world/pregen/MedievalPregenMethod.java index 26687e469..2f4541540 100644 --- a/core/src/main/java/art/arcane/iris/world/pregen/MedievalPregenMethod.java +++ b/core/src/main/java/art/arcane/iris/world/pregen/MedievalPregenMethod.java @@ -52,6 +52,7 @@ public class MedievalPregenMethod implements PregeneratorMethod { private static final long UNLOAD_TIMEOUT_SECONDS = 120L; private final World world; private final KList> futures; + private final AtomicInteger inFlight; private final Map lastUse; private final int maxFutures; private final AtomicBoolean directAsyncDisabled; @@ -67,6 +68,7 @@ public class MedievalPregenMethod implements PregeneratorMethod { this.chunkFlush = new PregenSerialWorker("Iris Pregen Chunk Flush", world == null ? "unknown" : world.getName()); this.chunkIoExecutor = chunkFlush.executor(); futures = new KList<>(); + this.inFlight = new AtomicInteger(); this.lastUse = new ConcurrentHashMap<>(); int configuredThreads = IrisSettings.getThreadCount(IrisSettings.get().getConcurrency().getParallelism()); this.maxFutures = J.isFolia() ? Math.max(2, Math.min(64, configuredThreads)) : Math.max(16, Math.min(128, configuredThreads * 4)); @@ -229,6 +231,7 @@ public class MedievalPregenMethod implements PregeneratorMethod { } listener.onChunkGenerating(x, z); + inFlight.incrementAndGet(); // Single choke point for failure accounting: without onChunkFailed a lossy run can // never satisfy allVisitsComplete, so a finished pregen reports as aborted forever. CompletableFuture chunkFuture = J.isFolia() @@ -247,6 +250,7 @@ public class MedievalPregenMethod implements PregeneratorMethod { }) : scheduleChunkLoad(x, z, listener); futures.add(chunkFuture.whenComplete((r, err) -> { + inFlight.decrementAndGet(); if (err != null) { listener.onChunkFailed(x, z); } @@ -358,6 +362,11 @@ public class MedievalPregenMethod implements PregeneratorMethod { } } + @Override + public int inFlightChunks() { + return Math.max(0, inFlight.get()); + } + @Override public Mantle getMantle() { if (IrisToolbelt.isIrisWorld(world)) { diff --git a/core/src/main/java/art/arcane/iris/world/pregen/PregeneratorMethod.java b/core/src/main/java/art/arcane/iris/world/pregen/PregeneratorMethod.java index ee0749620..c4cc24faa 100644 --- a/core/src/main/java/art/arcane/iris/world/pregen/PregeneratorMethod.java +++ b/core/src/main/java/art/arcane/iris/world/pregen/PregeneratorMethod.java @@ -95,5 +95,9 @@ public interface PregeneratorMethod { default void reclaimMemory() { } + default int inFlightChunks() { + return 0; + } + Mantle getMantle(); } diff --git a/core/src/test/java/art/arcane/iris/world/pregen/CachedPregenMethodCompletionTest.java b/core/src/test/java/art/arcane/iris/world/pregen/CachedPregenMethodCompletionTest.java index 2bd85cbfe..76eb37e94 100644 --- a/core/src/test/java/art/arcane/iris/world/pregen/CachedPregenMethodCompletionTest.java +++ b/core/src/test/java/art/arcane/iris/world/pregen/CachedPregenMethodCompletionTest.java @@ -62,6 +62,13 @@ public class CachedPregenMethodCompletionTest { assertEquals(1, listener.generated.get()); } + @Test + public void inFlightChunksReportTheWrappedMethodsOutstandingRequests() { + underlying.inFlight.set(37); + + assertEquals(37, method.inFlightChunks()); + } + @Test public void generateChunkFailureIsNotCached() { method.generateChunk(6, 9, listener); @@ -186,11 +193,17 @@ public class CachedPregenMethodCompletionTest { private final AtomicInteger centerBlockX = new AtomicInteger(); private final AtomicInteger centerBlockZ = new AtomicInteger(); private final AtomicReference capturedListener = new AtomicReference<>(); + private final AtomicInteger inFlight = new AtomicInteger(); @Override public void init() { } + @Override + public int inFlightChunks() { + return inFlight.get(); + } + @Override public void close() { }