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

[Daniel Fuchs]

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 } });