From 3af79e44fe0871cf938b402301aad031abdfb981 Mon Sep 17 00:00:00 2001 From: Mike Horgan Date: Thu, 13 Aug 2026 14:07:59 -0400 Subject: [PATCH 1/4] fix recycled read circulation --- CHANGELOG.md | 4 ++ .../local/context/LoadStepContextImpl.java | 19 ------ .../LoadGeneratorImplRecycleTest.java | 63 +++++++++++++++++++ .../context/LoadStepContextImplTest.java | 21 +++++++ 4 files changed, 88 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2c0b3bc1..f8c744f6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/), and this ## [Unreleased] +### Fixed + +- **Recycled read working-set coverage** — Route completed operations back through the shared generator so low-concurrency duration reads circulate through the complete input manifest before repeating objects. + ## [5.14.1] - 2026-08-11 ### Added diff --git a/engine/core/spt-base/src/main/java/com/dell/spt/base/load/step/local/context/LoadStepContextImpl.java b/engine/core/spt-base/src/main/java/com/dell/spt/base/load/step/local/context/LoadStepContextImpl.java index e2491023..8121a197 100644 --- a/engine/core/spt-base/src/main/java/com/dell/spt/base/load/step/local/context/LoadStepContextImpl.java +++ b/engine/core/spt-base/src/main/java/com/dell/spt/base/load/step/local/context/LoadStepContextImpl.java @@ -281,25 +281,6 @@ public LoadStepContextImpl( com.dell.spt.base.metrics.MetricsConstants.METADATA_LIST_SHARD_METRICS, this.listShardMetricsRecorder); } - // Enable fast-recycle on the driver when recycling simple data ops - // without content updates. The threshold is the configured concurrency - // limit (or a cap of 8 for unlimited). updateContents is excluded - // because the driver re-submits the original op directly and cannot - // safely mutate the shared DataItem offset concurrently with metrics. - if (this.recycleFlag && !this.updateContents && !this.listPathWorkload) { - final int driverConcurrency = this.driver.concurrencyLimit(); - final int threshold = driverConcurrency > 0 ? Math.min(driverConcurrency, 8) : 8; - this.driver.enableFastRecycle(threshold); - // Only quiesce when the configured concurrency is low enough that - // fast-recycle handles most operations inline. At higher concurrency - // (T8+), fast-recycle rarely fires and the generator/dispatch VTs - // need to stay responsive — yield is faster than park+unpark when - // ops flow through the recycleQueue continuously. - if (driverConcurrency > 0 && driverConcurrency <= 4) { - this.generator.enableFastRecycleQuiesce(); - this.driver.enableFastRecycleQuiesce(); - } - } } /** Resolve the MetricsContext for a given operation type (mixed-mode routing). */ diff --git a/engine/core/spt-base/src/test/java/com/dell/spt/base/load/generator/LoadGeneratorImplRecycleTest.java b/engine/core/spt-base/src/test/java/com/dell/spt/base/load/generator/LoadGeneratorImplRecycleTest.java index 4a59613f..c570290b 100644 --- a/engine/core/spt-base/src/test/java/com/dell/spt/base/load/generator/LoadGeneratorImplRecycleTest.java +++ b/engine/core/spt-base/src/test/java/com/dell/spt/base/load/generator/LoadGeneratorImplRecycleTest.java @@ -17,6 +17,7 @@ import com.dell.spt.base.storage.driver.StorageDriver; import com.github.akurilov.commons.io.Input; import com.github.akurilov.commons.io.Output; +import java.io.EOFException; import java.io.IOException; import java.util.ArrayList; import java.util.List; @@ -145,6 +146,68 @@ void batchRecycleOutputsAllOps() throws Exception { assertEquals(BATCH_SIZE, generator.generatedOpCount()); } + @Test + void initialManifestCirculatesCompletelyBeforeAnyRecycledOperation() throws Exception { + final int itemCount = 10; + final List sourceItems = new ArrayList<>(); + for (int i = 0; i < itemCount; i++) { + sourceItems.add(new DataItemImpl("manifest-" + i, 0, 1024)); + } + final AtomicInteger nextItem = new AtomicInteger(); + doAnswer(invocation -> { + @SuppressWarnings("unchecked") + final List buffer = invocation.getArgument(0); + final int limit = invocation.getArgument(1); + final int from = nextItem.get(); + if (from >= sourceItems.size()) { + throw new EOFException("end of manifest"); + } + final int to = Math.min(from + limit, sourceItems.size()); + buffer.addAll(sourceItems.subList(from, to)); + nextItem.set(to); + return to - from; + }).when(itemInput).get(anyList(), anyInt()); + doAnswer(invocation -> { + @SuppressWarnings("unchecked") + final List items = invocation.getArgument(0); + @SuppressWarnings("unchecked") + final List> operations = invocation.getArgument(1); + for (final DataItem item : items) { + operations.add(newOp(item.name())); + } + return null; + }).when(opsBuilder).buildOps(anyList(), anyList()); + + final LoadGeneratorImpl> circulatingGenerator = new LoadGeneratorImpl<>( + itemInput, opsBuilder, List.of(), output, BATCH_SIZE, + 0, 1000, true, false); + try { + for (int batch = 0; batch < 3; batch++) { + final int from = output.received.size(); + circulatingGenerator.doWork(); + final int to = output.received.size(); + for (int i = from; i < to; i++) { + circulatingGenerator.recycle(output.received.get(i)); + } + } + assertEquals(itemCount, output.received.size()); + assertEquals( + sourceItems.stream().map(DataItem::name).toList(), + output.received.stream().map(op -> op.item().name()).toList()); + + circulatingGenerator.doWork(); + assertTrue(circulatingGenerator.isItemInputFinished()); + circulatingGenerator.doWork(); + assertEquals(itemCount + BATCH_SIZE, output.received.size()); + assertEquals( + List.of("manifest-0", "manifest-1", "manifest-2", "manifest-3"), + output.received.subList(itemCount, itemCount + BATCH_SIZE) + .stream().map(op -> op.item().name()).toList()); + } finally { + circulatingGenerator.close(); + } + } + /** * Start the generator VT loop, recycle ops from the test thread, * and verify all ops eventually reach the output. diff --git a/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/LoadStepContextImplTest.java b/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/LoadStepContextImplTest.java index a6f65eb9..c09bcb48 100644 --- a/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/LoadStepContextImplTest.java +++ b/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/LoadStepContextImplTest.java @@ -66,6 +66,7 @@ import org.junit.jupiter.api.io.TempDir; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.after; import static org.mockito.Mockito.doNothing; @@ -133,6 +134,26 @@ public void loadStepRunTest() throws IOException { Assertions.assertTrue(stepCtx.isDone()); } + @Test + public void recycleWorkloadUsesSharedGeneratorCirculation() { + testConfig.val("load-op-retry", false); + testConfig.val("load-op-recycle-mode", true); + + @SuppressWarnings("unchecked") + final LoadGenerator> generatorMock = mock(LoadGenerator.class); + @SuppressWarnings("unchecked") + final StorageDriver> driverMock = mock(StorageDriver.class); + when(driverMock.concurrencyLimit()).thenReturn(4); + + new LoadStepContextImpl<>( + "recycle-through-generator", generatorMock, driverMock, null, + testConfig.configVal("load"), false); + + verify(driverMock, never()).enableFastRecycle(anyInt()); + verify(driverMock, never()).enableFastRecycleQuiesce(); + verify(generatorMock, never()).enableFastRecycleQuiesce(); + } + @Test public void putSingleOperationSuccessAndMetricsOutput() throws Exception { // enable recycle to populate latestSuccOpResultByItem map From 51895638441708cc6409507430bb0fb4b5a5c089 Mon Sep 17 00:00:00 2001 From: Mike Horgan Date: Thu, 13 Aug 2026 14:11:16 -0400 Subject: [PATCH 2/4] test complete recycled read traversal --- .../LoadGeneratorImplRecycleTest.java | 22 ++++++++++++++----- .../context/LoadStepContextImplTest.java | 8 ++++--- 2 files changed, 21 insertions(+), 9 deletions(-) diff --git a/engine/core/spt-base/src/test/java/com/dell/spt/base/load/generator/LoadGeneratorImplRecycleTest.java b/engine/core/spt-base/src/test/java/com/dell/spt/base/load/generator/LoadGeneratorImplRecycleTest.java index c570290b..60f0994d 100644 --- a/engine/core/spt-base/src/test/java/com/dell/spt/base/load/generator/LoadGeneratorImplRecycleTest.java +++ b/engine/core/spt-base/src/test/java/com/dell/spt/base/load/generator/LoadGeneratorImplRecycleTest.java @@ -197,12 +197,22 @@ void initialManifestCirculatesCompletelyBeforeAnyRecycledOperation() throws Exce circulatingGenerator.doWork(); assertTrue(circulatingGenerator.isItemInputFinished()); - circulatingGenerator.doWork(); - assertEquals(itemCount + BATCH_SIZE, output.received.size()); - assertEquals( - List.of("manifest-0", "manifest-1", "manifest-2", "manifest-3"), - output.received.subList(itemCount, itemCount + BATCH_SIZE) - .stream().map(op -> op.item().name()).toList()); + final List expectedNames = sourceItems.stream().map(DataItem::name).toList(); + for (int cycle = 0; cycle < 2; cycle++) { + final int from = output.received.size(); + for (int batch = 0; batch < 3; batch++) { + circulatingGenerator.doWork(); + } + final int to = output.received.size(); + assertEquals(itemCount, to - from); + assertEquals( + expectedNames, + output.received.subList(from, to) + .stream().map(op -> op.item().name()).toList()); + for (int i = from; i < to; i++) { + circulatingGenerator.recycle(output.received.get(i)); + } + } } finally { circulatingGenerator.close(); } diff --git a/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/LoadStepContextImplTest.java b/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/LoadStepContextImplTest.java index c09bcb48..67d49f47 100644 --- a/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/LoadStepContextImplTest.java +++ b/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/LoadStepContextImplTest.java @@ -134,8 +134,10 @@ public void loadStepRunTest() throws IOException { Assertions.assertTrue(stepCtx.isDone()); } - @Test - public void recycleWorkloadUsesSharedGeneratorCirculation() { + @org.junit.jupiter.params.ParameterizedTest + @org.junit.jupiter.params.provider.ValueSource(ints = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10 + }) + public void recycleWorkloadUsesSharedGeneratorCirculation(final int concurrency) { testConfig.val("load-op-retry", false); testConfig.val("load-op-recycle-mode", true); @@ -143,7 +145,7 @@ public void recycleWorkloadUsesSharedGeneratorCirculation() { final LoadGenerator> generatorMock = mock(LoadGenerator.class); @SuppressWarnings("unchecked") final StorageDriver> driverMock = mock(StorageDriver.class); - when(driverMock.concurrencyLimit()).thenReturn(4); + when(driverMock.concurrencyLimit()).thenReturn(concurrency); new LoadStepContextImpl<>( "recycle-through-generator", generatorMock, driverMock, null, From b1e85b6339deb4c8411d9094e5b8cbb9b9685c81 Mon Sep 17 00:00:00 2001 From: Mike Horgan Date: Thu, 13 Aug 2026 16:15:54 -0400 Subject: [PATCH 3/4] harden recycle circulation fix --- CHANGELOG.md | 6 +- .../com/dell/spt/base/item/op/Operation.java | 23 +- .../dell/spt/base/item/op/OperationImpl.java | 14 +- .../base/load/generator/LoadGenerator.java | 9 +- .../load/generator/LoadGeneratorImpl.java | 38 +- .../local/context/LoadStepContextImpl.java | 17 +- .../base/storage/driver/StorageDriver.java | 27 +- .../spt/base/item/op/OperationImplTest.java | 32 -- .../base/item/op/OperationResultCopyTest.java | 34 +- .../LoadGeneratorImplRecycleTest.java | 194 +-------- .../context/LoadStepContextImplTest.java | 23 -- .../RecycleCirculationIntegrationTest.java | 390 ++++++++++++++++++ .../driver/coop/CoopStorageDriverBase.java | 61 +-- .../coop/CoopStorageDriverBaseTest.java | 159 +------ .../coop/OperationDispatchTaskTest.java | 3 +- .../coop/netty/NettyStorageDriverBase.java | 44 +- .../coop/netty/NettyCompletionPathTest.java | 224 +--------- 17 files changed, 474 insertions(+), 824 deletions(-) create mode 100644 engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/RecycleCirculationIntegrationTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index f8c744f6..2932aa69 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,9 +6,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/), and this ## [Unreleased] +### Changed + +- **Recycled read dispatch** — Disabled the low-concurrency fast-recycle bypass introduced in v5.7.0 because directly resubmitting the same in-flight operations could restrict a duration-based read to a working set equal to concurrency. Completed operations now return through shared generator circulation, trading some low-concurrency dispatch efficiency for complete manifest coverage. + ### Fixed -- **Recycled read working-set coverage** — Route completed operations back through the shared generator so low-concurrency duration reads circulate through the complete input manifest before repeating objects. +- **Recycled read working-set coverage** — Low-concurrency duration reads now circulate through the complete input manifest before repeating objects. ## [5.14.1] - 2026-08-11 diff --git a/engine/core/spt-base/src/main/java/com/dell/spt/base/item/op/Operation.java b/engine/core/spt-base/src/main/java/com/dell/spt/base/item/op/Operation.java index 64712e66..c4010352 100644 --- a/engine/core/spt-base/src/main/java/com/dell/spt/base/item/op/Operation.java +++ b/engine/core/spt-base/src/main/java/com/dell/spt/base/item/op/Operation.java @@ -111,14 +111,23 @@ default void buildItemPath(final I item, final String itemPath) { void resetTiming(); /** - * Returns {@code true} if the storage driver has already recycled this operation - * via the fast-recycle path (re-submitted directly without going through the - * LoadGenerator recycle queue). When set, {@code LoadStepContextImpl} must - * skip calling {@code generator.recycle()} to avoid double-recycling. + * Legacy compatibility accessor for the removed direct fast-recycle path. + * + * @deprecated always returns {@code false} */ - boolean driverRecycled(); + @Deprecated + default boolean driverRecycled() { + return false; + } - void driverRecycled(boolean flag); + /** + * Legacy compatibility mutator for the removed direct fast-recycle path. + * + * @param flag ignored + * @deprecated always a no-op + */ + @Deprecated + default void driverRecycled(final boolean flag) {} /** * Number of times this operation has been retried after a failure via {@code @@ -136,7 +145,7 @@ default void incrementOpRetryCount() {} /** * Reset the whole-operation retry counter, e.g. after a terminal success, so a - * recycled/fast-recycled operation starts its next logical attempt with a clean budget + * recycled operation starts its next logical attempt with a clean budget * instead of accumulating retry counts across unrelated cycles. */ default void resetOpRetryCount() {} diff --git a/engine/core/spt-base/src/main/java/com/dell/spt/base/item/op/OperationImpl.java b/engine/core/spt-base/src/main/java/com/dell/spt/base/item/op/OperationImpl.java index e8005967..795c144a 100644 --- a/engine/core/spt-base/src/main/java/com/dell/spt/base/item/op/OperationImpl.java +++ b/engine/core/spt-base/src/main/java/com/dell/spt/base/item/op/OperationImpl.java @@ -25,6 +25,8 @@ public class OperationImpl implements Operation { protected volatile long reqTimeDone; protected volatile long respTimeStart; protected volatile long respTimeDone; + /** @deprecated inert compatibility field for subclasses compiled against the removed fast-recycle path */ + @Deprecated protected volatile boolean driverRecycled; protected volatile int opRetryCount; protected volatile String requestedVersionId; @@ -95,7 +97,6 @@ protected OperationImpl(final OperationImpl other) { this.reqTimeDone = other.reqTimeDone; this.respTimeStart = other.respTimeStart; this.respTimeDone = other.respTimeDone; - this.driverRecycled = other.driverRecycled; // Deliberately propagated (not reset in reset() below): this must survive across // the reset()+redispatch cycle a retried operation goes through, or the retry // counter added for load-op-retry could never reach its limit. @@ -127,17 +128,6 @@ public void reset() { returnedVersionId = null; responseRequestId = null; integrityVerificationResult = null; - driverRecycled = false; - } - - @Override - public final boolean driverRecycled() { - return driverRecycled; - } - - @Override - public final void driverRecycled(final boolean flag) { - this.driverRecycled = flag; } @Override diff --git a/engine/core/spt-base/src/main/java/com/dell/spt/base/load/generator/LoadGenerator.java b/engine/core/spt-base/src/main/java/com/dell/spt/base/load/generator/LoadGenerator.java index a7c63ded..e62112ae 100644 --- a/engine/core/spt-base/src/main/java/com/dell/spt/base/load/generator/LoadGenerator.java +++ b/engine/core/spt-base/src/main/java/com/dell/spt/base/load/generator/LoadGenerator.java @@ -76,12 +76,11 @@ default List drainPendingRetries() { } /** - * Enable idle-task quiescing for fast-recycle workloads. When enabled, the - * generator parks its task thread (instead of spin-waiting/yielding) when the recycle - * queue is empty, because the driver is handling recycling inline. The - * {@link #recycle} method will unpark the task immediately if an op falls back - * to the normal path. + * Legacy compatibility hook for the removed direct fast-recycle path. + * + * @deprecated always a no-op; completed operations must return through {@link #recycle} */ + @Deprecated default void enableFastRecycleQuiesce() {} /** diff --git a/engine/core/spt-base/src/main/java/com/dell/spt/base/load/generator/LoadGeneratorImpl.java b/engine/core/spt-base/src/main/java/com/dell/spt/base/load/generator/LoadGeneratorImpl.java index 17e182d0..4a6c673c 100644 --- a/engine/core/spt-base/src/main/java/com/dell/spt/base/load/generator/LoadGeneratorImpl.java +++ b/engine/core/spt-base/src/main/java/com/dell/spt/base/load/generator/LoadGeneratorImpl.java @@ -32,7 +32,6 @@ import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.LongAdder; import java.util.concurrent.locks.Lock; -import java.util.concurrent.locks.LockSupport; import java.util.concurrent.locks.ReentrantLock; import org.apache.logging.log4j.Level; import org.apache.logging.log4j.ThreadContext; @@ -71,9 +70,6 @@ public class LoadGeneratorImpl> extends T private final boolean shuffleFlag; private final Random rnd; private final String name; - private volatile boolean fastRecycleQuiesce = false; - private volatile boolean generatorParked = false; - private volatile Thread generatorThread; private final ThreadLocal> threadLocalOpBuff; private final LongAdder builtTasksCounter = new LongAdder(); private final LongAdder recycledOpCounter = new LongAdder(); @@ -144,7 +140,6 @@ public LoadGeneratorImpl( @Override protected void doInit() { ThreadContext.put(KEY_CLASS_NAME, CLS_NAME); - generatorThread = Thread.currentThread(); } @Override @@ -194,20 +189,9 @@ protected final void doWork() throws Exception { pendingOpCount += n; recycledOpCounter.add(n); } else { - // No recycled ops available right now. - if (fastRecycleQuiesce) { - // Fast-recycle is handling ops inline in the driver; - // park this task thread until recycle() unparks us or the - // timeout expires. 10ms keeps it idle - // while still bounding wake-up latency for fallback ops. - generatorParked = true; - LockSupport.parkNanos(10_000_000); - generatorParked = false; - } else { - // Yield the task thread so in-flight ops can complete - // without a timed parking syscall. - yieldThread(); - } + // Yield so in-flight operations can complete and return through + // recycleQueue without imposing a timed parking syscall. + yieldThread(); } } } else { @@ -453,21 +437,11 @@ public final void recycle(final O op) { recycleQueueFullState = true; Loggers.ERR.warn("{}: recycle queue exceeded configured capacity ({})", name, recycleQueueCapacity); } - // Wake the generator task if it's actually parked in the quiesce state. - // Checking generatorParked avoids spurious unpark() calls when the - // generator is running (e.g. at higher concurrency where fast-recycle - // doesn't handle all ops). - if (fastRecycleQuiesce && generatorParked) { - LockSupport.unpark(generatorThread); - } } @Override public final void retry(final O op) { retryQueue.add(op); - if (fastRecycleQuiesce && generatorParked) { - LockSupport.unpark(generatorThread); - } } /** @@ -564,10 +538,8 @@ public final List drainPendingRetries() { } @Override - public void enableFastRecycleQuiesce() { - fastRecycleQuiesce = true; - Loggers.MSG.info("{}: fast-recycle quiesce enabled (generator task will park when idle)", name); - } + @Deprecated + public void enableFastRecycleQuiesce() {} private boolean isFinished() { // Never actually finish while a load-op-retry redispatch is still waiting to be diff --git a/engine/core/spt-base/src/main/java/com/dell/spt/base/load/step/local/context/LoadStepContextImpl.java b/engine/core/spt-base/src/main/java/com/dell/spt/base/load/step/local/context/LoadStepContextImpl.java index 8121a197..bba8b01f 100644 --- a/engine/core/spt-base/src/main/java/com/dell/spt/base/load/step/local/context/LoadStepContextImpl.java +++ b/engine/core/spt-base/src/main/java/com/dell/spt/base/load/step/local/context/LoadStepContextImpl.java @@ -503,9 +503,9 @@ public final boolean put(final O opResult) { final Status status = opResult.status(); if (Status.SUCC.equals(status)) { // A terminal success ends this operation's current retry episode - clear the - // counter now so a recycled/fast-recycled reuse of this object (read-recycle - // mode, or Netty's fast-recycle short-circuit) starts its next attempt with a - // clean budget instead of accumulating retry counts across unrelated cycles. + // counter now so a recycled reuse of this object (read-recycle mode) starts its + // next attempt with a clean budget instead of accumulating retry counts across + // unrelated cycles. opResult.resetOpRetryCount(); final long reqDuration = opResult.duration(); final long respLatency = opResult.latency(); @@ -553,12 +553,7 @@ public final boolean put(final O opResult) { // TODO: possible change: remove dataItem.offset() to improve perf and increase variability dataItem.offset(dataItem.offset() + rand.get().nextLong()); } - // Skip generator.recycle() when the driver already re-submitted the - // original op via the fast-recycle path — calling recycle here would - // create a duplicate in-flight operation. - if (!opResult.driverRecycled()) { - generator.recycle(opResult); - } + generator.recycle(opResult); } // each recycled op's lat and dur should be written to file each time @@ -675,9 +670,7 @@ public final int put(final List opResults, final int from, final int to) { listShardMetricsRecorder.onRequeue(shardRef); } } - if (!opResult.driverRecycled()) { - generator.recycle(opResult); - } + generator.recycle(opResult); } // each recycled op's lat and dur should be written to file each time diff --git a/engine/core/spt-base/src/main/java/com/dell/spt/base/storage/driver/StorageDriver.java b/engine/core/spt-base/src/main/java/com/dell/spt/base/storage/driver/StorageDriver.java index 78c59dde..c636e33b 100644 --- a/engine/core/spt-base/src/main/java/com/dell/spt/base/storage/driver/StorageDriver.java +++ b/engine/core/spt-base/src/main/java/com/dell/spt/base/storage/driver/StorageDriver.java @@ -88,29 +88,22 @@ default IntegrityTerminalException terminalFailure() { void adjustIoBuffers(final long avgTransferSize, final OpType opType); /** - * Enable the fast-recycle dispatch path. When active and the current - * active-op count is at or below {@code concurrencyThreshold}, the driver - * will re-submit a successfully completed simple operation directly on the - * I/O thread instead of returning it through the LoadGenerator recycle queue. - *

