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

[김민주]

Hi Daniel,

Thank you so much for your detailed insights in the previous email. I
completely agree with your point that ensuring atomicity between the token
retrieval (sequenceGenerator.getAndIncrement()) and placing it in the queue
(queue.put()) is crucial to avoid issues such as thread interleaving, which
could lead to tokens being inserted in an unpredictable order.

That being said, my goal with the original test was to simulate a situation
where multiple threads are competing to put items into a queue that’s
already full. In this case, I wanted the threads to block naturally when
the queue reaches capacity and then resume in the same order when space
becomes available. Essentially, I was trying to observe how threads would
be blocked in the order of their token retrieval and resume their
operations when the queue allows.

I now see that using a ReentrantLock to ensure atomicity across both the
token retrieval and the queue insertion can indeed prevent such
interleaving. However, a challenge arises when the queue’s capacity is
reached. Since LinkedBlockingQueue.put() causes the thread to block when
the queue is full, the thread holding the lock will not be able to release
it until space becomes available in the queue. This can result in a
deadlock-like situation where other threads are unable to acquire the lock
to retrieve their own tokens.

In other words, while the use of a lock ensures atomicity, it can
inadvertently cause issues when the queue is full. Specifically, the thread
holding the lock gets blocked while waiting for space in the queue and
cannot release the lock, which in turn prevents other threads from
acquiring the lock to retrieve their own tokens. This leads to a situation
where multiple threads are stuck waiting for both the lock and space in the
queue, causing a deadlock-like scenario.

This was one of the key reasons why I initially implemented the token
retrieval and queue insertion steps with non-locking.

In the original approach, I intended for each thread to call put, confirm
that it has entered the waiting state, and then allow the next thread to
put the next sequence value and also enter the waiting state. This approach
was designed to simulate a situation where the queue is already full and
each thread has to wait in line for space to become available.

2024년 9월 4일 (수) 오후 8:11, Daniel Fuchs <daniel.fuchs at oracle.com>님이 작성:

> Hi Kim,
>
> When several threads compete to put items in a queue,
> regardless of queue locking strategies, nothing guarantees
> the order in which items will end up in the queue if
> the action of obtaining the next element and putting
> it in the queue is not atomic. This is because a thread
> can be preempted by the scheduler at any time between
> two operations.
>
> Consider the following scenario:
>
> you have a sequence of ordered items a, b, c, one queue q,
> and two threads T1, T2 and T3:
>
> T1 starts, obtains element a, and gets preempted (the thread
>     is paused by the scheduler).
> T2 starts, obtains element b and puts it in the queue.
> T3 starts, obtains element c and gets preempted.
> T1 gets scheduled again and puts a in the queue.
> T3 gets scheduled again and puts c in the queue.
>
> result: the queue contains b a c; regardless of whether
> the queue uses fair or unfair locking, the order of items
> in the queue will be unpredictable. And in this scenario,
> there isn't even any contention on the queue.
>
> To ensure that items will end up in the queue in the proper
> order would require an additional lock at the application
> level.
>
> That is - you would need to use a lock so that taking
> an element from the sequenceGenerator and putting it in the
> queue becomes atomic (with respect to the queue).
>
> so in your example you would need something like:
>
> static ReentrantLock lock = new ReentrantLock();
> ...
>
>        Thread thread = new Thread(() -> {
>              lock.lock();
>              try {
>                  var token = sequenceGenerator.getAndIncrement();
>                  // even if this thread gets preempted here, no
>                  // other thread will be able to obtain the next
>                  // token before our token has been put in the queue
>                  try {
>                      queue.put(token);
>                  } catch (InterruptedException e) {
>                      // restore the sequence
>                      var restored = sequenceGenerator.decrementAndGet();
>                      assert token == restored;
>                  }
>              } finally {
>                  lock.unlock();
>              }
>
>          });
>          thread.start();
>
> best regards,
>
> -- daniel
>
> On 04/09/2024 08:09, 김민주 wrote:
> > // This configuration ensures items enter the queue's waiting list in a
> > predictable sequence (10, 11, 12, ...) // allowing us to verify if they
> > are later consumed in the same order for (int i = 0; i < PRODUCER_COUNT;
> > i++) { Thread thread = new Thread(() -> { try {
> > queue.put(sequenceGenerator.getAndIncrement()); } catch
> > (InterruptedException e) { // ignore } });
>
>
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <https://mail.openjdk.org/pipermail/core-libs-dev/attachments/20240904/5c504f4f/attachment-0001.htm>