SpscLinkedQueue#offer implementation bugs #292
Description
Recently, I decided to add Chinese comments to JCTools. When I read the source code of SpscLinkedQueue, the implementation of the offer caused me some confusion, because it may cause consumerNode newer than producerNode and cause bugs. Therefore, I wrote SpscLinkedQueueTest for verification, and an exception occurred.
The producer should publish the new producerNode first, and then ensure that it is reachable from the consumerNode.
// final LinkedQueueNode<E> nextNode = newNode(e);
// final LinkedQueueNode<E> preNode = lpProducerNode();
// spProducerNode(nextNode);
// preNode.soNext(nextNode);
final LinkedQueueNode<E> nextNode = newNode(e);
lpProducerNode().soNext(nextNode);
spProducerNode(nextNode);
public class SpscLinkedQueueTest {
private static volatile boolean stop = false;
public static void main(String[] args) throws InterruptedException {
SpscLinkedQueue<String> messageQueue = new SpscLinkedQueue<>();
new Producer(messageQueue).start();
new Consumer(messageQueue).start();
try {
Thread.sleep(10 * 1000);
} finally {
stop = true;
}
}
private static class Producer extends Thread {
final SpscLinkedQueue<String> messageQueue;
long sequence = 0;
Producer(SpscLinkedQueue<String> messageQueue) {
this.messageQueue = messageQueue;
}
@Override
public void run() {
try {
while (!stop) {
messageQueue.offer(Long.toString(sequence++));
Thread.sleep(1);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
private static class Consumer extends Thread {
final SpscLinkedQueue<String> messageQueue;
private Consumer(SpscLinkedQueue<String> messageQueue) {
this.messageQueue = messageQueue;
}
@Override
public void run() {
while (!stop) {
messageQueue.poll();
if (!messageQueue.isEmpty()) {
final String e = messageQueue.poll();
if (null == e) {
throw new Error("MessageQueue.isEmpty() is false, messageQueue.poll() return null!");
}
}
}
}
}
}
Exception in thread "Thread-1" java.lang.Error: MessageQueue.isEmpty() is false, messageQueue.poll() return null!
at com.wjybxx.fastjgame.concurrenttest.SpscLinkedQueueTest$Consumer.run(SpscLinkedQueueTest.java:84)