From 45900e1548fdf2cd81b31a6e328a0925e6f23c13 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Thu, 09 Jul 2026 09:26:38 +0000
Subject: [PATCH] [#695] Fix race in TraditionalWorkQueue.isIdle() (#688)
---
opendj-server-legacy/src/main/java/org/opends/server/extensions/TraditionalWorkQueue.java | 54 ++++++++++++++++----------
opendj-server-legacy/src/main/java/org/opends/server/extensions/TraditionalWorkerThread.java | 11 +++++
opendj-server-legacy/src/test/java/org/opends/server/extensions/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 fa64931..0f8ffb9 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;
@@ -65,6 +67,15 @@
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 @@
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 @@
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 @@
}
opsSubmitted.incrementAndGet();
+ submitted = true;
}
finally
{
+ if (!submitted)
+ {
+ pendingOpsCount.decrementAndGet();
+ }
queueReadLock.unlock();
}
}
@@ -696,10 +718,12 @@
if (pendingOperation != null)
{
pendingOperation.abort(cancelRequest);
+ pendingOpsCount.decrementAndGet();
}
while ((pendingOperation = oldOpQueue.poll()) != null)
{
pendingOperation.abort(cancelRequest);
+ pendingOpsCount.decrementAndGet();
}
}
finally
@@ -714,28 +738,16 @@
@Override
public boolean isIdle()
{
- queueReadLock.lock();
- try
- {
- if (!opQueue.isEmpty())
- {
- return false;
- }
+ return pendingOpsCount.get() == 0;
+ }
- for (TraditionalWorkerThread t : workerThreads)
- {
- if (t.isActive())
- {
- return false;
- }
- }
-
- 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 1ceae70..3934f0e 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 @@
}
else
{
+ try
+ {
// The operation is not null, so process it. Make sure that when
// processing is complete.
@@ -172,6 +174,13 @@
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 7b5ae31..2df3c4a 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 @@
*
* @throws Exception If an unexpected problem occurs.
*/
- @Test(groups = { "slow" })
+ @Test
public void testWaitUntilIdleNoOpsInProgress()
throws Exception
{
@@ -145,7 +146,7 @@
*
* @throws Exception If an unexpected problem occurs.
*/
- @Test(groups = { "slow" }, timeOut=10000)
+ @Test(timeOut=10000)
public void testWaitUntilIdleNoOpsInProgressNoTimeout()
throws Exception
{
@@ -165,7 +166,7 @@
*
* @throws Exception If an unexpected problem occurs.
*/
- @Test(groups = { "slow" })
+ @Test
public void testWaitUntilIdleSlowOpInProgress()
throws Exception
{
@@ -189,7 +190,7 @@
*
* @throws Exception If an unexpected problem occurs.
*/
- @Test(groups = { "slow" })
+ @Test
public void testWaitUntilTimeoutWithIdleSlowOpInProgress()
throws Exception
{
--
Gitblit v1.10.0