- * The default implementation is a no-op; subclasses that support the - * optimisation (e.g. Netty-based drivers) override this. + * Legacy compatibility hook. Direct driver resubmission bypasses the shared + * generator and can prevent undispatched input items from entering circulation, + * so this method no longer enables an optimization. * - * @param concurrencyThreshold maximum active-op count at which the - * fast-recycle path is used (0 = disabled) + * @param concurrencyThreshold ignored + * @deprecated always a no-op; completed operations must return through the load generator */ + @Deprecated default void enableFastRecycle(final int concurrencyThreshold) {} /** - * Signal that the fast-recycle path is expected to handle most operations - * and the dispatch/generator VTs may quiesce (park on long waits). Only - * called when the configured concurrency is low enough that fast-recycle - * dominates; at higher concurrency the normal pipeline needs to stay - * responsive. - *

- * Default is a no-op; cooperative drivers override to inform their - * dispatch task. + * Legacy compatibility hook for the removed direct fast-recycle path. + * + * @deprecated always a no-op; generator and dispatch tasks retain their normal lifecycle */ + @Deprecated default void enableFastRecycleQuiesce() {} @Override diff --git a/engine/core/spt-base/src/test/java/com/dell/spt/base/item/op/OperationImplTest.java b/engine/core/spt-base/src/test/java/com/dell/spt/base/item/op/OperationImplTest.java index aa90a15d..10832be1 100644 --- a/engine/core/spt-base/src/test/java/com/dell/spt/base/item/op/OperationImplTest.java +++ b/engine/core/spt-base/src/test/java/com/dell/spt/base/item/op/OperationImplTest.java @@ -142,36 +142,4 @@ void latencyNeverReturnsEpochTimestamp() { "latency must not be epoch-scale in normal lifecycle"); } - // ---------- driverRecycled flag tests ---------- - - @Test - void driverRecycled_defaultIsFalse() { - final var op = new OperationImpl<>(0, OpType.READ, new ItemImpl("item"), null, null, null); - assertFalse(op.driverRecycled(), "driverRecycled should default to false"); - } - - @Test - void driverRecycled_setAndGet() { - final var op = new OperationImpl<>(0, OpType.READ, new ItemImpl("item"), null, null, null); - op.driverRecycled(true); - assertTrue(op.driverRecycled(), "driverRecycled should be true after setting"); - op.driverRecycled(false); - assertFalse(op.driverRecycled(), "driverRecycled should be false after clearing"); - } - - @Test - void driverRecycled_copiedByResult() { - final var op = new OperationImpl<>(0, OpType.READ, new ItemImpl("item"), null, "/path", null); - op.driverRecycled(true); - final var copy = op.result(); - assertTrue(copy.driverRecycled(), "driverRecycled should be preserved in result copy"); - } - - @Test - void driverRecycled_clearedByReset() { - final var op = new OperationImpl<>(0, OpType.READ, new ItemImpl("item"), null, "/path", null); - op.driverRecycled(true); - op.reset(); - assertFalse(op.driverRecycled(), "driverRecycled should be cleared by reset()"); - } } diff --git a/engine/core/spt-base/src/test/java/com/dell/spt/base/item/op/OperationResultCopyTest.java b/engine/core/spt-base/src/test/java/com/dell/spt/base/item/op/OperationResultCopyTest.java index 5aa621ce..3fd8da1c 100644 --- a/engine/core/spt-base/src/test/java/com/dell/spt/base/item/op/OperationResultCopyTest.java +++ b/engine/core/spt-base/src/test/java/com/dell/spt/base/item/op/OperationResultCopyTest.java @@ -8,7 +8,7 @@ /** * Tests that {@link OperationImpl#result()} produces a truly independent copy - * whose timing, status, and flag fields survive mutation of the original. + * whose timing and status fields survive mutation of the original. * These guarantees are critical because the completion path calls * {@code op.result()} to snapshot metrics, then the original op may be * recycled/reset/re-submitted. @@ -66,34 +66,10 @@ void resultCopy_preservesAllTimingFields() { assertEquals(op.respTimeDone(), copy.respTimeDone(), "respTimeDone must be copied"); } - @Test - void resultCopy_preservesDriverRecycledFlag() { - final var op = newTimedOp(); - op.driverRecycled(true); - op.finishResponse(); - - final var copy = op.result(); - - assertTrue(copy.driverRecycled(), - "driverRecycled flag must be preserved in the copy"); - } - - @Test - void resultCopy_driverRecycledDefaultsFalse() { - final var op = newTimedOp(); - op.finishResponse(); - - final var copy = op.result(); - - assertFalse(copy.driverRecycled(), - "driverRecycled should default to false in the copy"); - } - @Test void resetAfterResultCopy_doesNotAffectCopy() { final var op = newTimedOp(); op.finishResponse(); - op.driverRecycled(true); final long origDuration = op.duration(); final long origReqTimeStart = op.reqTimeStart(); @@ -108,16 +84,12 @@ void resetAfterResultCopy_doesNotAffectCopy() { "copy duration must survive original reset"); assertEquals(origReqTimeStart, copy.reqTimeStart(), "copy reqTimeStart must survive original reset"); - assertTrue(copy.driverRecycled(), - "copy driverRecycled must survive original reset"); // Verify original was actually reset assertEquals(Operation.Status.PENDING, op.status(), "original should be PENDING after reset"); assertEquals(0, op.reqTimeStart(), "original timing should be zeroed after reset"); - assertFalse(op.driverRecycled(), - "original driverRecycled should be false after reset"); } @Test @@ -226,8 +198,8 @@ void resetAfterIncrementOpRetryCount_survivesReset() { @Test void resetOpRetryCount_zeroesTheCounter() { - // Finding: without an explicit reset hook, a recycled (read-loop or Netty - // fast-recycle) operation that failed and retried once before eventually succeeding + // Finding: without an explicit reset hook, a recycled read-loop operation that failed + // and retried once before eventually succeeding // would keep an elevated opRetryCount forever across every future successful cycle, // eventually exhausting its retry budget from unrelated, non-consecutive failures. final var op = newTimedOp(); diff --git a/engine/core/spt-base/src/test/java/com/dell/spt/base/load/generator/LoadGeneratorImplRecycleTest.java b/engine/core/spt-base/src/test/java/com/dell/spt/base/load/generator/LoadGeneratorImplRecycleTest.java index 60f0994d..fa95f014 100644 --- a/engine/core/spt-base/src/test/java/com/dell/spt/base/load/generator/LoadGeneratorImplRecycleTest.java +++ b/engine/core/spt-base/src/test/java/com/dell/spt/base/load/generator/LoadGeneratorImplRecycleTest.java @@ -17,7 +17,6 @@ import com.dell.spt.base.storage.driver.StorageDriver; import com.github.akurilov.commons.io.Input; import com.github.akurilov.commons.io.Output; -import java.io.EOFException; import java.io.IOException; import java.util.ArrayList; import java.util.List; @@ -33,7 +32,7 @@ * Characterization tests for the recycle-poll path in {@link LoadGeneratorImpl#doWork()}. *

