Mpsc.java

@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.NANOSECONDS)
@State(Scope.Group)
@Measurement(iterations = 5)
@Warmup(iterations = 5)
@Fork(value = 1)
public class Mpsc {
  static long NANO_DELAY = 10;
  static int SIZE = 4096;
  private static final Integer ONE = 777;
  static MpscLinkedQueue<Integer> mpsc = new MpscLinkedQueue<>();
  static ArrayList<Integer> list = new ArrayList<>();
  static ConcurrentLinkedQueue<Integer> conc = new ConcurrentLinkedQueue<>();
  static AtomicInteger NUM = new AtomicInteger();

  @State(Scope.Thread)
  public static class RxThread {

    public void mpscOffer() {
      if (NUM.get() < SIZE) {
        mpsc.offer(ONE);
        NUM.incrementAndGet();
      } else {
        backoff();
      }
    }

    public void syncOffer() {
      synchronized (list) {
        if (NUM.get() < SIZE) {
          list.add(ONE);
          NUM.incrementAndGet();
        } else {
          backoff();
        }
      }
    }

    public void concOffer() {
      if (NUM.get() < SIZE) {
        conc.offer(ONE);
        NUM.incrementAndGet();
      } else {
        backoff();
      }
    }

    public void mpscPoll(Blackhole blackhole) {
      if (NUM.get() > 0) {
        blackhole.consume(mpsc.poll());
        NUM.decrementAndGet();
      } else {
        backoff();
      }
    }

    public void syncPoll(Blackhole blackhole) {
      synchronized (list) {
        if (NUM.get() > 0) {
          blackhole.consume(list.remove(0));
          NUM.decrementAndGet();
        } else {
          backoff();
        }
      }
    }

    public void concPoll(Blackhole blackhole) {
      if (NUM.get() > 0) {
        blackhole.consume(conc.poll());
        NUM.decrementAndGet();
      } else {
        backoff();
      }
    }

    public void backoff() {
      LockSupport.parkNanos(NANO_DELAY);
    }
  }

  @Benchmark
  @Group("mpsc")
  @GroupThreads(4)
  public void mpscOffer(RxThread t) {
    t.mpscOffer();
  }

  @Benchmark
  @Group("mpsc")
  @GroupThreads(1)
  public void mpscPoll(RxThread t, Blackhole blackhole) {
    t.mpscPoll(blackhole);
  }

  @Benchmark
  @Group("sync")
  @GroupThreads(4)
  public void syncOffer(RxThread t) {
    t.syncOffer();
  }

  @Benchmark
  @Group("sync")
  @GroupThreads(1)
  public void syncPoll(RxThread t, Blackhole blackhole) {
    t.syncPoll(blackhole);
  }

  @Benchmark
  @Group("conc")
  @GroupThreads(4)
  public void concOffer(RxThread t) {
    t.concOffer();
  }

  @Benchmark
  @Group("conc")
  @GroupThreads(1)
  public void concPoll(RxThread t, Blackhole blackhole) {
    t.concPoll(blackhole);
  }

  @TearDown(Level.Iteration)
  public void clear() {
    list.clear();
    mpsc.clear();
    conc.clear();
    NUM.set(0);
  }
}