From 43219b2673b90533699a8fecaa8f7aa1c640aa09 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Sat, 4 Jul 2026 12:08:26 +0300 Subject: [PATCH] Fix race in TraditionalWorkQueue.isIdle() isIdle() checked that the operation queue is empty and that no worker thread is active, but a worker is invisible to both checks in the window between taking an operation from the queue and assigning it to its operation field. waitUntilIdle() could therefore return immediately while the just-submitted operation was still being handed off to a worker, which made TraditionalWorkQueueTestCase.testWaitUntilIdleSlowOpInProgress flaky. Track the number of accepted-but-not-fully-processed operations in a dedicated counter: incremented before an operation is enqueued (and rolled back if it is rejected), decremented when a worker thread is done with it, and adjusted for operations cancelled while the queue is being resized. isIdle() now simply checks that this counter is zero. Move the four waitUntilIdle tests out of the "slow" group into the default build. --- .../extensions/TraditionalWorkQueue.java | 54 +++++++++++-------- .../extensions/TraditionalWorkerThread.java | 11 +++- .../TraditionalWorkQueueTestCase.java | 9 ++-- 3 files changed, 48 insertions(+), 26 deletions(-) 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 {