Add a benchmark for Redis

WORKSPACE

@@ -132,6 +132,7 @@ maven_install(
         "io.netty:netty-buffer:4.1.79.Final",
         "io.netty:netty5-common:5.0.0.Alpha4",
         "io.netty:netty5-buffer:5.0.0.Alpha4",
+        "io.lettuce:lettuce-core:6.2.0.RELEASE",
     ],
     fetch_sources = True,
     repositories = maven_repositories,

src/com/komanov/redis/BUILD

@@ -0,0 +1,12 @@
+package(default_visibility = ["//src/com/komanov/redis:__subpackages__"])
+
+scala_library(
+    name = "redis",
+    srcs = glob([
+        "*.scala",
+        "*.java",
+    ]),
+    deps = [
+        "@offheap_maven//:io_lettuce_lettuce_core",
+    ],
+)

src/com/komanov/redis/README.md

@@ -0,0 +1,27 @@
+## Redis Benchmark for Set (SISMEMBER command)
+
+Prepare data:
+```
+bazel run //src/com/komanov/redis/bin -- set-1m ~/uuid1m.txt
+bazel run //src/com/komanov/redis/bin -- set-10m ~/uuid10m.txt
+bazel run //src/com/komanov/redis/bin -- set-100k ~/uuid100k.txt
+```
+
+Run Redis in docker:
+```
+docker run --name my-redis -p 6379:6379 redis:7.0-alpine
+```
+
+Run cli:
+```
+docker exec -it my-redis redis-cli
+```
+
+Run benchmark:
+```
+time bazel run //src/com/komanov/redis/perf set-1m ~/uuid1m.txt
+time bazel run //src/com/komanov/redis/perf set-10m ~/uuid10m.txt
+time bazel run //src/com/komanov/redis/perf set-100k ~/uuid100k.txt
+```
+
+Results: https://docs.google.com/spreadsheets/d/1D5fhP-rxuxamOl58cGk7yLiLiViZnxWcE71klp6JC3I

src/com/komanov/redis/StringUuidCodec.scala

@@ -0,0 +1,27 @@
+package com.komanov.redis
+
+import io.lettuce.core.codec.{RedisCodec, StringCodec}
+
+import java.nio.ByteBuffer
+import java.util.UUID
+
+class StringUuidCodec extends RedisCodec[String, UUID] {
+  override def decodeKey(bytes: ByteBuffer): String =
+    StringCodec.ASCII.decodeKey(bytes)
+
+  override def decodeValue(bytes: ByteBuffer): UUID =
+    new UUID(bytes.getLong, bytes.getLong)
+
+  override def encodeKey(key: String): ByteBuffer =
+    StringCodec.ASCII.encodeKey(key)
+
+  override def encodeValue(value: UUID): ByteBuffer = {
+    val bb = ByteBuffer.allocate(16)
+    bb.putLong(value.getMostSignificantBits)
+    bb.putLong(value.getLeastSignificantBits)
+    bb.rewind()
+    bb
+  }
+}
+
+object StringUuidCodec extends StringUuidCodec

src/com/komanov/redis/bin/BUILD

@@ -0,0 +1,9 @@
+scala_binary(
+    name = "bin",
+    srcs = ["DataFiller.scala"],
+    main_class = "com.komanov.redis.bin.DataFiller",
+    deps = [
+        "//src/com/komanov/redis",
+        "@offheap_maven//:io_lettuce_lettuce_core",
+    ],
+)

src/com/komanov/redis/bin/DataFiller.scala

@@ -0,0 +1,43 @@
+package com.komanov.redis.bin
+
+import com.komanov.redis.StringUuidCodec
+import io.lettuce.core.RedisClient
+import io.lettuce.core.api.StatefulRedisConnection
+
+import java.nio.file.{Files, Paths}
+import java.util.UUID
+import scala.collection.mutable
+
+// Usage: SET_NAME INPUT_FILE
+object DataFiller extends App {
+
+  val Array(setName, input) = args
+
+  val inputPath = Paths.get(input)
+  require(Files.exists(inputPath), s"INPUT_FILE doesn't exist: $input")
+
+  val client = RedisClient.create("redis://localhost:6379")
+  val connection: StatefulRedisConnection[String, UUID] = client.connect(StringUuidCodec);
+  val sync = connection.sync();
+
+  val chunk = mutable.ListBuffer[UUID]()
+
+  def dumpChunk(): Unit = {
+    val num = sync.sadd(setName, chunk.toSeq: _*)
+    require(num == chunk.size, s"Failed to SADD: $num != ${chunk.size}")
+    chunk.clear()
+  }
+
+  Files.lines(inputPath).forEach { line =>
+    chunk += UUID.fromString(line)
+    if (chunk.size == 1000) {
+      dumpChunk()
+    }
+  }
+  if (chunk.nonEmpty) {
+    dumpChunk()
+  }
+
+  client.shutdown()
+
+}

