[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.cobbs at gmail.com>
Sent: Thursday, 5 September 2024 18:44
To: 김민주 <miiiinju00 at gmail.com>
Cc: Viktor Klang <viktor.klang at oracle.com>; Daniel FUCHS <daniel.fuchs at oracle.com>; core-libs-dev at openjdk.org <core-libs-dev at 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
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <https://mail.openjdk.org/pipermail/core-libs-dev/attachments/20240905/da8fc270/attachment-0001.htm>