* These tests document the current behavior of the spin-wait, yield, and output - * logic in the recycle branch, plus the fast-recycle quiesce behavior (park/unpark). + * logic in the shared recycle-queue branch. */ @SuppressWarnings("unchecked") class LoadGeneratorImplRecycleTest { @@ -146,78 +145,6 @@ void batchRecycleOutputsAllOps() throws Exception { assertEquals(BATCH_SIZE, generator.generatedOpCount()); } - @Test - void initialManifestCirculatesCompletelyBeforeAnyRecycledOperation() throws Exception { - final int itemCount = 10; - final List sourceItems = new ArrayList<>(); - for (int i = 0; i < itemCount; i++) { - sourceItems.add(new DataItemImpl("manifest-" + i, 0, 1024)); - } - final AtomicInteger nextItem = new AtomicInteger(); - doAnswer(invocation -> { - @SuppressWarnings("unchecked") - final List buffer = invocation.getArgument(0); - final int limit = invocation.getArgument(1); - final int from = nextItem.get(); - if (from >= sourceItems.size()) { - throw new EOFException("end of manifest"); - } - final int to = Math.min(from + limit, sourceItems.size()); - buffer.addAll(sourceItems.subList(from, to)); - nextItem.set(to); - return to - from; - }).when(itemInput).get(anyList(), anyInt()); - doAnswer(invocation -> { - @SuppressWarnings("unchecked") - final List items = invocation.getArgument(0); - @SuppressWarnings("unchecked") - final List> operations = invocation.getArgument(1); - for (final DataItem item : items) { - operations.add(newOp(item.name())); - } - return null; - }).when(opsBuilder).buildOps(anyList(), anyList()); - - final LoadGeneratorImpl> circulatingGenerator = new LoadGeneratorImpl<>( - itemInput, opsBuilder, List.of(), output, BATCH_SIZE, - 0, 1000, true, false); - try { - for (int batch = 0; batch < 3; batch++) { - final int from = output.received.size(); - circulatingGenerator.doWork(); - final int to = output.received.size(); - for (int i = from; i < to; i++) { - circulatingGenerator.recycle(output.received.get(i)); - } - } - assertEquals(itemCount, output.received.size()); - assertEquals( - sourceItems.stream().map(DataItem::name).toList(), - output.received.stream().map(op -> op.item().name()).toList()); - - circulatingGenerator.doWork(); - assertTrue(circulatingGenerator.isItemInputFinished()); - final List expectedNames = sourceItems.stream().map(DataItem::name).toList(); - for (int cycle = 0; cycle < 2; cycle++) { - final int from = output.received.size(); - for (int batch = 0; batch < 3; batch++) { - circulatingGenerator.doWork(); - } - final int to = output.received.size(); - assertEquals(itemCount, to - from); - assertEquals( - expectedNames, - output.received.subList(from, to) - .stream().map(op -> op.item().name()).toList()); - for (int i = from; i < to; i++) { - circulatingGenerator.recycle(output.received.get(i)); - } - } - } finally { - circulatingGenerator.close(); - } - } - /** * Start the generator VT loop, recycle ops from the test thread, * and verify all ops eventually reach the output. @@ -287,125 +214,6 @@ void recycleQueueStateTracking() throws Exception { assertEquals(3, output.received.size()); } - // --- fast-recycle quiesce tests --- - - /** - * With quiesce enabled and empty recycle queue, doWork() parks for up to 10ms - * instead of yielding. Verify it returns within a bounded time. - */ - @Test - void quiesceParksOnEmptyRecycleQueue() throws Exception { - generator.enableFastRecycleQuiesce(); - // Exhaust item input - generator.doWork(); - assertTrue(generator.isItemInputFinished()); - - // doWork with quiesce + empty queue parks for up to 10ms - final long start = System.nanoTime(); - generator.doWork(); - final long elapsedMs = (System.nanoTime() - start) / 1_000_000; - - // Should park for ~10ms (timeout), bounded well under 100ms - assertTrue(elapsedMs < 100, "doWork took " + elapsedMs + "ms; expected < 100ms"); - assertEquals(0, output.received.size()); - } - - /** - * With quiesce enabled, recycle() unparks the generator VT so it picks - * up the op quickly rather than waiting the full 10ms timeout. - */ - @Test - void recycleUnparksQuiescedGenerator() throws Exception { - final int opCount = 10; - final ConcurrentCollectingOutput> concurrentOutput = new ConcurrentCollectingOutput<>(opCount); - - final LoadGeneratorImpl> quiescedGen = new LoadGeneratorImpl<>( - itemInput, - opsBuilder, - List.of(), - concurrentOutput, - BATCH_SIZE, - 0, - 1000, - true, - false); - quiescedGen.enableFastRecycleQuiesce(); - - try { - quiescedGen.start(); - assertEventually(quiescedGen::isItemInputFinished, 2000); - - // Recycle ops with slight spacing to exercise the unpark path - final long t0 = System.nanoTime(); - for (int i = 0; i < opCount; i++) { - quiescedGen.recycle(newOp("quiesce-" + i)); - Thread.sleep(1); - } - - // All ops should arrive well under 10ms * opCount since unpark wakes - // the generator immediately for each one - assertTrue( - concurrentOutput.latch.await(5, TimeUnit.SECONDS), - "Expected " + opCount + " ops but got " + concurrentOutput.received.size()); - final long elapsedMs = (System.nanoTime() - t0) / 1_000_000; - assertTrue( - elapsedMs < 2000, - "Took " + elapsedMs + "ms for " + opCount + " ops; unpark may not be working"); - assertEquals(opCount, concurrentOutput.received.size()); - } finally { - quiescedGen.stop(); - quiescedGen.await(5, TimeUnit.SECONDS); - quiescedGen.close(); - } - } - - /** - * Without quiesce, the existing yield path is used (characterization of - * the non-quiesce path when fast-recycle is not enabled). - */ - @Test - void nonQuiesceUsesYieldPath() throws Exception { - // quiesce NOT enabled — default - generator.doWork(); // exhaust item input - - // doWork with empty queue should return very quickly (yield, not 10ms park) - final long start = System.nanoTime(); - generator.doWork(); - final long elapsedMs = (System.nanoTime() - start) / 1_000_000; - - // Yield returns in microseconds, not milliseconds - assertTrue(elapsedMs < 5, "doWork took " + elapsedMs + "ms; yield should be sub-millisecond"); - } - - /** - * Verify the generatorParked flag guards against spurious unpark calls. - * When quiesce is enabled but the generator is NOT parked (e.g. actively - * processing ops), recycle() should still enqueue ops and they should be - * picked up via the normal polling path in doWork(). - */ - @Test - void recycleWithQuiesceEnabledButGeneratorNotParked() throws Exception { - // Enable quiesce, but we'll call doWork() manually so the generator - // VT is never actually parked (doWork returns after one spin). - generator.enableFastRecycleQuiesce(); - generator.doWork(); // exhaust item input - - // generatorParked should be false since we're not in the park path - final var parkedField = LoadGeneratorImpl.class.getDeclaredField("generatorParked"); - parkedField.setAccessible(true); - assertFalse(parkedField.getBoolean(generator), - "generatorParked should be false when not in park path"); - - // Recycle an op while the generator is not parked - final DataOperation op = newOp("not-parked-1"); - generator.recycle(op); - - // doWork should still pick up the op via polling - generator.doWork(); - assertEquals(1, output.received.size()); - assertSame(op, output.received.get(0)); - } - @Test void terminalSingleOutputIsStickyAndNeverResubmitted() throws Exception { final var failure = new IntegrityTerminalException( diff --git a/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/LoadStepContextImplTest.java b/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/LoadStepContextImplTest.java index 67d49f47..a6f65eb9 100644 --- a/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/LoadStepContextImplTest.java +++ b/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/LoadStepContextImplTest.java @@ -66,7 +66,6 @@ import org.junit.jupiter.api.io.TempDir; import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.after; import static org.mockito.Mockito.doNothing; @@ -134,28 +133,6 @@ public void loadStepRunTest() throws IOException { Assertions.assertTrue(stepCtx.isDone()); } - @org.junit.jupiter.params.ParameterizedTest - @org.junit.jupiter.params.provider.ValueSource(ints = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10 - }) - public void recycleWorkloadUsesSharedGeneratorCirculation(final int concurrency) { - testConfig.val("load-op-retry", false); - testConfig.val("load-op-recycle-mode", true); - - @SuppressWarnings("unchecked") - final LoadGenerator> generatorMock = mock(LoadGenerator.class); - @SuppressWarnings("unchecked") - final StorageDriver> driverMock = mock(StorageDriver.class); - when(driverMock.concurrencyLimit()).thenReturn(concurrency); - - new LoadStepContextImpl<>( - "recycle-through-generator", generatorMock, driverMock, null, - testConfig.configVal("load"), false); - - verify(driverMock, never()).enableFastRecycle(anyInt()); - verify(driverMock, never()).enableFastRecycleQuiesce(); - verify(generatorMock, never()).enableFastRecycleQuiesce(); - } - @Test public void putSingleOperationSuccessAndMetricsOutput() throws Exception { // enable recycle to populate latestSuccOpResultByItem map diff --git a/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/RecycleCirculationIntegrationTest.java b/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/RecycleCirculationIntegrationTest.java new file mode 100644 index 00000000..d6fdb203 --- /dev/null +++ b/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/RecycleCirculationIntegrationTest.java @@ -0,0 +1,390 @@ +package com.dell.spt.base.load.step.local.context; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.dell.spt.base.concurrent.AsyncRunnableBase; +import com.dell.spt.base.config.TestConfigBuilder; +import com.dell.spt.base.item.DataItem; +import com.dell.spt.base.item.DataItemImpl; +import com.dell.spt.base.item.ItemFactory; +import com.dell.spt.base.item.ItemType; +import com.dell.spt.base.item.op.OpType; +import com.dell.spt.base.item.op.Operation; +import com.dell.spt.base.item.op.data.DataOperation; +import com.dell.spt.base.load.generator.LoadGenerator; +import com.dell.spt.base.load.generator.LoadGeneratorBuilder; +import com.dell.spt.base.load.generator.LoadGeneratorBuilderImpl; +import com.dell.spt.base.load.generator.LoadGeneratorImpl; +import com.dell.spt.base.metrics.context.MetricsContext; +import com.dell.spt.base.metrics.context.MetricsContextImpl; +import com.dell.spt.base.metrics.snapshot.AllMetricsSnapshot; +import com.dell.spt.base.storage.driver.ListOptions; +import com.dell.spt.base.storage.driver.StorageDriver; +import com.github.akurilov.commons.io.Input; +import com.github.akurilov.commons.io.Output; +import com.github.akurilov.commons.system.SizeInBytes; +import com.github.akurilov.confuse.Config; +import java.io.EOFException; +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +/** End-to-end in-process coverage for duration-style recycled workload circulation. */ +@SuppressWarnings({"deprecation", "unchecked" +}) +class RecycleCirculationIntegrationTest { + + private static final int MANIFEST_ITEM_COUNT = 50; + private static final int COMPLETION_COUNT = 200; + private static final long ITEM_SIZE = 64; + private static final int COMPLETION_TIMEOUT_SECONDS = 10; + + /** + * Proves the product invariant at every low-concurrency shape affected by the former + * fast-recycle optimization: all manifest items enter circulation before any item is + * repeated. The test driver models the old public fast-recycle contract so replaying + * this test against v5.14.1 activates the broken direct-resubmission path and fails. + */ + @ParameterizedTest + @ValueSource(ints = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10 + }) + void completeManifestCirculatesBeforeAnyRepeat(final int concurrency) throws Exception { + final Config config = TestConfigBuilder.config(); + config.val("item-type", "data"); + config.val("item-data-ranges-concat", null); + config.val("load-op-type", "read"); + config.val("load-op-retry", false); + config.val("load-op-recycle-mode", true); + config.val("load-op-recycle-content-update", false); + config.val("load-op-limit-count", COMPLETION_COUNT); + config.val("load-op-wait-finish", true); + config.val("load-op-wait-limit", COMPLETION_TIMEOUT_SECONDS); + + final var itemInput = new ManifestItemInput(MANIFEST_ITEM_COUNT); + final var driver = new CirculationCanaryDriver(concurrency, COMPLETION_COUNT); + final ItemType itemType = ItemType.DATA; + final ItemFactory itemFactory = (ItemFactory) ItemType.getItemFactory(itemType); + final LoadGeneratorBuilder, LoadGeneratorImpl>> generatorBuilder = new LoadGeneratorBuilderImpl, LoadGeneratorImpl>>() + .itemConfig(config.configVal("item")) + .loadConfig(config.configVal("load")) + .itemType(itemType) + .itemFactory(itemFactory) + .itemInput(itemInput) + .loadOperationsOutput(driver) + .authConfig(config.configVal("storage").configVal("auth")) + .originIndex(0); + final LoadGenerator> generator = generatorBuilder.build(); + final MetricsContext metrics = buildMetrics(concurrency); + final var stepContext = new LoadStepContextImpl<>( + "recycle-circulation-t" + concurrency, + generator, + driver, + metrics, + config.configVal("load"), + false); + + stepContext.start(); + try { + assertTrue( + driver.awaitCompletions(COMPLETION_TIMEOUT_SECONDS, TimeUnit.SECONDS), + "timed out after " + driver.completedOpCount() + " completions at T" + concurrency); + + final List observed = driver.observedItemNames(); + assertEquals(COMPLETION_COUNT, observed.size()); + assertEquals( + MANIFEST_ITEM_COUNT, + new HashSet<>(observed.subList(0, MANIFEST_ITEM_COUNT)).size(), + "an item repeated before the complete manifest entered circulation at T" + concurrency); + + final Set expectedNames = ManifestItemInput.expectedNames(MANIFEST_ITEM_COUNT); + assertEquals(expectedNames, new HashSet<>(observed), "the driver saw an unexpected or missing item"); + final Map counts = new HashMap<>(); + observed.forEach(name -> counts.merge(name, 1, Integer::sum)); + final int minCount = counts.values().stream().mapToInt(Integer::intValue).min().orElseThrow(); + final int maxCount = counts.values().stream().mapToInt(Integer::intValue).max().orElseThrow(); + assertTrue( + maxCount - minCount <= concurrency, + "circulation imbalance exceeds the in-flight boundary at T" + concurrency + ": " + counts); + + assertEquals(COMPLETION_COUNT, driver.scheduledOpCount()); + assertEquals(COMPLETION_COUNT, driver.completedOpCount()); + assertEquals(0, driver.activeOpCount(), "all bounded concurrency permits must be released"); + } finally { + stepContext.stop(); + stepContext.shutdown(); + stepContext.close(); + } + } + + private static MetricsContext buildMetrics(final int concurrency) { + final MetricsContext metrics = MetricsContextImpl.builder() + .loadStepId("recycle-circulation-t" + concurrency) + .opType(OpType.READ) + .actualConcurrencyGauge(() -> 0) + .concurrencyLimit(concurrency) + .concurrencyThreshold(0) + .itemDataSize(new SizeInBytes(ITEM_SIZE)) + .outputPeriodSec(1) + .stdOutColorFlag(false) + .runId(0) + .build(); + metrics.start(); + return metrics; + } + + private static final class ManifestItemInput implements Input { + private final int itemCount; + private int nextIndex; + + ManifestItemInput(final int itemCount) { + this.itemCount = itemCount; + } + + static Set expectedNames(final int itemCount) { + final Set names = new HashSet<>(); + for (int i = 0; i < itemCount; i++) { + names.add(itemName(i)); + } + return names; + } + + private static String itemName(final int index) { + return String.format("manifest-item-%03d", index); + } + + @Override + public DataItem get() { + return nextIndex < itemCount ? new DataItemImpl(itemName(nextIndex++), 0, ITEM_SIZE) : null; + } + + @Override + public int get(final List buffer, final int limit) { + if (nextIndex >= itemCount) { + com.github.akurilov.commons.lang.Exceptions.throwUnchecked(new EOFException()); + } + final int start = nextIndex; + while (nextIndex < itemCount && nextIndex - start < limit) { + buffer.add(new DataItemImpl(itemName(nextIndex++), 0, ITEM_SIZE)); + } + return nextIndex - start; + } + + @Override + public long skip(final long itemsCount) { + final long skipped = Math.min(itemsCount, itemCount - nextIndex); + nextIndex += (int) skipped; + return skipped; + } + + @Override + public void reset() { + nextIndex = 0; + } + + @Override + public void close() {} + + @Override + public String toString() { + return "ManifestItemInput"; + } + } + + /** Protocol-free asynchronous driver with a bounded, observable concurrency contract. */ + private static final class CirculationCanaryDriver extends AsyncRunnableBase + implements StorageDriver> { + private final int concurrency; + private final int completionLimit; + private final Semaphore permits; + private final ExecutorService executor; + private final AtomicInteger scheduled = new AtomicInteger(); + private final AtomicInteger completed = new AtomicInteger(); + private final CountDownLatch completionLatch; + private final List observedNames = new CopyOnWriteArrayList<>(); + private volatile Output> resultOutput; + private volatile boolean legacyDirectRecycle; + + CirculationCanaryDriver(final int concurrency, final int completionLimit) { + this.concurrency = concurrency; + this.completionLimit = completionLimit; + this.permits = new Semaphore(concurrency, true); + this.executor = Executors.newFixedThreadPool(concurrency); + this.completionLatch = new CountDownLatch(completionLimit); + } + + @Override + public boolean put(final Operation op) { + if (!isStarted() || !permits.tryAcquire()) { + return false; + } + if (!claimCompletion()) { + permits.release(); + return false; + } + recordDispatch(op); + executor.execute(() -> complete(op)); + return true; + } + + @Override + public int put(final List> ops, final int from, final int to) { + int i = from; + while (i < to && put(ops.get(i))) { + i++; + } + return i - from; + } + + @Override + public int put(final List> ops) { + return put(ops, 0, ops.size()); + } + + private boolean claimCompletion() { + int current; + do { + current = scheduled.get(); + if (current >= completionLimit) { + return false; + } + } while (!scheduled.compareAndSet(current, current + 1)); + return true; + } + + private void recordDispatch(final Operation op) { + final String itemName = op.item().name(); + observedNames.add(itemName.startsWith("/") ? itemName.substring(1) : itemName); + } + + private void complete(final Operation op) { + op.reset(); + op.startRequest(); + op.finishRequest(); + op.startResponse(); + if (op instanceof DataOperation dataOperation) { + dataOperation.startDataResponse(); + try { + dataOperation.countBytesDone(op.item().size()); + } catch (final IOException e) { + throw new AssertionError(e); + } + } + op.finishResponse(); + op.status(Operation.Status.SUCC); + completed.incrementAndGet(); + + if (legacyDirectRecycle) { + op.driverRecycled(true); + resultOutput.put(op.result()); + if (claimCompletion()) { + recordDispatch(op); + executor.execute(() -> complete(op)); + } else { + permits.release(); + } + } else { + permits.release(); + resultOutput.put(op.result()); + } + completionLatch.countDown(); + } + + @Override + public void operationResultOutput(final Output> resultOutput) { + this.resultOutput = resultOutput; + } + + @Override + public List list( + final ItemFactory itemFactory, + final String path, + final String prefix, + final int idRadix, + final DataItem lastPrevItem, + final int count) { + return List.of(); + } + + @Override + public List list( + final ItemFactory itemFactory, + final String path, + final String prefix, + final int idRadix, + final DataItem lastPrevItem, + final int count, + final ListOptions options) { + return List.of(); + } + + @Override + public Input> getInput() { + throw new AssertionError(); + } + + @Override + public int concurrencyLimit() { + return concurrency; + } + + @Override + public int activeOpCount() { + return concurrency - permits.availablePermits(); + } + + @Override + public long scheduledOpCount() { + return scheduled.get(); + } + + @Override + public long completedOpCount() { + return completed.get(); + } + + @Override + public boolean isIdle() { + return activeOpCount() == 0; + } + + @Override + public void adjustIoBuffers(final long avgTransferSize, final OpType opType) {} + + @Override + public void enableFastRecycle(final int concurrencyThreshold) { + legacyDirectRecycle = concurrencyThreshold > 0; + } + + boolean awaitCompletions(final long timeout, final TimeUnit unit) throws InterruptedException { + return completionLatch.await(timeout, unit); + } + + List observedItemNames() { + return new ArrayList<>(observedNames); + } + + @Override + protected void doShutdown() { + executor.shutdown(); + } + + @Override + protected void doClose() { + executor.shutdownNow(); + } + } +} diff --git a/engine/extensions/storage-drivers/primitives/coop/src/main/java/com/dell/spt/storage/driver/coop/CoopStorageDriverBase.java b/engine/extensions/storage-drivers/primitives/coop/src/main/java/com/dell/spt/storage/driver/coop/CoopStorageDriverBase.java index 8ada2260..01a14203 100644 --- a/engine/extensions/storage-drivers/primitives/coop/src/main/java/com/dell/spt/storage/driver/coop/CoopStorageDriverBase.java +++ b/engine/extensions/storage-drivers/primitives/coop/src/main/java/com/dell/spt/storage/driver/coop/CoopStorageDriverBase.java @@ -53,8 +53,9 @@ public abstract class CoopStorageDriverBase 0; - } - - @Override - public void enableFastRecycleQuiesce() { - this.fastRecycleQuiesceActive = true; - Loggers.MSG.info("{}: fast-recycle quiesce active (dispatch task will extend idle wait)", toString()); + return false; } - /** - * Returns {@code true} when quiesce mode is active — i.e. the configured - * concurrency is low enough that fast-recycle handles most operations and - * the dispatch/generator VTs may park on long waits. - */ + /** @deprecated always returns {@code false}; direct fast-recycle quiescing was removed */ + @Deprecated protected boolean isFastRecycleQuiesceActive() { - return fastRecycleQuiesceActive; + return false; } /** - * Check whether the given completed operation is eligible for the fast-recycle - * short-circuit. Returns {@code true} only when: - *

    - *
  • fast-recycle has been enabled (threshold > 0)
    • - *
  • the current active-op count is ≤ the threshold
    • - *
  • the op finished successfully
    • - *
  • the op is a simple (non-composite, non-partial) operation
    • - *
  • the driver is still running
    • - *
+ * @param op ignored + * @deprecated always returns {@code false}; completed operations use shared circulation */ + @Deprecated protected boolean isFastRecycleEligible(final O op) { - final int threshold = fastRecycleConcurrencyThreshold; - return threshold > 0 - && activeOpCount() <= threshold - && op.status() == Operation.Status.SUCC - && !(op instanceof CompositeOperation) - && !(op instanceof PartialOperation) - && isStarted(); + return false; } + @Override + @Deprecated + public void enableFastRecycle(final int concurrencyThreshold) {} + + @Override + @Deprecated + public void enableFastRecycleQuiesce() {} + @Override protected void doShutdown() { opDispatchTask.stop(); diff --git a/engine/extensions/storage-drivers/primitives/coop/src/test/java/com/dell/spt/storage/driver/coop/CoopStorageDriverBaseTest.java b/engine/extensions/storage-drivers/primitives/coop/src/test/java/com/dell/spt/storage/driver/coop/CoopStorageDriverBaseTest.java index 052690f7..9262f7a9 100644 --- a/engine/extensions/storage-drivers/primitives/coop/src/test/java/com/dell/spt/storage/driver/coop/CoopStorageDriverBaseTest.java +++ b/engine/extensions/storage-drivers/primitives/coop/src/test/java/com/dell/spt/storage/driver/coop/CoopStorageDriverBaseTest.java @@ -10,7 +10,6 @@ import com.dell.spt.base.item.op.OpType; import com.dell.spt.base.item.op.Operation; import com.dell.spt.base.item.op.composite.data.CompositeDataOperationImpl; -import com.dell.spt.base.item.op.partial.PartialOperation; import com.dell.spt.base.item.op.partial.data.PartialDataOperationImpl; import com.dell.spt.base.logging.Loggers; import com.dell.spt.base.storage.driver.StorageDriverBase; @@ -426,13 +425,13 @@ void isIdleWhenAllPermitsFree() throws Exception { assertTrue(driver.isIdle(), "should be idle after release"); } - // ---------- Simple-op completion path (characterization for fast-recycle) ---------- + // ---------- Simple-op completion path ---------- @Test void handleCompleted_simpleSuccessOpSendsResultToOutput() throws Exception { final var driver = newRetryTestDriver(); - // A simple (non-composite, non-partial) op — this is the path fast-recycle targets + // A simple (non-composite, non-partial) operation final Operation op = mock(Operation.class); final Operation resultCopy = mock(Operation.class); when(op.status()).thenReturn(Operation.Status.SUCC); @@ -642,160 +641,6 @@ void handleCompleted_reEnqueuesParentWhenAllPartsDone() throws Exception { assertTrue(queue.contains(parent), "parent should be re-enqueued when all parts done"); } - // ---------- Fast-recycle eligibility tests ---------- - - @Test - void enableFastRecycle_setsThreshold() throws Exception { - final var driver = newFastRecycleDriver(4); - final var threshField = CoopStorageDriverBase.class.getDeclaredField("fastRecycleConcurrencyThreshold"); - threshField.setAccessible(true); - assertEquals(4, threshField.getInt(driver), "threshold should be set by enableFastRecycle"); - } - - @Test - void isFastRecycleEligible_trueForSimpleSuccessUnderThreshold() throws Exception { - final var driver = newFastRecycleDriver(4); - - final Operation op = mock(Operation.class); - when(op.status()).thenReturn(Operation.Status.SUCC); - - assertTrue(driver.isFastRecycleEligible(op), - "simple SUCC op under threshold should be eligible"); - } - - @Test - void isFastRecycleEligible_falseWhenDisabled() throws Exception { - // threshold = 0 means disabled - final var driver = newFastRecycleDriver(0); - - final Operation op = mock(Operation.class); - when(op.status()).thenReturn(Operation.Status.SUCC); - - assertFalse(driver.isFastRecycleEligible(op), - "should not be eligible when fast-recycle is disabled"); - } - - @Test - void isFastRecycleEligible_falseForFailedOp() throws Exception { - final var driver = newFastRecycleDriver(4); - - final Operation op = mock(Operation.class); - when(op.status()).thenReturn(Operation.Status.FAIL_IO); - - assertFalse(driver.isFastRecycleEligible(op), - "failed op should not be eligible for fast-recycle"); - } - - @Test - void isFastRecycleEligible_falseForCompositeOp() throws Exception { - final var driver = newFastRecycleDriver(4); - - final var baseItem = new DataItemImpl("obj", 0, 4096); - baseItem.dataInput(DataInput.instance(null, "7a42d9c483244167", new SizeInBytes("64KB"), 4, false, 0.0, true)); - final var compositeOp = new CompositeDataOperationImpl( - 0, OpType.CREATE, baseItem, null, "/bucket", null, null, 0, 1024); - compositeOp.status(Operation.Status.SUCC); - - assertFalse(driver.isFastRecycleEligible((Operation) (Operation) compositeOp), - "composite op should not be eligible for fast-recycle"); - } - - @Test - void isFastRecycleEligible_falseForPartialOp() throws Exception { - final var driver = newFastRecycleDriver(4); - - final PartialOperation partialOp = mock(PartialOperation.class); - when(partialOp.status()).thenReturn(Operation.Status.SUCC); - - assertFalse(driver.isFastRecycleEligible((Operation) partialOp), - "partial op should not be eligible for fast-recycle"); - } - - @Test - void isFastRecycleEligible_falseWhenActiveCountExceedsThreshold() throws Exception { - final var driver = newFastRecycleDriver(2); - - // Acquire 3 permits to simulate 3 active ops (threshold=2) - final var semField = CoopStorageDriverBase.class.getDeclaredField("concurrencyThrottle"); - semField.setAccessible(true); - final Semaphore sem = (Semaphore) semField.get(driver); - sem.acquire(3); - - final Operation op = mock(Operation.class); - when(op.status()).thenReturn(Operation.Status.SUCC); - - assertFalse(driver.isFastRecycleEligible(op), - "should not be eligible when active count exceeds threshold"); - - sem.release(3); - } - - @Test - void isFastRecycleEligible_trueWhenActiveCountEqualsThreshold() throws Exception { - final var driver = newFastRecycleDriver(2); - - // Acquire exactly 2 permits (threshold=2) - final var semField = CoopStorageDriverBase.class.getDeclaredField("concurrencyThrottle"); - semField.setAccessible(true); - final Semaphore sem = (Semaphore) semField.get(driver); - sem.acquire(2); - - final Operation op = mock(Operation.class); - when(op.status()).thenReturn(Operation.Status.SUCC); - - assertTrue(driver.isFastRecycleEligible(op), - "should be eligible when active count equals threshold (boundary)"); - - sem.release(2); - } - - // ---------- Fast-recycle quiesce tests ---------- - - @Test - void enableFastRecycleQuiesce_activatesQuiesceState() throws Exception { - final var driver = newFastRecycleDriver(4); - - assertFalse(driver.isFastRecycleQuiesceActive(), - "quiesce should be inactive by default"); - - driver.enableFastRecycleQuiesce(); - - assertTrue(driver.isFastRecycleQuiesceActive(), - "quiesce should be active after enableFastRecycleQuiesce()"); - } - - @Test - void quiesceInactiveByDefault() throws Exception { - final var driver = newFastRecycleDriver(4); - assertFalse(driver.isFastRecycleQuiesceActive(), - "quiesce must be inactive when only enableFastRecycle was called"); - } - - /** Set up a driver with fast-recycle infrastructure for eligibility tests. */ - private CoopStorageDriverBase> newFastRecycleDriver(int threshold) throws Exception { - final var driver = mock(CoopStorageDriverBase.class, withSettings().defaultAnswer(CALLS_REAL_METHODS)); - - // concurrencyLimit - final var limitField = CoopStorageDriverBase.class.getSuperclass().getDeclaredField("concurrencyLimit"); - limitField.setAccessible(true); - limitField.set(driver, 8); - - // concurrencyThrottle - final var semField = CoopStorageDriverBase.class.getDeclaredField("concurrencyThrottle"); - semField.setAccessible(true); - semField.set(driver, new Semaphore(8, true)); - - // fastRecycleConcurrencyThreshold - final var threshField = CoopStorageDriverBase.class.getDeclaredField("fastRecycleConcurrencyThreshold"); - threshField.setAccessible(true); - threshField.set(driver, threshold); - - // Mark as started so isStarted() returns true - when(driver.isStarted()).thenReturn(true); - - return driver; - } - // ---------- Output-full side-effect tests ---------- @Test diff --git a/engine/extensions/storage-drivers/primitives/coop/src/test/java/com/dell/spt/storage/driver/coop/OperationDispatchTaskTest.java b/engine/extensions/storage-drivers/primitives/coop/src/test/java/com/dell/spt/storage/driver/coop/OperationDispatchTaskTest.java index 1f8f3378..636423ca 100644 --- a/engine/extensions/storage-drivers/primitives/coop/src/test/java/com/dell/spt/storage/driver/coop/OperationDispatchTaskTest.java +++ b/engine/extensions/storage-drivers/primitives/coop/src/test/java/com/dell/spt/storage/driver/coop/OperationDispatchTaskTest.java @@ -351,8 +351,7 @@ void backpressureRecoveryViaSignal() throws Exception { @Test void untimedAwaitWakesOnSignal() throws Exception { // The dispatch task uses untimed await() — verify it still wakes - // promptly when signaled, regardless of fast-recycle quiesce state. - when(driverMock.isFastRecycleQuiesceActive()).thenReturn(true); + // promptly when signaled. final Operation op = mock(Operation.class); when(driverMock.submit(any(Operation.class))).thenReturn(true); diff --git a/engine/extensions/storage-drivers/primitives/netty/src/main/java/com/dell/spt/storage/driver/coop/netty/NettyStorageDriverBase.java b/engine/extensions/storage-drivers/primitives/netty/src/main/java/com/dell/spt/storage/driver/coop/netty/NettyStorageDriverBase.java index 41299718..268be85e 100755 --- a/engine/extensions/storage-drivers/primitives/netty/src/main/java/com/dell/spt/storage/driver/coop/netty/NettyStorageDriverBase.java +++ b/engine/extensions/storage-drivers/primitives/netty/src/main/java/com/dell/spt/storage/driver/coop/netty/NettyStorageDriverBase.java @@ -683,48 +683,8 @@ public void complete(final Channel channel, final O op) { channel.close(); } - // Fast-recycle path: keep the concurrency permit, release the channel, - // report metrics, then directly prepare + re-submit the original op. - // This avoids the VirtualThread scheduling overhead of the normal - // LoadGenerator recycleQueue → OperationDispatchTask path. - if (channel != null && isFastRecycleEligible(op)) { - if (!channel.attr(ATTR_KEY_RELEASED).getAndSet(Boolean.TRUE)) { - connPool.release(channel); - } - // isFastRecycleEligible() requires status == SUCC, so this is always a - // successful completion: clear load-op-retry's counter on the *original* op - // here too, same as LoadStepContextImpl.put() does for the result copy it - // sees - handleCompleted() below only hands that copy a snapshot, it never - // touches this live object, which is what actually gets reused below. - op.resetOpRetryCount(); - // Mark BEFORE handleCompleted so the result copy carries the flag - op.driverRecycled(true); - handleCompleted(op); - // Prepare and re-submit directly (we still hold the concurrency permit) - prepare(op); - try { - final Channel conn = leaseActiveConnection(); - conn.attr(ATTR_KEY_OPERATION).set(op); - op.nodeAddr(conn.attr(ATTR_KEY_NODE).get()); - op.startRequest(); - sendRequest(conn, op); - } catch (final ConnectException e) { - LogUtil.exception(Level.WARN, e, "Fast-recycle: failed to lease connection"); - op.status(Operation.Status.FAIL_IO); - concurrencyThrottle.release(); - handleCompleted(op); - } catch (final Throwable thrown) { - throwUncheckedIfInterrupted(thrown); - LogUtil.exception(Level.WARN, thrown, "Fast-recycle: failed to re-submit"); - op.status(Operation.Status.FAIL_UNKNOWN); - concurrencyThrottle.release(); - handleCompleted(op); - } - return; - } - - // Normal path: release permit + channel, then let the LoadGenerator - // recycle queue handle re-dispatch. + // Release the permit and channel before reporting completion. Recycled + // operations return through the shared LoadGenerator queue for redispatch. if (channel != null && !channel.attr(ATTR_KEY_RELEASED).getAndSet(Boolean.TRUE)) { concurrencyThrottle.release(); connPool.release(channel); diff --git a/engine/extensions/storage-drivers/primitives/netty/src/test/java/com/dell/spt/storage/driver/coop/netty/NettyCompletionPathTest.java b/engine/extensions/storage-drivers/primitives/netty/src/test/java/com/dell/spt/storage/driver/coop/netty/NettyCompletionPathTest.java index 6ae9c6ad..f673063c 100644 --- a/engine/extensions/storage-drivers/primitives/netty/src/test/java/com/dell/spt/storage/driver/coop/netty/NettyCompletionPathTest.java +++ b/engine/extensions/storage-drivers/primitives/netty/src/test/java/com/dell/spt/storage/driver/coop/netty/NettyCompletionPathTest.java @@ -31,8 +31,8 @@ /** * Characterization tests for {@link NettyStorageDriverBase#complete(io.netty.channel.Channel, - * Operation)}. These capture the current completion path behavior as a safety net before - * introducing fast-recycle dispatch. + * Operation)}. These capture the normal completion path and guard against deprecated + * fast-recycle hooks bypassing it. */ @SuppressWarnings("unchecked") class NettyCompletionPathTest { @@ -282,83 +282,11 @@ void complete_incrementsCompletedCount() { channel.close(); } - // ---------- Fast-recycle path tests ---------- - - private void enableFastRecycle(int threshold) throws Exception { - final var threshField = CoopStorageDriverBase.class.getDeclaredField("fastRecycleConcurrencyThreshold"); - threshField.setAccessible(true); - threshField.set(driver, threshold); - when(driver.isStarted()).thenReturn(true); - } - - @Test - void fastRecycle_setsDriverRecycledFlagOnOp() throws Exception { - enableFastRecycle(4); - // Use higher concurrency so we have room - final var limitField = StorageDriverBase.class.getDeclaredField("concurrencyLimit"); - limitField.setAccessible(true); - limitField.set(driver, 4); - final var sem = new Semaphore(4, true); - sem.acquire(1); // 1 active op (under threshold=4) - final var semField = CoopStorageDriverBase.class.getDeclaredField("concurrencyThrottle"); - semField.setAccessible(true); - semField.set(driver, sem); - - final var channel = newChannelWithReleasedFlag(); - final Operation op = mock(Operation.class); - final Operation resultCopy = mock(Operation.class); - when(op.status()).thenReturn(Operation.Status.SUCC); - when(op.result()).thenReturn(resultCopy); - - // Mock connPool.lease() to return a new channel for re-submit - final var resubmitChannel = newResubmitChannel(); - when(connPool.lease()).thenReturn(resubmitChannel); - - driver.complete(channel, op); - - // The op should have driverRecycled set to true before result() is called - verify(op).driverRecycled(true); - channel.close(); - resubmitChannel.close(); - } - - @Test - void fastRecycle_keepsPermitAndReleasesChannel() throws Exception { - enableFastRecycle(4); - final var limitField = StorageDriverBase.class.getDeclaredField("concurrencyLimit"); - limitField.setAccessible(true); - limitField.set(driver, 4); - final var sem = new Semaphore(4, true); - sem.acquire(1); - final var semField = CoopStorageDriverBase.class.getDeclaredField("concurrencyThrottle"); - semField.setAccessible(true); - semField.set(driver, sem); - - final var channel = newChannelWithReleasedFlag(); - final Operation op = mock(Operation.class); - when(op.status()).thenReturn(Operation.Status.SUCC); - when(op.result()).thenReturn(mock(Operation.class)); - - // Mock connPool.lease() for re-submit - final var resubmitChannel = newResubmitChannel(); - when(connPool.lease()).thenReturn(resubmitChannel); - - driver.complete(channel, op); - - // Channel should be released to the pool - verify(connPool).release(channel); - // Permit should NOT be released — it's held for the re-submitted op. - // Before complete: 3 available (4 total - 1 acquired). - // After fast-recycle: still 3 available (permit retained for new submit). - assertEquals(3, sem.availablePermits(), - "permit should be retained for the re-submitted op"); - channel.close(); - resubmitChannel.close(); - } - @Test - void fastRecycle_resetsOpForResubmit() throws Exception { - enableFastRecycle(4); + @SuppressWarnings("deprecation") + void deprecatedFastRecycleHookCannotBypassNormalCompletion() throws Exception { + // Compatibility calls from an extension must remain inert. + driver.enableFastRecycle(4); final var limitField = StorageDriverBase.class.getDeclaredField("concurrencyLimit"); limitField.setAccessible(true); limitField.set(driver, 4); @@ -373,102 +301,10 @@ void fastRecycle_resetsOpForResubmit() throws Exception { when(op.status()).thenReturn(Operation.Status.SUCC); when(op.result()).thenReturn(mock(Operation.class)); - final var resubmitChannel = newResubmitChannel(); - when(connPool.lease()).thenReturn(resubmitChannel); - driver.complete(channel, op); - // prepare() calls op.reset() internally — verify reset was called - verify(op).reset(); - // Also verify the op was re-submitted (startRequest called on the re-prepared op) - verify(op, atLeast(1)).startRequest(); - channel.close(); - resubmitChannel.close(); - } - - @Test - void fastRecycleClearsReleasedChannelAndRebindsOperationToResubmissionChannel() throws Exception { - enableFastRecycle(4); - final var limitField = StorageDriverBase.class.getDeclaredField("concurrencyLimit"); - limitField.setAccessible(true); - limitField.set(driver, 4); - final var sem = new Semaphore(4, true); - sem.acquire(1); - final var semField = CoopStorageDriverBase.class.getDeclaredField("concurrencyThrottle"); - semField.setAccessible(true); - semField.set(driver, sem); - - final var releasedChannel = newChannelWithReleasedFlag(); - final Operation op = mock(Operation.class); - when(op.status()).thenReturn(Operation.Status.SUCC); - when(op.result()).thenReturn(mock(Operation.class)); - releasedChannel.attr(NettyStorageDriver.ATTR_KEY_OPERATION).set(op); - - final var resubmitChannel = newResubmitChannel(); - when(connPool.lease()).thenReturn(resubmitChannel); - doAnswer(inv -> { - assertNull(releasedChannel.attr(NettyStorageDriver.ATTR_KEY_OPERATION).get()); - return null; - }).when(connPool).release(releasedChannel); - - driver.complete(releasedChannel, op); - - assertSame(op, resubmitChannel.attr(NettyStorageDriver.ATTR_KEY_OPERATION).get()); - releasedChannel.close(); - resubmitChannel.close(); - } - - @Test - void fastRecycle_fallsBackToNormalPathWhenNotEligible() throws Exception { - // Fast-recycle enabled but op failed — should use normal path - enableFastRecycle(4); - final var limitField = StorageDriverBase.class.getDeclaredField("concurrencyLimit"); - limitField.setAccessible(true); - limitField.set(driver, 4); - final var sem = new Semaphore(4, true); - sem.acquire(1); - final var semField = CoopStorageDriverBase.class.getDeclaredField("concurrencyThrottle"); - semField.setAccessible(true); - semField.set(driver, sem); - - final var channel = newChannelWithReleasedFlag(); - final Operation op = mock(Operation.class); - when(op.status()).thenReturn(Operation.Status.FAIL_IO); - when(op.result()).thenReturn(mock(Operation.class)); - - driver.complete(channel, op); - - // Normal path: permit released, no driverRecycled flag set - verify(op, never()).driverRecycled(anyBoolean()); - assertEquals(4, sem.availablePermits(), "permit should be released in normal path"); - channel.close(); - } - - @Test - void fastRecycle_releasesPermitOnLeaseFailure() throws Exception { - enableFastRecycle(4); - final var limitField = StorageDriverBase.class.getDeclaredField("concurrencyLimit"); - limitField.setAccessible(true); - limitField.set(driver, 4); - final var sem = new Semaphore(4, true); - sem.acquire(1); - final var semField = CoopStorageDriverBase.class.getDeclaredField("concurrencyThrottle"); - semField.setAccessible(true); - semField.set(driver, sem); - - final var channel = newChannelWithReleasedFlag(); - final Operation op = mock(Operation.class); - when(op.status()).thenReturn(Operation.Status.SUCC); - when(op.result()).thenReturn(mock(Operation.class)); - - // connPool.lease() throws — simulating no connections available - when(connPool.lease()).thenThrow(new java.net.ConnectException("no connections")); - - driver.complete(channel, op); - - // After lease failure, permit should be released - assertEquals(4, sem.availablePermits(), - "permit should be released after fast-recycle lease failure"); + verify(connPool, never()).lease(); + assertEquals(4, sem.availablePermits(), "deprecated hook must not retain the completion permit"); channel.close(); } @@ -574,50 +410,6 @@ void complete_mixedSuccessAndFailure_noPermitLeak() throws Exception { "all 3 ops should be counted as completed"); } - // ---------- Fast-recycle driverRecycled flag on result copy ---------- - - @Test - void fastRecycle_resultCopyCarriesDriverRecycledFlag() throws Exception { - enableFastRecycle(4); - final var limitField = StorageDriverBase.class.getDeclaredField("concurrencyLimit"); - limitField.setAccessible(true); - limitField.set(driver, 4); - final var sem = new Semaphore(4, true); - sem.acquire(1); - final var semField = CoopStorageDriverBase.class.getDeclaredField("concurrencyThrottle"); - semField.setAccessible(true); - semField.set(driver, sem); - - final var channel = newChannelWithReleasedFlag(); - - // Use a real OperationImpl so driverRecycled flag is actually stored - final var item = new ItemImpl("recycle-flag-obj"); - final OperationImpl realOp = new OperationImpl<>( - 0, OpType.CREATE, item, null, "/bucket", null); - realOp.startRequest(); - realOp.finishRequest(); - realOp.startResponse(); - realOp.status(Operation.Status.SUCC); - - // Capture the result copy to verify the flag - final List> capturedResults = new ArrayList<>(); - when(opResultOut.put(any(Operation.class))).thenAnswer(inv -> { - capturedResults.add(inv.getArgument(0)); - return true; - }); - - final var resubmitChannel = newResubmitChannel(); - when(connPool.lease()).thenReturn(resubmitChannel); - - driver.complete(channel, (Operation) (Operation) realOp); - - assertEquals(1, capturedResults.size(), "one result should be captured"); - assertTrue(capturedResults.get(0).driverRecycled(), - "result copy must carry driverRecycled=true from fast-recycle path"); - channel.close(); - resubmitChannel.close(); - } - // ---------- Channel release ordering ---------- @Test From fb082dd7464d0c92e922649feb3809ea590dbfdd Mon Sep 17 00:00:00 2001 From: Mike Horgan Date: Thu, 13 Aug 2026 16:44:57 -0400 Subject: [PATCH 4/4] strengthen recycle compatibility guards --- .../dell/spt/base/item/op/OperationImpl.java | 5 +- .../load/generator/LoadGeneratorImpl.java | 4 - .../RecycleCirculationIntegrationTest.java | 134 +++++++++++++----- .../driver/coop/CoopStorageDriverBase.java | 23 ++- .../coop/CoopStorageDriverBaseTest.java | 39 +++++ 5 files changed, 165 insertions(+), 40 deletions(-) diff --git a/engine/core/spt-base/src/main/java/com/dell/spt/base/item/op/OperationImpl.java b/engine/core/spt-base/src/main/java/com/dell/spt/base/item/op/OperationImpl.java index 795c144a..7cff2e21 100644 --- a/engine/core/spt-base/src/main/java/com/dell/spt/base/item/op/OperationImpl.java +++ b/engine/core/spt-base/src/main/java/com/dell/spt/base/item/op/OperationImpl.java @@ -25,7 +25,10 @@ public class OperationImpl implements Operation { protected volatile long reqTimeDone; protected volatile long respTimeStart; protected volatile long respTimeDone; - /** @deprecated inert compatibility field for subclasses compiled against the removed fast-recycle path */ + /** + * @deprecated retained for binary compatibility only; SPT never reads this field and writes + * have no effect + */ @Deprecated protected volatile boolean driverRecycled; protected volatile int opRetryCount; diff --git a/engine/core/spt-base/src/main/java/com/dell/spt/base/load/generator/LoadGeneratorImpl.java b/engine/core/spt-base/src/main/java/com/dell/spt/base/load/generator/LoadGeneratorImpl.java index 4a6c673c..96e60b23 100644 --- a/engine/core/spt-base/src/main/java/com/dell/spt/base/load/generator/LoadGeneratorImpl.java +++ b/engine/core/spt-base/src/main/java/com/dell/spt/base/load/generator/LoadGeneratorImpl.java @@ -537,10 +537,6 @@ public final List drainPendingRetries() { return drained; } - @Override - @Deprecated - public void enableFastRecycleQuiesce() {} - private boolean isFinished() { // Never actually finish while a load-op-retry redispatch is still waiting to be // drained - otherwise stop() below would tear this generator down before diff --git a/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/RecycleCirculationIntegrationTest.java b/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/RecycleCirculationIntegrationTest.java index d6fdb203..3eeeb3aa 100644 --- a/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/RecycleCirculationIntegrationTest.java +++ b/engine/core/spt-base/src/test/java/com/dell/spt/base/load/step/local/context/RecycleCirculationIntegrationTest.java @@ -1,6 +1,8 @@ package com.dell.spt.base.load.step.local.context; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import com.dell.spt.base.concurrent.AsyncRunnableBase; @@ -40,6 +42,7 @@ import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; @@ -63,6 +66,65 @@ class RecycleCirculationIntegrationTest { @ValueSource(ints = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10 }) void completeManifestCirculatesBeforeAnyRepeat(final int concurrency) throws Exception { + try (final var fixture = newFixture(concurrency)) { + fixture.start(); + final var driver = fixture.driver(); + assertTrue( + driver.awaitCompletions(COMPLETION_TIMEOUT_SECONDS, TimeUnit.SECONDS), + "timed out after " + driver.completedOpCount() + " completions at T" + concurrency); + + assertCirculationInvariant(driver, concurrency); + assertFalse( + driver.legacyDirectRecycleEnabled(), + "the production step context must not activate the legacy direct-recycle model"); + } + } + + @Test + void legacyDirectRecycleNegativeControlBreaksCirculationInvariant() throws Exception { + final int concurrency = 4; + try (final var fixture = newFixture(concurrency)) { + final var driver = fixture.driver(); + driver.enableLegacyDirectRecycleForNegativeControl(); + assertTrue(driver.legacyDirectRecycleEnabled(), "negative control must activate the legacy model"); + fixture.start(); + assertTrue( + driver.awaitCompletions(COMPLETION_TIMEOUT_SECONDS, TimeUnit.SECONDS), + "negative control timed out after " + driver.completedOpCount() + " completions"); + + final AssertionError failure = assertThrows( + AssertionError.class, () -> assertCirculationInvariant(driver, concurrency)); + assertTrue( + failure.getMessage().contains("repeated before the complete manifest"), + "negative control must fail the first-circulation assertion: " + failure.getMessage()); + } + } + + private static void assertCirculationInvariant( + final CirculationCanaryDriver driver, final int concurrency) { + final List observed = driver.observedItemNames(); + assertEquals(COMPLETION_COUNT, observed.size()); + assertEquals( + MANIFEST_ITEM_COUNT, + new HashSet<>(observed.subList(0, MANIFEST_ITEM_COUNT)).size(), + "an item repeated before the complete manifest entered circulation at T" + concurrency); + + final Set expectedNames = ManifestItemInput.expectedNames(MANIFEST_ITEM_COUNT); + assertEquals(expectedNames, new HashSet<>(observed), "the driver saw an unexpected or missing item"); + final Map counts = new HashMap<>(); + observed.forEach(name -> counts.merge(name, 1, Integer::sum)); + final int minCount = counts.values().stream().mapToInt(Integer::intValue).min().orElseThrow(); + final int maxCount = counts.values().stream().mapToInt(Integer::intValue).max().orElseThrow(); + assertTrue( + maxCount - minCount <= concurrency, + "circulation imbalance exceeds the in-flight boundary at T" + concurrency + ": " + counts); + + assertEquals(COMPLETION_COUNT, driver.scheduledOpCount()); + assertEquals(COMPLETION_COUNT, driver.completedOpCount()); + assertEquals(0, driver.activeOpCount(), "all bounded concurrency permits must be released"); + } + + private static CirculationFixture newFixture(final int concurrency) throws IOException { final Config config = TestConfigBuilder.config(); config.val("item-type", "data"); config.val("item-data-ranges-concat", null); @@ -96,38 +158,7 @@ void completeManifestCirculatesBeforeAnyRepeat(final int concurrency) throws Exc metrics, config.configVal("load"), false); - - stepContext.start(); - try { - assertTrue( - driver.awaitCompletions(COMPLETION_TIMEOUT_SECONDS, TimeUnit.SECONDS), - "timed out after " + driver.completedOpCount() + " completions at T" + concurrency); - - final List observed = driver.observedItemNames(); - assertEquals(COMPLETION_COUNT, observed.size()); - assertEquals( - MANIFEST_ITEM_COUNT, - new HashSet<>(observed.subList(0, MANIFEST_ITEM_COUNT)).size(), - "an item repeated before the complete manifest entered circulation at T" + concurrency); - - final Set expectedNames = ManifestItemInput.expectedNames(MANIFEST_ITEM_COUNT); - assertEquals(expectedNames, new HashSet<>(observed), "the driver saw an unexpected or missing item"); - final Map counts = new HashMap<>(); - observed.forEach(name -> counts.merge(name, 1, Integer::sum)); - final int minCount = counts.values().stream().mapToInt(Integer::intValue).min().orElseThrow(); - final int maxCount = counts.values().stream().mapToInt(Integer::intValue).max().orElseThrow(); - assertTrue( - maxCount - minCount <= concurrency, - "circulation imbalance exceeds the in-flight boundary at T" + concurrency + ": " + counts); - - assertEquals(COMPLETION_COUNT, driver.scheduledOpCount()); - assertEquals(COMPLETION_COUNT, driver.completedOpCount()); - assertEquals(0, driver.activeOpCount(), "all bounded concurrency permits must be released"); - } finally { - stepContext.stop(); - stepContext.shutdown(); - stepContext.close(); - } + return new CirculationFixture(driver, metrics, stepContext); } private static MetricsContext buildMetrics(final int concurrency) { @@ -146,6 +177,35 @@ private static MetricsContext buildMetrics(final int concurr return metrics; } + private static final class CirculationFixture implements AutoCloseable { + private final CirculationCanaryDriver driver; + private final MetricsContext metrics; + private final LoadStepContextImpl> stepContext; + + CirculationFixture( + final CirculationCanaryDriver driver, + final MetricsContext metrics, + final LoadStepContextImpl> stepContext) { + this.driver = driver; + this.metrics = metrics; + this.stepContext = stepContext; + } + + CirculationCanaryDriver driver() { + return driver; + } + + void start() { + stepContext.start(); + } + + @Override + public void close() throws IOException { + stepContext.close(); + metrics.close(); + } + } + private static final class ManifestItemInput implements Input { private final int itemCount; private int nextIndex; @@ -266,6 +326,8 @@ private boolean claimCompletion() { } private void recordDispatch(final Operation op) { + // Operation.buildItemPath() prefixes pathless data-item names with "/"; compare the + // storage object-key form used by the source manifest. final String itemName = op.item().name(); observedNames.add(itemName.startsWith("/") ? itemName.substring(1) : itemName); } @@ -369,6 +431,14 @@ public void enableFastRecycle(final int concurrencyThreshold) { legacyDirectRecycle = concurrencyThreshold > 0; } + void enableLegacyDirectRecycleForNegativeControl() { + legacyDirectRecycle = true; + } + + boolean legacyDirectRecycleEnabled() { + return legacyDirectRecycle; + } + boolean awaitCompletions(final long timeout, final TimeUnit unit) throws InterruptedException { return completionLatch.await(timeout, unit); } diff --git a/engine/extensions/storage-drivers/primitives/coop/src/main/java/com/dell/spt/storage/driver/coop/CoopStorageDriverBase.java b/engine/extensions/storage-drivers/primitives/coop/src/main/java/com/dell/spt/storage/driver/coop/CoopStorageDriverBase.java index 01a14203..7ce65cf0 100644 --- a/engine/extensions/storage-drivers/primitives/coop/src/main/java/com/dell/spt/storage/driver/coop/CoopStorageDriverBase.java +++ b/engine/extensions/storage-drivers/primitives/coop/src/main/java/com/dell/spt/storage/driver/coop/CoopStorageDriverBase.java @@ -53,9 +53,13 @@ public abstract class CoopStorageDriverBase>( + "deprecated-fast-recycle", + dataInput, + storageConfigForMultipartLimits(0, 0), + false, + 16)) { + logger.addAppender(appender); + logger.setLevel(Level.WARN); + + driver.enableFastRecycle(4); + driver.enableFastRecycle(8); + driver.enableFastRecycleQuiesce(); + awaitCapturedEvents(appender, 1, 2000); + + final var warningMessages = appender.events().stream() + .filter(e -> Level.WARN.equals(e.getLevel())) + .map(e -> e.getMessage().getFormattedMessage()) + .filter(msg -> msg.contains("deprecated fast-recycle request ignored")) + .toList(); + assertEquals(1, warningMessages.size(), "deprecated hooks should warn once per driver"); + assertTrue(warningMessages.get(0).contains("shared generator circulation")); + } finally { + logger.removeAppender(appender); + logger.setLevel(originalLevel); + appender.stop(); + } + } + // ---------- Simple-op completion path ---------- @Test