Testcase to reproduce a bug in JCTools MpscCompoundQueue.relaxedOffer that leads to infinite loop.

JcToolsMpscCompoundQueueRelaxedOfferInfiniteLoop.java

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;

import org.jctools.queues.MessagePassingQueue;
import org.jctools.queues.MpscCompoundQueue;

public class JcToolsMpscCompoundQueueRelaxedOfferInfiniteLoop {

  public static final int PARALLELISM = 8;
  public static final int DRAIN_SLEEP_NANOS = 100000;
  public static final int MIN_FAILED_DRAIN_ITERATIONS = 1000;

  public static void main(String[] args) throws Exception {
    final MpscCompoundQueue<String> strings = new MpscCompoundQueue<String>(16, 2);
    final AtomicInteger threadsReportingFullBitSet = new AtomicInteger(0);
    final ProducerThread[] threads = new ProducerThread[PARALLELISM];
    for (int i = 0; i < PARALLELISM; i++) {
      ProducerThread producerThread = new ProducerThread(i, strings, threadsReportingFullBitSet);
      producerThread.setDaemon(true);
      producerThread.start();
      threads[i] = producerThread;
    }

    Thread drainThread = new Thread() {
      @Override
      public void run() {
        int iterationsWhereAtLeastOneThreadDidNotReportFull = 0;
        while (iterationsWhereAtLeastOneThreadDidNotReportFull < MIN_FAILED_DRAIN_ITERATIONS) {
          if (threadsReportingFullBitSet.get() == ((1 << PARALLELISM) - 1)) {
            strings.drain(new MessagePassingQueue.Consumer<String>() {
              @Override
              public void accept(String e) {

              }
            }, strings.capacity());
            threadsReportingFullBitSet.set(0);
            iterationsWhereAtLeastOneThreadDidNotReportFull = 0;
          } else {
            iterationsWhereAtLeastOneThreadDidNotReportFull++;
          }
          LockSupport.parkNanos(DRAIN_SLEEP_NANOS);
        }
        System.out.println("Not all threads reported full queue after " + MIN_FAILED_DRAIN_ITERATIONS
            + " iterations - thread in infinite loop?");
      }
    };
    drainThread.setDaemon(true);
    drainThread.start();

    System.out.println("Waiting for concurrency issue to happen. This may take some time...");
    drainThread.join();
    for (int i = 0; i < PARALLELISM; i++) {
      System.out.println("Joining thread " + i);
      threads[i].interrupt();
      threads[i].join(2000);
      if (threads[i].isAlive()) {
        System.out.println("Thread " + i + " didn't join after 10 seconds - probably caught in infinite loop");
      }
    }

    System.out.println("Press enter to exit");
    System.in.read();
  }

  private static class ProducerThread extends Thread {
    private final int id;
    private final MpscCompoundQueue<String> strings;
    private final AtomicInteger full;

    public ProducerThread(int id, MpscCompoundQueue<String> strings, AtomicInteger full) {
      this.id = id;
      this.strings = strings;
      this.full = full;
    }

    @Override
    public void run() {
      for (int j = 0; j < 10000; j++) {
        for (int k = 0; k < 100000; k++) {
          boolean inserted = strings.relaxedOffer(String.valueOf(j * 100000 + k));
          if (! inserted) {
            int oldFull = full.get();
            full.compareAndSet(oldFull, oldFull | (1 << id));
          }
          if (Thread.interrupted()) {
            System.out.println("Thread " + id + " interrupted");
            return;
          }
        }
      }
      System.out.println("Finishing thread " + id + " regularly because all work is done");
    }
  }
}