0

Recently I came across this question : There are say 3 consumer threads and need to implement a lock-free queue(can't use synchronization) so that no consuming thread is blocked. Assume that queue already contains the data.

I thought about it for a while and came across Atomic operations which if used carefully can help. My implementation is as shown below. As data is already there in queue I have not implemented the enqueue method and populating the array inside the constructor.

public class SPMCQueue {

    private AtomicInteger index = new AtomicInteger(0);

    public int[] arr;

    public SPMCQueue(int size) {
        arr = IntStream.range(0, size).toArray();
    }

    public Integer dequeue() {
        Integer ret = null;
        int x = index.getAndIncrement();
        if (x < arr.length) {
            ret = arr[x];
            System.out.println(arr[x] + " by " + Thread.currentThread().getName());
        }
        else {
            throw new RuntimeException("Queue is empty");
        }
        return ret;
    }
}

class QueueTest {
    public static void main(String[] args) {

        SPMCQueueq = new SPMCQueue(40);

        Runnable t1 = () -> {
            try {
            while (true) {
                q.dequeue();
            }
            }catch(Exception e) {

            }
        };

        Runnable t2 = () -> { 
            try {
            while(true) { q.dequeue(); }
            }catch(Exception e) {

            }
        };

        Runnable r3 = () -> { 

            try {
                while(true) { q.dequeue(); }
            } catch (Exception e) {
                // TODO Auto-generated catch block
                //e.printStackTrace();
            }

        };

        Thread thread1 = new Thread(t1);
        Thread thread2 = new Thread(t2);
        Thread thread3 = new Thread(r3);

        thread1.start();
        thread2.start();
        thread3.start();

    }
}

I have executed the above program and the result shows that the all 3 consumers are consuming the data albeit out of order and some threads are consuming more data than the other threads but I don't see any of the data appearing multiple times in the o/p.

I have the following questions:

  1. Is there any issue in the above implementation?

  2. What are the other ways to implement lock-free consumer queue?

Yug Singh
  • 3,112
  • 5
  • 27
  • 52
  • 2
    *"What are the other ways to implement lock-free consumer queue?"* See `java.util.concurrent.ConcurrentLinkedQueue` – Andreas Apr 30 '19 at 21:36
  • "no consuming thread is blocked" - in case the queue become empty, what do you expect the consumer thread to do? – Alexei Kaigorodov May 01 '19 at 16:42
  • @AlexeiKaigorodov assumption is queue has infinite data. Also, the problem is more about how to consume it in lock-free way without using synchronization. – Yug Singh May 01 '19 at 17:39
  • @YugSingh I mean, when producer is slow and consumers are fast, the queue becomes empty. What do you expect consumer should do in this case, if not to block? Sleep and poll? Or waste CPU cycles until next item is produced? – Alexei Kaigorodov May 02 '19 at 02:21

1 Answers1

0

I want to give my reply in tandem with answer: https://stackoverflow.com/a/21101904/7340499 since it is a similar question to yours.

So, your questions:

but I don't see any of the data appearing multiple times in the o/p.

This is expected. Because, getAndIncrement() is atomic, and anytime that function is accessed, you will get a different x value, hence a different output. However, due to combining "getAndIncrement()" and "System.out.print()" functions in a single non-atomic dequeue operation, your output may sometimes be out of order, e.g. you get x=1 on one thread, and another thread interrupts, gets x=2 and prints it, and then your initial thread finalizes printing. I believe this also points out the issues of your implementation, as asked in (1). Is your application okay with queue being processed out of order?

What are the other ways to implement lock-free consumer queue?

Well, atomic operations are one way, as you have noticed. But in essence, they are very much like locks, anyway. They just operate on a smaller scale (there are some implementation differences, though). So it is hard to conceptualize them as different methods, at least for myself. Other than that, I believe there are some nice resources over here: Lock Free Queue -- Single Producer, Multiple Consumers

  • In *this* use-case it's not super different from locks, but in cases where you don't need any atomic RMW in the read side, you have perfect scalability to any number of readers, with no contention between readers. (e.g. a simple SeqLock, or even wait-free readers with RCU). Even in this case, you can have basically no contention between a reader and a writer. Multiple readers have to contend with each other, though, to sort out which one gets which queue slot. – Peter Cordes May 01 '19 at 02:36
  • @PeterCordes Right, so if read isn't atomic, you can have as many readers on a RMW operation as you said. However, what I mean by atomic ~ locks similarity was that in order for an operation to be atomic, the CPU/OS/Runtime Env. (or whatever your environment is) has to make sure that the process is completed as if it was a single cycle instruction, much similar to many assembly instructions in x86 and/or ARM. Now, if the process isn't "truly" atomic, as far as I know, the only option left is to somehow limit any other process interfering with the current one, by using some sort of a lock. –  May 01 '19 at 13:19
  • @PeterCordes So saying something like, I'm not gonna use locks, I'm gonna use atomic operations instead, is very much like saying I'm gonna use locks but I call them atomic. Please correct me if I'm wrong, or if there are other ways generating lock free RMW or RCW operations, without any type of locking/synchronization mechanism. I wonder if there is one, because I don't follow much on x86 architecture anymore, and there are some very cool instructions there that supports many many cool stuff, so I may not know of it. –  May 01 '19 at 13:24
  • Read or write of a single register-width variable usually *is* atomic (and lock-free) without any involvement from the OS. e.g. [Why is integer assignment on a naturally aligned variable atomic on x86?](//stackoverflow.com/q/36624881). But basically the same applies to ARM `ldr` / `str` or PowerPC `lw`/`sw` or equivalent for any other ISA, for aligned words. For atomic RMW, x86 has `lock add [mem], eax` or whatever ([Can num++ be atomic for 'int num'?](//stackoverflow.com/q/39393850)), and most other ISAs have some form of LL/SC retry loop. – Peter Cordes May 01 '19 at 17:57
  • Atomic RMW is lock-free because it's impossible for a thread to sleep while holding a lock and block all other threads trying to access the shared variable. This is a fundamental *qualitative* difference. https://preshing.com/20130605/the-worlds-simplest-lock-free-hash-table/ shows an example of a fixed-size hash table (with add but not deletion) using lock-free atomic operations so writes can happen in parallel with readers and even other writers. – Peter Cordes May 01 '19 at 17:58
  • Lock-free atomics let you do things that are just plain not possible with locks, like have readers that *never* have to wait even while writers are modifying the data structure. For example RCU (Read Copy Update) https://en.wikipedia.org/wiki/Read-copy-update and https://lwn.net/Articles/262464/ – Peter Cordes May 01 '19 at 18:05
  • @PeterCordes I agree with everything you said. But those operations are as you said, supported by the CPU. So in a sense, they are "truely" atomic and therefore do not incorporate a locking mechanism. But what if a process was not atomic, i.e. no hardware instruction for that, but we wanted to call it atomic anyway. Is there a method to do that, other than creating a software lock? For example, java getAndIncrement() atomic operation. https://stackoverflow.com/questions/19765127/the-getandincrement-implementation-of-atomicinteger shows that the trick is using for(;;) loop and compareandset() –  May 01 '19 at 22:31
  • Let us [continue this discussion in chat](https://chat.stackoverflow.com/rooms/192691/discussion-between-maximus-and-peter-cordes). –  May 01 '19 at 22:31