src/com/komanov/redis/perf/BUILD

@@ -0,0 +1,9 @@
+java_binary(
+    name = "perf",
+    srcs = ["PerfTester.java"],
+    main_class = "com.komanov.redis.perf.PerfTester",
+    deps = [
+        "//src/com/komanov/redis",
+        "@offheap_maven//:io_lettuce_lettuce_core",
+    ],
+)

src/com/komanov/redis/perf/PerfTester.java

@@ -0,0 +1,131 @@
+package com.komanov.redis.perf;
+
+import com.komanov.redis.StringUuidCodec$;
+import io.lettuce.core.RedisClient;
+import io.lettuce.core.api.StatefulRedisConnection;
+import io.lettuce.core.api.async.RedisAsyncCommands;
+
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.stream.Collectors;
+
+public class PerfTester {
+    private static class InFlight {
+        public final int index;
+        public long startTime;
+
+        public InFlight(int index) {
+            this.index = index;
+        }
+    }
+
+    public static void main(String[] args) throws Throwable {
+        assert args.length == 2;
+
+        String setName = args[0];
+        List<UUID> uuids = Files.readAllLines(Paths.get(args[1]))
+                .stream()
+                .map(UUID::fromString)
+                .limit(1_000_000)
+                .collect(Collectors.toList());
+
+        RedisClient client = RedisClient.create("redis://localhost:6379");
+        StatefulRedisConnection<String, UUID> connection = client.connect(StringUuidCodec$.MODULE$);
+        RedisAsyncCommands<String, UUID> async = connection.async();
+
+        System.out.println("About to run warmup");
+        int warmupCount = 10_000;
+        CountDownLatch latch = new CountDownLatch(warmupCount);
+        uuids.stream().limit(warmupCount).forEach(v -> {
+            async.sismember(setName, v).handle((r, e) -> {
+                latch.countDown();
+                return r;
+            });
+        });
+        System.out.println("Awaiting for warmup to complete");
+        latch.await();
+        System.out.println("Warmup completed!");
+
+        ArrayList<ArrayList<String>> results = new ArrayList<>();
+        for (int parallelism = 0; parallelism <= 90; parallelism += 10) {
+            results.add(runBenchmark(async, uuids, setName, parallelism == 0 ? 1 : parallelism));
+        }
+        for (int parallelism = 100; parallelism <= 1000; parallelism += 100) {
+            results.add(runBenchmark(async, uuids, setName, parallelism));
+        }
+        results.add(runBenchmark(async, uuids, setName, 2000));
+
+        for (int num = 0; num < results.get(0).size(); ++num) {
+            for (int i = 0; i < results.size(); ++i) {
+                ArrayList<String> single = results.get(i);
+                System.out.print(single.get(num) + "\t");
+            }
+            System.out.println();
+        }
+
+        connection.close();
+        client.shutdown();
+    }
+
+    private static ArrayList<String> runBenchmark(RedisAsyncCommands<String, UUID> async, List<UUID> uuids, String setName, int parallelism)
+            throws Throwable {
+        long[] durations = new long[uuids.size()];
+        LinkedBlockingDeque<InFlight> queue = new LinkedBlockingDeque<>(parallelism);
+        long overallStartTime = System.nanoTime();
+        for (int i = 0; i < uuids.size(); i++) {
+            final InFlight item = new InFlight(i);
+            queue.put(item);
+            item.startTime = System.nanoTime();
+            async
+                    .sismember(setName, uuids.get(i))
+                    .handle((isMember, e) -> {
+                        if (e == null) {
+                            if (!isMember) {
+                                System.out.println("NOT MEMBER!");
+                                System.exit(1);
+                            }
+                            durations[item.index] = System.nanoTime() - item.startTime;
+                            if (!queue.remove(item)) {
+                                System.out.println("NOT IN QUEUE!");
+                                System.exit(2);
+                            }
+                        } else {
+                            e.printStackTrace();
+                            System.exit(1);
+                        }
+                        return Boolean.TRUE;
+                    })
+            ;
+        }
+
+        while (!queue.isEmpty()) {
+            Thread.sleep(0);
+        }
+
+        long overallDuration = System.nanoTime() - overallStartTime;
+
+        ArrayList<String> results = new ArrayList<>();
+        results.add("" + uuids.size());
+        results.add("" + parallelism);
+        results.add("" + overallDuration);
+        results.add("" + overallDuration / uuids.size());
+
+        Arrays.sort(durations);
+        Arrays.stream(durations).limit(5).forEach(d -> {
+            results.add("" + d);
+        });
+        Arrays.stream(durations).skip(durations.length / 2).limit(5).forEach(d -> {
+            results.add("" + d);
+        });
+        Arrays.stream(durations).skip(durations.length - 5).forEach(d -> {
+            results.add("" + d);
+        });
+        return results;
+    }
+}