diff --git a/opendj-server-legacy/src/main/java/org/opends/server/extensions/TraditionalWorkQueue.java b/opendj-server-legacy/src/main/java/org/opends/server/extensions/TraditionalWorkQueue.java index fa649314b4..0f8ffb9f9b 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/extensions/TraditionalWorkQueue.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/extensions/TraditionalWorkQueue.java @@ -13,6 +13,7 @@ * * Copyright 2006-2010 Sun Microsystems, Inc. * Portions Copyright 2013-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC */ package org.opends.server.extensions; @@ -23,6 +24,7 @@ import java.util.List; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock; @@ -64,6 +66,15 @@ public class TraditionalWorkQueue extends WorkQueue /** The number of operations that have been submitted to the work queue for processing. */ private AtomicLong opsSubmitted; + /** + * The number of operations that have been accepted by the work queue and are + * not yet fully processed: this covers operations sitting in the queue as + * well as operations currently handled by worker threads, including the + * hand-off window in between, which is invisible to both the queue and the + * worker thread activity flag. + */ + private final AtomicInteger pendingOpsCount = new AtomicInteger(); + /** * The number of times that an attempt to submit a new request has been * rejected because the work queue is already at its maximum capacity. @@ -139,6 +150,7 @@ public void initializeWorkQueue(TraditionalWorkQueueCfg configuration) killThreads = false; opsSubmitted = new AtomicLong(0); queueFullRejects = new AtomicLong(0); + pendingOpsCount.set(0); // Register to be notified of any configuration changes. configuration.addTraditionalChangeListener(this); @@ -302,6 +314,11 @@ public boolean trySubmitOperation(Operation operation) private void submitOperation(Operation operation, boolean blockEnqueuingWhenFull) throws DirectoryException { + // Count the operation before enqueuing it so that isIdle() can never + // observe an empty queue while the operation is in the process of being + // submitted; the count is rolled back if the operation is rejected. + pendingOpsCount.incrementAndGet(); + boolean submitted = false; queueReadLock.lock(); try { @@ -359,9 +376,14 @@ private void submitOperation(Operation operation, } opsSubmitted.incrementAndGet(); + submitted = true; } finally { + if (!submitted) + { + pendingOpsCount.decrementAndGet(); + } queueReadLock.unlock(); } } @@ -696,10 +718,12 @@ public ConfigChangeResult applyConfigurationChange( if (pendingOperation != null) { pendingOperation.abort(cancelRequest); + pendingOpsCount.decrementAndGet(); } while ((pendingOperation = oldOpQueue.poll()) != null) { pendingOperation.abort(cancelRequest); + pendingOpsCount.decrementAndGet(); } } finally @@ -714,28 +738,16 @@ public ConfigChangeResult applyConfigurationChange( @Override public boolean isIdle() { - queueReadLock.lock(); - try - { - if (!opQueue.isEmpty()) - { - return false; - } - - for (TraditionalWorkerThread t : workerThreads) - { - if (t.isActive()) - { - return false; - } - } + return pendingOpsCount.get() == 0; + } - return true; - } - finally - { - queueReadLock.unlock(); - } + /** + * Notifies this work queue that a worker thread is done handling an + * operation previously taken from the queue. + */ + void operationDone() + { + pendingOpsCount.decrementAndGet(); } /** diff --git a/opendj-server-legacy/src/main/java/org/opends/server/extensions/TraditionalWorkerThread.java b/opendj-server-legacy/src/main/java/org/opends/server/extensions/TraditionalWorkerThread.java index 1ceae70400..3934f0e03f 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/extensions/TraditionalWorkerThread.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/extensions/TraditionalWorkerThread.java @@ -13,7 +13,7 @@ * * Copyright 2006-2010 Sun Microsystems, Inc. * Portions Copyright 2011-2016 ForgeRock AS. - * Portions Copyright 2025 3A Systems, LLC. + * Portions Copyright 2025-2026 3A Systems, LLC. */ package org.opends.server.extensions; @@ -147,6 +147,8 @@ public void run() } else { + try + { // The operation is not null, so process it. Make sure that when // processing is complete. @@ -172,6 +174,13 @@ public void run() operation.setResultCode(ResultCode.SUCCESS); operation.getClientConnection().sendResponse(operation); } + } + finally + { + // Whatever the outcome, tell the queue this operation is no + // longer pending so that isIdle() stays accurate. + workQueue.operationDone(); + } } } catch (Throwable t) diff --git a/opendj-server-legacy/src/test/java/org/opends/server/extensions/TraditionalWorkQueueTestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/extensions/TraditionalWorkQueueTestCase.java index 7b5ae31a11..2df3c4a2a2 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/extensions/TraditionalWorkQueueTestCase.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/extensions/TraditionalWorkQueueTestCase.java @@ -13,6 +13,7 @@ * * Copyright 2006-2008 Sun Microsystems, Inc. * Portions Copyright 2013-2016 ForgeRock AS. + * Portions Copyright 2026 3A Systems, LLC */ package org.opends.server.extensions; @@ -125,7 +126,7 @@ public void testChangingNumWorkerThreads() * * @throws Exception If an unexpected problem occurs. */ - @Test(groups = { "slow" }) + @Test public void testWaitUntilIdleNoOpsInProgress() throws Exception { @@ -145,7 +146,7 @@ public void testWaitUntilIdleNoOpsInProgress() * * @throws Exception If an unexpected problem occurs. */ - @Test(groups = { "slow" }, timeOut=10000) + @Test(timeOut=10000) public void testWaitUntilIdleNoOpsInProgressNoTimeout() throws Exception { @@ -165,7 +166,7 @@ public void testWaitUntilIdleNoOpsInProgressNoTimeout() * * @throws Exception If an unexpected problem occurs. */ - @Test(groups = { "slow" }) + @Test public void testWaitUntilIdleSlowOpInProgress() throws Exception { @@ -189,7 +190,7 @@ public void testWaitUntilIdleSlowOpInProgress() * * @throws Exception If an unexpected problem occurs. */ - @Test(groups = { "slow" }) + @Test public void testWaitUntilTimeoutWithIdleSlowOpInProgress() throws Exception {