Lossy SPSC Ring Buffer

Test.java

package test;


import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import java.util.concurrent.atomic.AtomicLong;
import org.openjdk.jcstress.annotations.Actor;
import org.openjdk.jcstress.annotations.Expect;
import org.openjdk.jcstress.annotations.JCStressTest;
import org.openjdk.jcstress.annotations.Outcome;
import org.openjdk.jcstress.annotations.State;
import org.openjdk.jcstress.infra.results.I_Result;

@JCStressTest
@Outcome(id = "32", expect = Expect.ACCEPTABLE, desc = "All Read")
@Outcome(id = "31", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "30", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "29", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "28", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "27", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "26", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "25", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "24", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "23", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "22", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "21", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "20", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "19", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "18", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "17", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "16", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "15", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "14", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "13", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "12", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "11", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "10", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "9", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "8", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "7", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "6", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "5", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "4", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "3", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "2", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "1", expect = Expect.ACCEPTABLE, desc = "Default outcome.")
@Outcome(id = "0", expect = Expect.ACCEPTABLE, desc = "No data outcome.")
@Outcome(id = "-1", expect = Expect.FORBIDDEN, desc = "race")
@State
public class Test {

  private static final int SIZE = 8; // must be a power of two
  private static final long MODMASK = 4*SIZE - 1;
  private static final long MASK = SIZE - 1;
  private static final VarHandle writerDataHandle;
  private static final VarHandle IDX;

  static {
    try {
      writerDataHandle = MethodHandles.arrayElementVarHandle(int[].class);
      IDX = MethodHandles.lookup().findVarHandle(ConcurrencyTest.class, "idx", long.class);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  // Writer state
  private final int[] writerData = new int[SIZE];
  private long idx;

  // Cached reader state
  private final int[] readerData = new int[SIZE];

  @Actor
  public void writer() {
    for (int k = 0; k < SIZE*4; k++) {
      long idx = (long) IDX.getAcquire(this);
      int i = (int) (idx & MASK);
      // Make sure to write different data each time through.
      writerDataHandle.setOpaque(writerData, i, (int) (idx & MODMASK));
      // writerData[i] = (int) (idx % MOD);
      IDX.setRelease(this, idx + 1);
    }
  }

  @Actor
  public void reader(I_Result r) {
    long startIdx = (long) IDX.getVolatile(this);
    int copy = (int) Math.min(startIdx, SIZE);
    for (int i = 0; i < SIZE; i++) {
       readerData[i] = (int) writerDataHandle.getOpaque(writerData, i);
    }
    // System.arraycopy(writerData, 0, readerData, 0, copy);
    long endIdx = (long) IDX.getVolatile(this);
    long elementsToDrop = endIdx - startIdx + 1;
    int ret = 0;
    for (int k = 0; k < copy - elementsToDrop; k++) {
      long idx = startIdx - k - 1;
      int i = (int) (idx & MASK);
      if (readerData[i] == (int)(idx & MODMASK)) {
        ret = 1;
      } else {
        ret = -1;
        break;
      }
    }
    if (ret == -1) {
      synchronized(ConcurrencyTest.class) {
        System.err.println("RDATA: " + java.util.Arrays.toString(readerData));
        System.err.println("IDX: " + startIdx + " " + endIdx + " " + elementsToDrop + " " + copy);
        for (int k = 0; k < copy - elementsToDrop; k++) {
          long idx = startIdx - k - 1;
          int i = (int) (idx & MASK);
          System.err.println(k + " " + idx + " " + i + " " + (int)(idx & MODMASK));
        }
      }
    }
    if (ret == 1) {
      ret = (int) endIdx;
    }
    r.r1 = ret;
  }
}