Re: [External] : Re: [POTENTIAL BUG] Potential FIFO violation in BlockingQueue under high contention and suggestion for fair mode in ArrayBlockingQueue and LinkedBlockingQueue

[Viktor Klang]

Good point, Archie—I completely forgot to mention the fact that polling thread 
state doesn't necessarily ensure that the thread has been enqueued once the 
thread state is moved to blocking/waiting.

Thread state polling aside, for as long as Condition::await() is allowed to 
spuriously wake, FIFO just cannot be "guaranteed".

Cheers,
√


Viktor Klang
Software Architect, Java Platform Group
Oracle
________________________________
From: Archie Cobbs <archie.co...@gmail.com>
Sent: Thursday, 5 September 2024 18:44
To: 김민주 <miiiinj...@gmail.com>
Cc: Viktor Klang <viktor.kl...@oracle.com>; Daniel FUCHS 
<daniel.fu...@oracle.com>; core-libs-dev@openjdk.org <core-libs-dev@openjdk.org>
Subject: Re: [External] : Re: [POTENTIAL BUG] Potential FIFO violation in 
BlockingQueue under high contention and suggestion for fair mode in 
ArrayBlockingQueue and LinkedBlockingQueue

Hi Kim,

I think there may be an issue with your test. Specifically, this code is 
suspect:

    // Wait for the producer thread to enter BLOCKED state
    // This ensures that the thread is waiting on the full queue
    while (thread.getState() == State.RUNNABLE || thread.getState() == 
State.NEW);

I don't think that actually guarantees that the thread is blocked in put().

I was able to reproduce the bug using ArrayBlockingQueue(), but could no longer 
reproduce it after I changed the above code to this:

    // Wait for the producer thread to block in queue.put()
    // This ensures that the thread is waiting on the full queue
    while (!queue.isPutting(thread));

after also making this change to ArrayBlockingQueue.java:

--- a/src/java.base/share/classes/java/util/concurrent/ArrayBlockingQueue.java
+++ b/src/java.base/share/classes/java/util/concurrent/ArrayBlockingQueue.java
@@ -365,10 +365,31 @@ public void put(E e) throws InterruptedException {
         Objects.requireNonNull(e);
         final ReentrantLock lock = this.lock;
         lock.lockInterruptibly();
+        final boolean added = puttingThreads.add(Thread.currentThread());
         try {
             while (count == items.length)
                 notFull.await();
             enqueue(e);
+        } finally {
+            if (added)
+                puttingThreads.remove(Thread.currentThread());
+            lock.unlock();
+        }
+    }
+
+    private final java.util.HashSet<Thread> puttingThreads = new 
java.util.HashSet<>();
+
+    /**
+     * Test for putting thread.
+     *
+     * @param thread thread to test
+     * @return true if thread is a putting thread
+     */
+    public boolean isPutting(Thread thread) {
+        final ReentrantLock lock = this.lock;
+        lock.lock();
+        try {
+            return puttingThreads.contains(thread);
         } finally {
             lock.unlock();
         }

So the original test may not be correct due to Java memory model subtleties, 
etc.

-Archie

--
Archie L. Cobbs