Method bufferedIncrementStressTest

test/TestIntegration.java:573–643

Lots of buffered counter increments from multiple threads.

()

  /** Lots of buffered counter increments from multiple threads. */
  @Test
  public void bufferedIncrementStressTest() throws Exception {
    client.setFlushInterval(FAST_FLUSH);
    final byte[] table = TestIntegration.table.getBytes();
    final byte[] key1 = "cnt1".getBytes();  // Spread the increments..
    final byte[] key2 = "cnt2".getBytes();  // .. over these two counters.
    final byte[] family = TestIntegration.family.getBytes();
    final byte[] qual = { 'q' };
    final DeleteRequest del1 = new DeleteRequest(table, key1, family, qual);
    final DeleteRequest del2 = new DeleteRequest(table, key2, family, qual);
    Deferred.group(client.delete(del1), client.delete(del2)).join();

    final int nthreads = Runtime.getRuntime().availableProcessors() * 2;
    // The magic number comes from the limit on callbacks that Deferred
    // imposes.  We spread increments over two counters, hence the x 2.
    final int incr_per_thread = 8192 / nthreads * 2;
    final boolean[] successes = new boolean[nthreads];

    final class IncrementThread extends Thread {
      private final int num;
      public IncrementThread(final int num) {
        super("IncrementThread-" + num);
        this.num = num;
      }
      public void run() {
        try {
          doIncrements();
          successes[num] = true;
        } catch (Throwable e) {
          successes[num] = false;
          LOG.error("Uncaught exception", e);
        }
      }
      private void doIncrements() {
        for (int i = 0; i < incr_per_thread; i++) {
          final byte[] key = i % 2 == 0 ? key1 : key2;
          bufferIncrement(table, key, family, qual, 1);
        }
      }
    }

    final IncrementThread[] threads = new IncrementThread[nthreads];
    for (int i = 0; i < nthreads; i++) {
      threads[i] = new IncrementThread(i);
    }
    LOG.info("Starting to generate increments");
    for (int i = 0; i < nthreads; i++) {
      threads[i].start();
    }
    for (int i = 0; i < nthreads; i++) {
      threads[i].join();
    }

    LOG.info("Flushing all buffered increments.");
    client.flush().joinUninterruptibly();
    LOG.info("Done flushing all buffered increments.");

    // Check that we the counters have the expected value.