Adding support for jedis

Author: tramchamploo

.circleci/config.yml

@@ -6,6 +6,7 @@ jobs:
 
     docker:
       - image: circleci/openjdk:8-jdk-browsers
+      - image: circleci/redis:latest
 
     steps:
       - checkout

benchmark/pom.xml

@@ -65,6 +65,11 @@
       <optional>true</optional>
     </dependency>
 
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>bufferslayer-jedis</artifactId>
+    </dependency>
+
     <dependency>
       <groupId>ch.qos.logback</groupId>
       <artifactId>logback-classic</artifactId>

benchmark/src/main/java/io/github/tramchamploo/bufferslayer/AbstractBatchJdbcTemplateBenchmark.java

@@ -25,7 +25,7 @@ public abstract class AbstractBatchJdbcTemplateBenchmark {
   private BatchJdbcTemplate batch;
   private JdbcTemplate unbatch;
   private Reporter<Sql, Integer> reporter;
-  private static SenderProxy proxy;
+  private static SenderProxy<Sql, Integer> proxy;
   private static AtomicLong counter = new AtomicLong();
 
   private static final String CREATE_DATABASE = "CREATE DATABASE IF NOT EXISTS test";
@@ -51,7 +51,7 @@ public void setup() {
     JdbcTemplate delegate = new JdbcTemplate(dataSource);
     delegate.setDataSource(dataSource);
 
-    proxy = new SenderProxy(new JdbcTemplateSender(delegate));
+    proxy = new SenderProxy<>(new JdbcTemplateSender(delegate));
     proxy.onMessages(updated -> counter.addAndGet(updated.size()));
 
     reporter = reporter(proxy);

benchmark/src/main/java/io/github/tramchamploo/bufferslayer/AbstractBatchJedisBenchmark.java

@@ -0,0 +1,116 @@
+package io.github.tramchamploo.bufferslayer;
+
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.openjdk.jmh.annotations.AuxCounters;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Group;
+import org.openjdk.jmh.annotations.GroupThreads;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPool;
+
+/**
+ * Jedis benchmark that executing jedis commands.
+ */
+public abstract class AbstractBatchJedisBenchmark {
+
+  private JedisPool unbatch;
+  private BatchJedis batch;
+  private Reporter<RedisCommand, Object> reporter;
+  private static SenderProxy<RedisCommand, Object> proxy;
+  private static AtomicLong counter = new AtomicLong();
+
+  static String propertyOr(String key, String fallback) {
+    return System.getProperty(key, fallback);
+  }
+
+  protected abstract Reporter<RedisCommand, Object> reporter(Sender<RedisCommand, Object> sender);
+
+  @Setup
+  public void setup() {
+    unbatch = new JedisPool(propertyOr("redisHost", "127.0.0.1"),
+        Integer.parseInt(propertyOr("redisPort", "6379")));
+
+    proxy = new SenderProxy<>(new JedisSender(unbatch));
+    proxy.onMessages(updated -> counter.addAndGet(updated.size()));
+
+    reporter = reporter(proxy);
+    batch = new BatchJedis(unbatch, reporter);
+  }
+
+  @TearDown(Level.Iteration)
+  public void flushDB() {
+    try (Jedis jedis = unbatch.getResource()) {
+      jedis.flushDB();
+    }
+  }
+
+  @AuxCounters
+  @State(Scope.Thread)
+  public static class AtomicLongCounter {
+
+    public long updated() {
+      return counter.get();
+    }
+
+    @Setup(Level.Iteration)
+    public void clean() {
+      counter.set(0);
+    }
+  }
+
+  @State(Scope.Benchmark)
+  public static class Lagging {
+
+    @Setup(Level.Iteration)
+    public void lag() throws InterruptedException {
+      TimeUnit.SECONDS.sleep(1);
+    }
+  }
+
+  static String randomString() {
+    return String.valueOf(ThreadLocalRandom.current().nextLong());
+  }
+
+  @Benchmark @Group("no_contention_batched") @GroupThreads(1)
+  public void no_contention_batched_set(Lagging l, AtomicLongCounter counters) {
+    batch.set(randomString(), randomString());
+  }
+
+  @Benchmark @Group("no_contention_unbatched") @GroupThreads(1)
+  public void no_contention_unbatched_set(Lagging l) {
+    try (Jedis jedis = unbatch.getResource()) {
+      jedis.set(randomString(), randomString());
+    }
+  }
+
+  @Benchmark @Group("mild_contention_batched") @GroupThreads(2)
+  public void mild_contention_batched_set(Lagging l, AtomicLongCounter counters) {
+    batch.set(randomString(), randomString());
+  }
+
+  @Benchmark @Group("mild_contention_unbatched") @GroupThreads(2)
+  public void mild_contention_unbatched_set(Lagging l) {
+    try (Jedis jedis = unbatch.getResource()) {
+      jedis.set(randomString(), randomString());
+    }
+  }
+
+  @Benchmark @Group("high_contention_batched") @GroupThreads(8)
+  public void high_contention_batched_set(Lagging l, AtomicLongCounter counters) {
+    batch.set(randomString(), randomString());
+  }
+
+  @Benchmark @Group("high_contention_unbatched") @GroupThreads(8)
+  public void high_contention_unbatched_set(Lagging l) {
+    try (Jedis jedis = unbatch.getResource()) {
+      jedis.set(randomString(), randomString());
+    }
+  }
+}

benchmark/src/main/java/io/github/tramchamploo/bufferslayer/AbstractTimeUsedComparison.java

@@ -21,7 +21,7 @@ static String propertyOr(String key, String fallback) {
   protected void run() throws Exception {
     BatchJdbcTemplate batch;
     JdbcTemplate unbatch;
-    SenderProxy proxy;
+    SenderProxy<Sql, Integer> proxy;
 
     DriverManagerDataSource dataSource = new DriverManagerDataSource();
     dataSource.setDriverClassName("com.mysql.jdbc.Driver");
@@ -42,7 +42,7 @@ protected void run() throws Exception {
 
     CountDownLatch countDown = new CountDownLatch(1);
 
-    proxy = new SenderProxy(new JdbcTemplateSender(delegate));
+    proxy = new SenderProxy<>(new JdbcTemplateSender(delegate));
     proxy.onMessages(updated -> {
       if (counter.addAndGet(updated.size()) == 5050) {
         countDown.countDown();

benchmark/src/main/java/io/github/tramchamploo/bufferslayer/AsyncBatchJedisBenchmark.java

@@ -0,0 +1,38 @@
+package io.github.tramchamploo.bufferslayer;
+
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+@Measurement(iterations = 5, time = 1)
+@Warmup(iterations = 3, time = 1)
+@Fork(3)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@State(Scope.Group)
+public class AsyncBatchJedisBenchmark extends AbstractBatchJedisBenchmark {
+
+  protected Reporter<RedisCommand, Object> reporter(Sender<RedisCommand, Object> sender) {
+    return AsyncReporter.builder(sender)
+        .pendingKeepalive(1, TimeUnit.SECONDS)
+        .build();
+  }
+
+  public static void main(String[] args) throws RunnerException {
+    Options opt = new OptionsBuilder()
+        .include(".*" + AsyncBatchJedisBenchmark.class.getSimpleName() + ".*")
+        .build();
+
+    new Runner(opt).run();
+  }
+}

boundedqueue/src/test/java/io/github/tramchamploo/bufferslayer/AsyncReporterTest.java

@@ -67,6 +67,8 @@ public void flushIfExceedMaxSize() throws InterruptedException {
     assertTrue(countDown.await(300, TimeUnit.MILLISECONDS));
     assertEquals(50, sender.sent.size());
 
+    // wait for the queue to be released
+    Thread.sleep(100);
     // make sure the queue is released
     assertEquals(0, reporter.synchronizer.deque.size());
     assertEquals(0, reporter.synchronizer.keyToReady.size());

circle.yml

@@ -0,0 +1,19 @@
+machine:
+  java:
+    version: oraclejdk8
+  services:
+    - redis
+
+dependencies:
+  pre:
+    - openssl aes-256-cbc -d -in .buildscript/secret-env-cipher -k $KEY >> ~/.circlerc
+  override:
+    - mvn --fail-never dependency:go-offline || true
+  cache_directories:
+    - "~/.m2"
+
+test:
+  post:
+    - mkdir -p $CIRCLE_TEST_REPORTS/junit/
+    - find . -type f -regex ".*/target/surefire-reports/.*xml" -exec cp {} $CIRCLE_TEST_REPORTS/junit/ \;
+    - .buildscript/release.sh

core/src/main/java/io/github/tramchamploo/bufferslayer/internal/CompositeFuture.java

@@ -14,7 +14,7 @@ public abstract class CompositeFuture extends AbstractFuture<CompositeFuture> {
    * When the list is empty, the returned future will be already completed.
    */
   public static CompositeFuture all(List<? extends Future<?>> futures) {
-    return DefaultCompositeFuture.all(futures.toArray(new Future[futures.size()]));
+    return DefaultCompositeFuture.all(futures.toArray(new Future[0]));
   }
 
   /**

core/src/main/java/io/github/tramchamploo/bufferslayer/internal/Promises.java

@@ -6,6 +6,7 @@
 import io.github.tramchamploo.bufferslayer.Message;
 import io.github.tramchamploo.bufferslayer.MessageDroppedException;
 import io.github.tramchamploo.bufferslayer.OverflowStrategy.Strategy;
+import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
 
@@ -19,7 +20,12 @@ public static <R> void allSuccess(List<R> result, List<MessagePromise<R>> promis
     for (int i = 0; i < result.size(); i++) {
       MessagePromise<R> promise = promises.get(i);
       R ret = result.get(i);
-      promise.setSuccess(ret);
+      if (ret instanceof Throwable) {
+        promise.setFailure(MessageDroppedException.dropped((Throwable) ret,
+            Collections.singletonList(promise.message())));
+      } else {
+        promise.setSuccess(ret);
+      }
     }
   }