This is an automated email from the ASF dual-hosted git repository.
garydgregory pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/commons-lang.git
The following commit(s) were added to refs/heads/master by this push:
new e5f7f1b56 Add optional FIFO fairness to TimedSemaphore (LANG-1790)
(#1840)
e5f7f1b56 is described below
commit e5f7f1b56776fb2f6a77fc3a64bc47f7e849a425
Author: Gary Gregory <[email protected]>
AuthorDate: Sun Oct 25 13:57:36 2026 -0400
Add optional FIFO fairness to TimedSemaphore (LANG-1790) (#1840)
Add Builder.setFair(boolean) and queue blocking acquisitions in FIFO
order
when fairness is enabled. Preserve the default non-fair behavior and the
existing tryAcquire() API.
Keep permit accounting synchronized, remove queued acquisitions on
interruption or shutdown, and allow new unlimited acquisitions to
proceed
without waiting. Share the wait-and-shutdown check between acquisition
paths.
Document fairness guarantees and exceptions, and add regression coverage
for ordering, cancellation, limit changes, and builder reuse.
---
src/changes/changes.xml | 1 +
.../commons/lang3/concurrent/TimedSemaphore.java | 73 +++-
.../lang3/concurrent/TimedSemaphoreTest.java | 367 +++++++++++++++++++++
3 files changed, 433 insertions(+), 8 deletions(-)
diff --git a/src/changes/changes.xml b/src/changes/changes.xml
index c2079f682..a0c460ef9 100644
--- a/src/changes/changes.xml
+++ b/src/changes/changes.xml
@@ -78,6 +78,7 @@ The <action> type attribute can be add,update,fix,remove.
<action type="fix" dev="ggregory" due-to="alhuda, Gary
Gregory">Restrict LocaleUtils.toLocale to ASCII language, country, and numeric
area codes; previously accepted non-ASCII codes now cause
IllegalArgumentException (#1838).</action>
<action type="fix" dev="ggregory" due-to="Gary
Gregory">Deprecate ExceptionUtils.removeCommonFrames(), which is now used only
by tests. Root-cause stack trace generation compares StackTraceElement objects
directly.</action>
<!-- ADD -->
+ <action issue="LANG-1790" type="add" dev="ggregory">Add optional FIFO
ordering for TimedSemaphore blocking acquisitions.</action>
<!-- UPDATE -->
<action type="update" dev="ggregory" due-to="Gary Gregory">Bump
org.apache.commons:commons-parent from 105 to 106.</action>
</release>
diff --git
a/src/main/java/org/apache/commons/lang3/concurrent/TimedSemaphore.java
b/src/main/java/org/apache/commons/lang3/concurrent/TimedSemaphore.java
index 6567cf35f..232be52ff 100644
--- a/src/main/java/org/apache/commons/lang3/concurrent/TimedSemaphore.java
+++ b/src/main/java/org/apache/commons/lang3/concurrent/TimedSemaphore.java
@@ -17,6 +17,8 @@
package org.apache.commons.lang3.concurrent;
+import java.util.ArrayDeque;
+import java.util.Deque;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -123,6 +125,7 @@ public static class Builder implements
Supplier<TimedSemaphore> {
private long period;
private TimeUnit timeUnit;
private int limit;
+ private boolean fair;
/**
* Constructs a new Builder.
@@ -136,6 +139,24 @@ public TimedSemaphore get() {
return new TimedSemaphore(this);
}
+ /**
+ * Sets whether blocking acquisitions use FIFO ordering.
+ *
+ * <p>
+ * Fairness is disabled by default. When enabled, calls to {@link
TimedSemaphore#acquire()} are ordered by entry into the internal waiting queue,
+ * not by thread start time or method return order. New calls made
with the limit disabled and calls to {@link TimedSemaphore#tryAcquire()} may
+ * bypass waiting callers.
+ * </p>
+ *
+ * @param fair Whether to enable FIFO ordering.
+ * @return {@code this} instance.
+ * @since 3.22.0
+ */
+ public Builder setFair(final boolean fair) {
+ this.fair = fair;
+ return this;
+ }
+
/**
* Sets the limit.
*
@@ -235,8 +256,15 @@ public static Builder builder() {
/** A flag whether shutdown() was called. */
private boolean shutdown; // @GuardedBy("this")
+ /** Whether blocking acquisitions use FIFO ordering. */
+ private final boolean fair;
+
+ /** Tokens for fair acquisitions, guarded by this object's monitor. */
+ private final Deque<Object> waiters = new ArrayDeque<>();
+
private TimedSemaphore(final Builder builder) {
Validate.inclusiveBetween(1, Long.MAX_VALUE, builder.period, "Time
period must be greater than 0.");
+ fair = builder.fair;
period = builder.period;
unit = builder.timeUnit;
if (builder.service != null) {
@@ -287,21 +315,45 @@ public TimedSemaphore(final ScheduledExecutorService
service, final long timePer
* already been invoked, calling this method will cause an exception. The
very first call of this method starts the timer task which monitors the time
* period set for this {@link TimedSemaphore}. From now on the semaphore
is active.
*
+ * <p>
+ * When fairness is enabled, waiting callers acquire permits in the order
they enter the internal queue. This does not guarantee thread start or
+ * method return order. New calls made with the limit disabled do not wait
for queued callers. Calls to {@link #tryAcquire()} may also bypass
+ * queued callers.
+ * </p>
+ *
* @throws InterruptedException Thrown if the thread gets interrupted.
* @throws IllegalStateException Thrown if this semaphore is already shut
down.
*/
public synchronized void acquire() throws InterruptedException {
prepareAcquire();
- boolean canPass;
- do {
- canPass = acquirePermit();
- if (!canPass) {
- wait();
- if (shutdown) {
- throw new IllegalStateException("TimedSemaphore is shut
down.");
+ if (fair && getLimit() > NO_LIMIT) {
+ // A private token avoids invoking an overridden Thread.equals()
during cancellation.
+ final Object waiter = new Object();
+ waiters.addLast(waiter);
+ try {
+ while (waiters.peekFirst() != waiter || !acquirePermit()) {
+ wait();
+ if (shutdown) {
+ throw new IllegalStateException("TimedSemaphore is
shut down.");
+ }
}
+ } finally {
+ waiters.removeFirstOccurrence(waiter);
+ // Let the next caller use any permits remaining in this
period, including after interruption.
+ notifyAll();
}
- } while (!canPass);
+ } else {
+ boolean canPass;
+ do {
+ canPass = acquirePermit();
+ if (!canPass) {
+ wait();
+ if (shutdown) {
+ throw new IllegalStateException("TimedSemaphore is
shut down.");
+ }
+ }
+ } while (!canPass);
+ }
}
/**
@@ -475,6 +527,11 @@ protected ScheduledFuture<?> startTimer() {
* Tries to acquire a permit from this semaphore. If the limit of this
semaphore has not yet been reached, a permit is acquired, and this method
returns
* <strong>true</strong>. Otherwise, this method returns immediately with
the result <strong>false</strong>.
*
+ * <p>
+ * This method does not honor the fairness setting: it may acquire a
permit ahead of waiting callers. It does not test or clear the calling thread's
+ * interrupted status.
+ * </p>
+ *
* @return <strong>true</strong> if a permit could be acquired;
<strong>false</strong> otherwise.
* @throws IllegalStateException Thrown if this semaphore is already shut
down.
* @since 3.5
diff --git
a/src/test/java/org/apache/commons/lang3/concurrent/TimedSemaphoreTest.java
b/src/test/java/org/apache/commons/lang3/concurrent/TimedSemaphoreTest.java
index aab586fd7..88516ab91 100644
--- a/src/test/java/org/apache/commons/lang3/concurrent/TimedSemaphoreTest.java
+++ b/src/test/java/org/apache/commons/lang3/concurrent/TimedSemaphoreTest.java
@@ -19,13 +19,18 @@
import static
org.apache.commons.lang3.LangAssertions.assertIllegalArgumentException;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.time.Duration;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
@@ -41,6 +46,46 @@
*/
class TimedSemaphoreTest extends AbstractLangTest {
+ private static class FairAcquirer extends Thread {
+ private final TimedSemaphore semaphore;
+ private final BlockingQueue<Integer> order;
+ private final int id;
+ private volatile Exception failure;
+
+ FairAcquirer(final TimedSemaphore semaphore, final
BlockingQueue<Integer> order, final int id) {
+ this.semaphore = semaphore;
+ this.order = order;
+ this.id = id;
+ setDaemon(true);
+ }
+
+ @Override
+ public void run() {
+ try {
+ semaphore.acquire();
+ order.add(id);
+ } catch (final InterruptedException | IllegalStateException ex) {
+ failure = ex;
+ }
+ }
+ }
+
+ private static final class EqualFairAcquirer extends FairAcquirer {
+ EqualFairAcquirer(final TimedSemaphore semaphore, final
BlockingQueue<Integer> order, final int id) {
+ super(semaphore, order, id);
+ }
+
+ @Override
+ public boolean equals(final Object other) {
+ return other instanceof EqualFairAcquirer;
+ }
+
+ @Override
+ public int hashCode() {
+ return 1;
+ }
+ }
+
/**
* A test thread class that will be used by tests for triggering the
* semaphore. The thread calls the semaphore a configurable number of
times.
@@ -196,6 +241,54 @@ public void run() {
/** Constant for the default limit. */
private static final int LIMIT = 10;
+ private static void assertAcquireOrder(final TimedSemaphore semaphore,
final boolean fair) throws Exception {
+ final BlockingQueue<Integer> order = new LinkedBlockingQueue<>();
+ final FairAcquirer first = new FairAcquirer(semaphore, order, 1);
+ final FutureTask<Void> newcomerTask = new FutureTask<>(() -> {
+ synchronized (semaphore) {
+ semaphore.endOfPeriod();
+ // Retain the monitor so the older caller cannot run before
this acquisition attempt.
+ semaphore.acquire();
+ order.add(2);
+ }
+ return null;
+ });
+ final Thread newcomer = new Thread(newcomerTask);
+ newcomer.setDaemon(true);
+ try {
+ semaphore.acquire();
+ first.start();
+ awaitWaiting(first);
+ newcomer.start();
+ assertEquals(Integer.valueOf(fair ? 1 : 2), order.poll(5,
TimeUnit.SECONDS));
+ assertEquals(1, semaphore.getAcquireCount());
+ semaphore.endOfPeriod();
+ newcomerTask.get(5, TimeUnit.SECONDS);
+ assertEquals(Integer.valueOf(fair ? 2 : 1), order.poll(5,
TimeUnit.SECONDS));
+ assertNull(first.failure);
+ } finally {
+ newcomer.interrupt();
+ try {
+ stopAcquirers(semaphore, first);
+ } finally {
+ newcomer.join(5000);
+ assertFalse(newcomer.isAlive());
+ }
+ }
+ }
+
+ private static void awaitWaiting(final Thread thread) throws
InterruptedException {
+ final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ while (thread.getState() != Thread.State.WAITING && thread.isAlive()
&& System.nanoTime() < deadline) {
+ Thread.sleep(1);
+ }
+ assertEquals(Thread.State.WAITING, thread.getState(), "Caller must be
queued before the test proceeds");
+ }
+
+ private static TimedSemaphore newFairSemaphore(final int limit) {
+ return
TimedSemaphore.builder().setPeriod(1).setTimeUnit(TimeUnit.DAYS).setLimit(limit).setFair(true).get();
+ }
+
/**
* Prepares an executor service mock to expect the start of the timer.
*
@@ -208,6 +301,15 @@ private void prepareStartTimer(final
ScheduledExecutorService service,
EasyMock.expectLastCall().andReturn(future);
}
+ private static void stopAcquirers(final TimedSemaphore semaphore, final
FairAcquirer... threads) throws InterruptedException {
+ semaphore.shutdown();
+ for (final FairAcquirer thread : threads) {
+ thread.interrupt();
+ thread.join(5000);
+ assertFalse(thread.isAlive(), "Acquirer did not terminate");
+ }
+ }
+
/**
* Tests the acquire() method if a limit is set.
*
@@ -315,6 +417,264 @@ void testAcquireNoLimit() throws InterruptedException {
EasyMock.verify(service, future);
}
+ @Test
+ void testBuilderReusePreservesFairness() throws Exception {
+ final TimedSemaphore.Builder builder =
TimedSemaphore.builder().setPeriod(1).setTimeUnit(TimeUnit.DAYS).setLimit(1).setFair(true);
+ final TimedSemaphore fairSemaphore = builder.get();
+ final TimedSemaphore nonFairSemaphore = builder.setFair(false).get();
+ try {
+ // Check each instance while the builder has the opposite setting.
+ assertAcquireOrder(fairSemaphore, true);
+ builder.setFair(true);
+ assertAcquireOrder(nonFairSemaphore, false);
+ } finally {
+ fairSemaphore.shutdown();
+ nonFairSemaphore.shutdown();
+ }
+ }
+
+ @Test
+ void testFairAcquireDisabledLimitDoesNotQueueNewCallers() throws
InterruptedException {
+ final TimedSemaphore semaphore = newFairSemaphore(1);
+ final BlockingQueue<Integer> order = new LinkedBlockingQueue<>();
+ final FairAcquirer waiting = new FairAcquirer(semaphore, order, 1);
+ final FairAcquirer newcomer = new FairAcquirer(semaphore, order, 2);
+ try {
+ semaphore.acquire();
+ waiting.start();
+ awaitWaiting(waiting);
+ semaphore.setLimit(0);
+ newcomer.start();
+ newcomer.join(5000);
+ assertFalse(newcomer.isAlive(), "Unlimited acquisitions must not
wait for the next period");
+ assertNull(newcomer.failure);
+ assertTrue(order.contains(2));
+ } finally {
+ stopAcquirers(semaphore, waiting, newcomer);
+ }
+ }
+
+ @Test
+ void testFairAcquireDisablingLimitReleasesWaitersAtRollover() throws
InterruptedException {
+ final TimedSemaphore semaphore = newFairSemaphore(1);
+ final BlockingQueue<Integer> order = new LinkedBlockingQueue<>();
+ final FairAcquirer first = new FairAcquirer(semaphore, order, 1);
+ final FairAcquirer second = new FairAcquirer(semaphore, order, 2);
+ try {
+ semaphore.acquire();
+ first.start();
+ awaitWaiting(first);
+ second.start();
+ awaitWaiting(second);
+ semaphore.setLimit(0);
+ semaphore.endOfPeriod();
+ assertNotNull(order.poll(5, TimeUnit.SECONDS));
+ assertNotNull(order.poll(5, TimeUnit.SECONDS));
+ assertNull(first.failure);
+ assertNull(second.failure);
+ assertEquals(2, semaphore.getAcquireCount());
+ } finally {
+ stopAcquirers(semaphore, first, second);
+ }
+ }
+
+ @Test
+ void testFairAcquireInterruptedHeadReleasesNextWaiter() throws
InterruptedException {
+ final TimedSemaphore semaphore = newFairSemaphore(1);
+ final BlockingQueue<Integer> order = new LinkedBlockingQueue<>();
+ final FairAcquirer first = new FairAcquirer(semaphore, order, 1);
+ final FairAcquirer second = new FairAcquirer(semaphore, order, 2);
+ try {
+ semaphore.acquire();
+ first.start();
+ awaitWaiting(first);
+ second.start();
+ awaitWaiting(second);
+ synchronized (semaphore) {
+ semaphore.setLimit(2);
+ first.interrupt();
+ }
+ first.join(5000);
+ assertFalse(first.isAlive());
+ assertInstanceOf(InterruptedException.class, first.failure);
+ assertEquals(Integer.valueOf(2), order.poll(5, TimeUnit.SECONDS));
+ assertNull(second.failure);
+ assertEquals(2, semaphore.getAcquireCount());
+ } finally {
+ stopAcquirers(semaphore, first, second);
+ }
+ }
+
+ @Test
+ void testFairAcquireInterruptedMiddlePreservesQueue() throws
InterruptedException {
+ final TimedSemaphore semaphore = newFairSemaphore(1);
+ final BlockingQueue<Integer> order = new LinkedBlockingQueue<>();
+ // Distinct callers must remain distinct even if their Thread
subclasses compare equal.
+ final FairAcquirer first = new EqualFairAcquirer(semaphore, order, 1);
+ final FairAcquirer second = new EqualFairAcquirer(semaphore, order, 2);
+ final FairAcquirer third = new EqualFairAcquirer(semaphore, order, 3);
+ try {
+ semaphore.acquire();
+ first.start();
+ awaitWaiting(first);
+ second.start();
+ awaitWaiting(second);
+ third.start();
+ awaitWaiting(third);
+ second.interrupt();
+ second.join(5000);
+ assertFalse(second.isAlive());
+ assertInstanceOf(InterruptedException.class, second.failure);
+ semaphore.endOfPeriod();
+ assertEquals(Integer.valueOf(1), order.poll(5, TimeUnit.SECONDS));
+ semaphore.endOfPeriod();
+ assertEquals(Integer.valueOf(3), order.poll(5, TimeUnit.SECONDS));
+ assertNull(first.failure);
+ assertNull(third.failure);
+ assertEquals(1, semaphore.getAcquireCount());
+ } finally {
+ stopAcquirers(semaphore, first, second, third);
+ }
+ }
+
+ @Test
+ void testFairAcquireNewcomerCannotBypassQueuedCaller() throws Exception {
+ assertAcquireOrder(newFairSemaphore(1), true);
+ }
+
+ @Test
+ void testFairAcquireOrderAcrossPeriods() throws InterruptedException {
+ final TimedSemaphore semaphore = newFairSemaphore(1);
+ final BlockingQueue<Integer> order = new LinkedBlockingQueue<>();
+ final FairAcquirer first = new FairAcquirer(semaphore, order, 1);
+ final FairAcquirer second = new FairAcquirer(semaphore, order, 2);
+ final FairAcquirer third = new FairAcquirer(semaphore, order, 3);
+ try {
+ semaphore.acquire();
+ first.start();
+ awaitWaiting(first);
+ second.start();
+ awaitWaiting(second);
+ third.start();
+ awaitWaiting(third);
+ for (int expected = 1; expected <= 3; expected++) {
+ semaphore.endOfPeriod();
+ assertEquals(Integer.valueOf(expected), order.poll(5,
TimeUnit.SECONDS));
+ assertEquals(1, semaphore.getAcquireCount());
+ assertEquals(0, semaphore.getAvailablePermits());
+ }
+ assertNull(first.failure);
+ assertNull(second.failure);
+ assertNull(third.failure);
+ } finally {
+ stopAcquirers(semaphore, first, second, third);
+ }
+ }
+
+ @Test
+ void testFairAcquireReleasesMultipleWaiters() throws InterruptedException {
+ final TimedSemaphore semaphore = newFairSemaphore(1);
+ final BlockingQueue<Integer> order = new LinkedBlockingQueue<>();
+ final FairAcquirer first = new FairAcquirer(semaphore, order, 1);
+ final FairAcquirer second = new FairAcquirer(semaphore, order, 2);
+ try {
+ semaphore.acquire();
+ first.start();
+ awaitWaiting(first);
+ second.start();
+ awaitWaiting(second);
+ semaphore.setLimit(2);
+ semaphore.endOfPeriod();
+ assertNotNull(order.poll(5, TimeUnit.SECONDS));
+ assertNotNull(order.poll(5, TimeUnit.SECONDS));
+ assertEquals(2, semaphore.getAcquireCount());
+ assertNull(first.failure);
+ assertNull(second.failure);
+ } finally {
+ stopAcquirers(semaphore, first, second);
+ }
+ }
+
+ @Test
+ void testFairAcquireShutdownReleasesWaiters() throws InterruptedException {
+ final TimedSemaphore semaphore = newFairSemaphore(1);
+ final BlockingQueue<Integer> order = new LinkedBlockingQueue<>();
+ final FairAcquirer first = new FairAcquirer(semaphore, order, 1);
+ final FairAcquirer second = new FairAcquirer(semaphore, order, 2);
+ try {
+ semaphore.acquire();
+ first.start();
+ awaitWaiting(first);
+ second.start();
+ awaitWaiting(second);
+ semaphore.shutdown();
+ first.join(5000);
+ second.join(5000);
+ assertFalse(first.isAlive());
+ assertFalse(second.isAlive());
+ assertInstanceOf(IllegalStateException.class, first.failure);
+ assertInstanceOf(IllegalStateException.class, second.failure);
+ assertTrue(order.isEmpty());
+ } finally {
+ stopAcquirers(semaphore, first, second);
+ }
+ }
+
+ @Test
+ void testFairLimitChangesPreserveAccounting() throws InterruptedException {
+ final TimedSemaphore semaphore = newFairSemaphore(3);
+ try {
+ semaphore.acquire();
+ assertTrue(semaphore.tryAcquire());
+ semaphore.setLimit(1);
+ assertFalse(semaphore.tryAcquire());
+ semaphore.setLimit(3);
+ assertEquals(1, semaphore.getAvailablePermits());
+ assertTrue(semaphore.tryAcquire());
+ semaphore.setLimit(3);
+ assertFalse(semaphore.tryAcquire());
+ semaphore.endOfPeriod();
+ assertEquals(3, semaphore.getLastAcquiresPerPeriod());
+ assertEquals(3, semaphore.getAvailablePermits());
+ assertEquals(0, semaphore.getAcquireCount());
+ semaphore.setLimit(-1);
+ for (int i = 0; i < 10; i++) {
+ semaphore.acquire();
+ }
+ assertEquals(10, semaphore.getAcquireCount());
+ } finally {
+ semaphore.shutdown();
+ }
+ }
+
+ @Test
+ void testFairTryAcquireCanBypassWaitersWithoutClearingInterrupt() throws
InterruptedException {
+ final TimedSemaphore semaphore = newFairSemaphore(1);
+ final BlockingQueue<Integer> order = new LinkedBlockingQueue<>();
+ final FairAcquirer waiter = new FairAcquirer(semaphore, order, 1);
+ try {
+ semaphore.acquire();
+ waiter.start();
+ awaitWaiting(waiter);
+ synchronized (semaphore) {
+ semaphore.endOfPeriod();
+ Thread.currentThread().interrupt();
+ try {
+ assertTrue(semaphore.tryAcquire());
+ assertTrue(Thread.currentThread().isInterrupted());
+ assertFalse(semaphore.tryAcquire());
+ } finally {
+ Thread.interrupted();
+ }
+ }
+ semaphore.endOfPeriod();
+ assertEquals(Integer.valueOf(1), order.poll(5, TimeUnit.SECONDS));
+ assertNull(waiter.failure);
+ } finally {
+ stopAcquirers(semaphore, waiter);
+ }
+ }
+
/**
* Tests whether the available non-blocking calls can be queried.
*
@@ -399,6 +759,12 @@ void testInitInvalidPeriod() {
assertIllegalArgumentException(() -> new TimedSemaphore(0L, UNIT,
LIMIT));
}
+ @Test
+ void testNonFairAcquireNewcomerCanBypassQueuedCaller() throws Exception {
+ final TimedSemaphore semaphore =
TimedSemaphore.builder().setPeriod(1).setTimeUnit(TimeUnit.DAYS).setLimit(1).setFair(false).get();
+ assertAcquireOrder(semaphore, false);
+ }
+
/**
* Tries to call acquire() after shutdown(). This should cause an
exception.
*/
@@ -588,4 +954,5 @@ void testTryAcquireAfterShutdown() {
semaphore.shutdown();
assertThrows(IllegalStateException.class, semaphore::tryAcquire);
}
+
}
