package stormpot.benchmark;

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

public class MultiThreadedThroughputBenchmark extends Benchmark {
  private static final int THREAD_COUNT = Integer.getInteger("thread.count",
        Runtime.getRuntime().availableProcessors());

  private WorkerThread[] threads;
  private volatile CountDownLatch doneLatch;
  
  @Override
  protected String getBenchmarkName() {
    return "Stormpot Multi-Threaded Throughput Benchmark";
  }

  @Override
  protected int[] warmupCycles() {
    return new int[] {1, 13, 1, 1, 13, 1, 1};
  }

  @Override
  protected void warmup(Bench bench, int steps) throws Exception {
    System.out.println(
        "Warming up " + bench.getName() + " with " + steps + " steps.");
    for (int i = 0; i < steps; i++) {
      prepareAndRunBenchmark(bench, 1000);
      System.out.printf("%02d/%02d.", i + 1, steps);
    }
    System.out.println("\ndone.");
  }

  @Override
  protected void beforeBenchmark(Bench bench, long trialTimeMillis) {
    super.beforeBenchmark(bench, trialTimeMillis);
    ensureThreadsCreated();
    doneLatch = new CountDownLatch(THREAD_COUNT);
    for (WorkerThread th : threads) {
      th.bench = bench;
      th.doneLatch = doneLatch;
    }
  }

  protected void ensureThreadsCreated() {
    if (threads == null) {
      threads = new WorkerThread[THREAD_COUNT];
      for (int i = 0; i < THREAD_COUNT; i++) {
        threads[i] = new WorkerThread();
        threads[i].start();
      }
    }
  }

  @Override
  protected void benchmark(Bench bench, long trialTimeMillis) throws Exception {
    long start = System.currentTimeMillis();
    long deadline = start + trialTimeMillis;
    start(threads, deadline);
    doneLatch.await();
    long end = System.currentTimeMillis();
    bench.recordPeriod(end - start);
  }
  
  private void start(WorkerThread[] threads, long deadline) {
    for (WorkerThread th : threads) {
      th.deadline = deadline;
      LockSupport.unpark(th);
    }
  }

  private static class WorkerThread extends Thread {
    private static final AtomicInteger counter = new AtomicInteger();
    public Bench bench;
    public volatile long deadline;
    public volatile CountDownLatch doneLatch;
    
    public WorkerThread() {
      super("WorkerThread-" + counter.incrementAndGet());
      setDaemon(true);
    }
    
    @Override
    public void run() {
      try {
        runInterruptibly();
      } catch (InterruptedException e) {
        // and now we're done...
      } catch (Exception e) {
        e.printStackTrace();
      }
    }

    private void runInterruptibly() throws Exception {
      for (;;) {
        waitForStartingSignal();
        doBenchmarkWork();
      }
    }

    private void waitForStartingSignal() {
      LockSupport.park();
    }

    protected void doBenchmarkWork() throws Exception {
      CountDownLatch doneLatch = this.doneLatch;
      Bench bench = this.bench;
      long deadline = this.deadline;
      do {
        runCycles(bench, 100);
      } while (System.currentTimeMillis() < deadline);
      doneLatch.countDown();
    }
  }
}

@@ -119,55 +119,8 @@
 ger;
-%0Aimport java.util.concurrent.locks.LockSupport;
 %0A%0Apu
@@ -348,16 +348,62 @@
 hreads;%0A
+  private volatile CountDownLatch startLatch;%0A
   privat
@@ -1149,37 +1149,54 @@
 s);%0A    
-ensureThreadsCreated(
+%0A    startLatch = new CountDownLatch(1
 );%0A    d
@@ -1244,184 +1244,8 @@
 T);%0A
-    for (WorkerThread th : threads) %7B%0A      th.bench = bench;%0A      th.doneLatch = doneLatch;%0A    %7D%0A  %7D%0A%0A  protected void ensureThreadsCreated() %7B%0A    if (threads == null) %7B%0A  
     
@@ -1286,18 +1286,16 @@
 COUNT%5D;%0A
-  
     for 
@@ -1333,26 +1333,24 @@
 ++) %7B%0A      
-  
 threads%5Bi%5D =
@@ -1367,21 +1367,47 @@
 rThread(
+bench, startLatch, doneLatch
 );%0A
-  
       th
@@ -1420,32 +1420,24 @@
 i%5D.start();%0A
-      %7D%0A
     %7D%0A  %7D%0A%0A 
@@ -1626,31 +1626,103 @@
     
-start(threads, deadline
+for (WorkerThread th : threads) %7B%0A      th.deadline = deadline;%0A    %7D%0A    startLatch.countDown(
 );%0A 
@@ -1834,179 +1834,8 @@
 %0A  %0A
-  private void start(WorkerThread%5B%5D threads, long deadline) %7B%0A    for (WorkerThread th : threads) %7B%0A      th.deadline = deadline;%0A      LockSupport.unpark(th);%0A    %7D%0A  %7D%0A%0A
   pr
@@ -1957,32 +1957,8 @@
 ();%0A
-    public Bench bench;%0A
     
@@ -1997,501 +1997,482 @@
    p
-ublic volatile CountDownLatch doneLatch;%0A    %0A    public WorkerThread() %7B%0A      super(%22WorkerThread-%22 + counter.incrementAndGet());%0A      setDaemon(true);%0A    %7D%0A    %0A    @Override%0A    public void run() %7B%0A      try %7B%0A        runInterruptibly();%0A      %7D catch (InterruptedException e) %7B%0A        // and now we're done...%0A      %7D catch (Exception e) %7B
+rivate final Bench bench;%0A    private final CountDownLatch doneLatch;%0A    private final CountDownLatch startLatch;%0A    %0A    public WorkerThread(Bench bench, CountDownLatch startLatch, CountDownLatch doneLatch) %7B%0A      super(%22WorkerThread-%22 + counter.incrementAndGet());%0A      setDaemon(true);%0A      this.bench = bench;%0A      this.startLatch = startLatch;%0A      this.doneLatch = doneLatch;%0A    %7D
 %0A    
+%0A
     
-e.printStackTrace();%0A      %7D%0A    %7D%0A%0A    private void runInterruptibly() throws Exception %7B%0A      for (;;) %7B%0A        waitForStartingSignal
+@Override%0A    public void run() %7B%0A      try %7B%0A        startLatch.await
 ();%0A
@@ -2508,85 +2508,99 @@
     
+  
-%7D%0A    %7D%0A%0A    private void waitForStartingSignal() %7B%0A      LockSupport.park();
+doneLatch.countDown();%0A      %7D catch (Exception e) %7B%0A        e.printStackTrace();%0A      %7D
 %0A   
@@ -2663,57 +2663,8 @@
 n %7B%0A
-      CountDownLatch doneLatch = this.doneLatch;%0A
     
@@ -2829,37 +2829,8 @@
 e);%0A
-      doneLatch.countDown();%0A
     

simplify the multi-threaded benchmark by not reusing threads, hoping to be rid of that occasional liveness failure.