Java client jni netty (#32)

* Add Java-client benchmarking app Signed-off-by: acarbonetto <[email protected]> * spotless apply Signed-off-by: acarbonetto <[email protected]> * Update on command line options Signed-off-by: acarbonetto <[email protected]> * Update README Signed-off-by: acarbonetto <[email protected]> * Spotless apply: Signed-off-by: acarbonetto <[email protected]> * Update README example Signed-off-by: acarbonetto <[email protected]> * update commandline defaults for review comments Signed-off-by: acarbonetto <[email protected]> * Remove TLS flag argument from option Signed-off-by: acarbonetto <[email protected]> * Add lettuce clients for benchmarking Signed-off-by: acarbonetto <[email protected]> * Spotless apply Signed-off-by: acarbonetto <[email protected]> * Add Jedis clients Signed-off-by: acarbonetto <[email protected]> * Add to app Signed-off-by: acarbonetto <[email protected]> * Add for-loop for data size list Signed-off-by: acarbonetto <[email protected]> * Add TPS for all async items Signed-off-by: acarbonetto <[email protected]> * spotless apply Signed-off-by: acarbonetto <[email protected]> * Fix TPS calculations Signed-off-by: acarbonetto <[email protected]> * Accept TLS as a flag Signed-off-by: acarbonetto <[email protected]> * Start threads; then wait for results Signed-off-by: acarbonetto <[email protected]> * Add java-jni client Signed-off-by: acarbonetto <[email protected]> * Handle Exceptions from client; add JniSyncClient fixes Signed-off-by: acarbonetto <[email protected]> * Clean up latency and add error checking Signed-off-by: acarbonetto <[email protected]> * Minor fixes. Signed-off-by: Yury-Fridlyand <[email protected]> * Fix result printing. Signed-off-by: Yury-Fridlyand <[email protected]> * Add TPS. Signed-off-by: Yury-Fridlyand <[email protected]> * Remove duplicates. Reorganize and fix imports. Signed-off-by: Yury-Fridlyand <[email protected]> * Int ctor fix. Signed-off-by: Yury-Fridlyand <[email protected]> * Iteration 1. Signed-off-by: Yury-Fridlyand <[email protected]> * Iteration 2: connected! Signed-off-by: Yury-Fridlyand <[email protected]> * Iteration 3: `get` and `set`. Signed-off-by: Yury-Fridlyand <[email protected]> * Iteration 4: benchmark. Signed-off-by: Yury-Fridlyand <[email protected]> * Iteration 5: some fixes. Signed-off-by: Yury-Fridlyand <[email protected]> * Change number of threads in Benchmarking threadpool * Revert "Change number of threads in Benchmarking threadpool" This reverts commit e3f7596. * Add more flushing rules and UT. Signed-off-by: Yury-Fridlyand <[email protected]> * Client clean up. Signed-off-by: Yury-Fridlyand <[email protected]> * Client optimizations. (#37) * Client optimizations. Signed-off-by: Yury-Fridlyand <[email protected]> * minor cleanup. Signed-off-by: Yury-Fridlyand <[email protected]> * Optimize building a command. Signed-off-by: Yury-Fridlyand <[email protected]> * Typo fix. Signed-off-by: Yury-Fridlyand <[email protected]> * Minor rename. Signed-off-by: Yury-Fridlyand <[email protected]> * Clean up Redis close connection Signed-off-by: Andrew Carbonetto <[email protected]> * Clean up Redis close connection Signed-off-by: Andrew Carbonetto <[email protected]> * Minor changes. Signed-off-by: Yury-Fridlyand <[email protected]> * Add todos to closeConnection() Signed-off-by: Andrew Carbonetto <[email protected]> --------- Signed-off-by: Yury-Fridlyand <[email protected]> Signed-off-by: Andrew Carbonetto <[email protected]> Co-authored-by: Andrew Carbonetto <[email protected]> * Address PR feedback. Signed-off-by: Yury-Fridlyand <[email protected]> * Rename Signed-off-by: Yury-Fridlyand <[email protected]> * Rename2 Signed-off-by: Yury-Fridlyand <[email protected]> * Fix CI Signed-off-by: Yury-Fridlyand <[email protected]> * More fixes. Signed-off-by: Yury-Fridlyand <[email protected]> * Some changes. Signed-off-by: Yury-Fridlyand <[email protected]> * add null check Signed-off-by: Yury-Fridlyand <[email protected]> * autoflush Signed-off-by: Yury-Fridlyand <[email protected]> * Apply suggestions from code review Signed-off-by: Yury-Fridlyand <[email protected]> Co-authored-by: Andrew Carbonetto <[email protected]> * minor changes Signed-off-by: Yury-Fridlyand <[email protected]> --------- Signed-off-by: acarbonetto <[email protected]> Signed-off-by: Yury-Fridlyand <[email protected]> Signed-off-by: Andrew Carbonetto <[email protected]> Co-authored-by: acarbonetto <[email protected]> Co-authored-by: Jonathan Louie <[email protected]>
Bit-Quill · Nov 22, 2023 · b50d57e · b50d57e
1 parent bcf188c
commit b50d57e
Show file tree

Hide file tree

Showing 25 changed files with 817 additions and 179 deletions.
diff --git a/.github/workflows/java.yml b/.github/workflows/java.yml
@@ -41,12 +41,10 @@ jobs:
           distribution: "temurin"
           java-version: ${{ matrix.java }}
 
-      - name: Install and run protoc (protobuf)
+      - name: Install protoc (protobuf)
         run: |
           sudo apt update
           sudo apt install -y protobuf-compiler
-          mkdir -p java/client/src/main/java/org/babushka/javababushka/generated
-          protoc -Iprotobuf=babushka-core/src/protobuf/ --java_out=java/client/src/main/java/org/babushka/javababushka/generated babushka-core/src/protobuf/*.proto
 
       - name: Build rust part
         working-directory: java

diff --git a/benchmarks/utilities/csv_exporter.py b/benchmarks/utilities/csv_exporter.py
@@ -1,3 +1,5 @@
+#!/bin/python3
+
 import csv
 import json
 import os

diff --git a/java/Cargo.toml b/java/Cargo.toml
@@ -15,6 +15,8 @@ babushka = { path = "../babushka-core" }
 tokio = { version = "^1", features = ["rt", "macros", "rt-multi-thread", "time"] }
 logger_core = {path = "../logger_core"}
 tracing-subscriber = "0.3.16"
+jni = "0.21.1"
+log = "0.4.20"
 
 [profile.release]
 lto = true

diff --git a/java/benchmarks/build.gradle b/java/benchmarks/build.gradle
@@ -1,6 +1,7 @@
 plugins {
     // Apply the application plugin to add support for building a CLI application in Java.
     id 'application'
+    id 'io.freefair.lombok'
 }
 
 repositories {
@@ -9,6 +10,8 @@ repositories {
 }
 
 dependencies {
+    implementation project(':client')
+
     // Use JUnit test framework.
     testImplementation 'org.junit.jupiter:junit-jupiter:5.9.2'
 
@@ -34,12 +37,14 @@ java {
 application {
     // Define the main class for the application.
     mainClass = 'javababushka.benchmarks.BenchmarkingApp'
+    applicationDefaultJvmArgs += "-Djava.library.path=${projectDir}/../target/release:${projectDir}/../target/debug"
 }
 
-tasks.withType(Test) { 
-  testLogging {
-    exceptionFormat "full"
-    events "started", "skipped", "passed", "failed"
-    showStandardStreams true
-  }
+tasks.withType(Test) {
+    testLogging {
+        exceptionFormat "full"
+        events "started", "skipped", "passed", "failed"
+        showStandardStreams true
+    }
+    jvmArgs "-Djava.library.path=${projectDir}/../target/release:${projectDir}/../target/debug"
 }
diff --git a/java/benchmarks/src/main/java/javababushka/benchmarks/AsyncClient.java b/java/benchmarks/src/main/java/javababushka/benchmarks/AsyncClient.java
diff --git a/java/benchmarks/src/main/java/javababushka/benchmarks/BenchmarkingApp.java b/java/benchmarks/src/main/java/javababushka/benchmarks/BenchmarkingApp.java
@@ -5,11 +5,12 @@
 import java.util.Arrays;
 import java.util.Optional;
 import java.util.stream.Stream;
-import javababushka.benchmarks.jedis.JedisClient;
-import javababushka.benchmarks.jedis.JedisPseudoAsyncClient;
-import javababushka.benchmarks.lettuce.LettuceAsyncClient;
-import javababushka.benchmarks.lettuce.LettuceAsyncClusterClient;
-import javababushka.benchmarks.lettuce.LettuceClient;
+import javababushka.benchmarks.clients.babushka.JniNettyClient;
+import javababushka.benchmarks.clients.jedis.JedisClient;
+import javababushka.benchmarks.clients.jedis.JedisPseudoAsyncClient;
+import javababushka.benchmarks.clients.lettuce.LettuceAsyncClient;
+import javababushka.benchmarks.clients.lettuce.LettuceAsyncClusterClient;
+import javababushka.benchmarks.clients.lettuce.LettuceClient;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.CommandLineParser;
 import org.apache.commons.cli.DefaultParser;
@@ -48,9 +49,11 @@ public static void main(String[] args) {
     for (ClientName client : runConfiguration.clients) {
       switch (client) {
         case JEDIS:
+          // run testClientSetGet on JEDIS sync client
           testClientSetGet(JedisClient::new, runConfiguration, false);
           break;
         case JEDIS_ASYNC:
+          // run testClientSetGet on JEDIS pseudo-async client
           testClientSetGet(JedisPseudoAsyncClient::new, runConfiguration, true);
           break;
         case LETTUCE:
@@ -63,8 +66,11 @@ public static void main(String[] args) {
             testClientSetGet(LettuceAsyncClient::new, runConfiguration, true);
           }
           break;
+        case BABUSHKA:
+          testClientSetGet(() -> new JniNettyClient(false), runConfiguration, false);
+          break;
         case BABUSHKA_ASYNC:
-          System.out.println("Babushka async not yet configured");
+          testClientSetGet(() -> new JniNettyClient(true), runConfiguration, true);
           break;
       }
     }
@@ -93,8 +99,8 @@ private static Options getOptions() {
         Option.builder("clients")
             .hasArg(true)
             .desc(
-                "one of: all|jedis|jedis_async|lettuce|lettuce_async"
-                    + "|babushka_async|all_async|all_sync [all]")
+                "one of: all|jedis|jedis_async|lettuce|lettuce_async|"
+                    + "babushka|babushka_async|all_async|all_sync")
             .build());
     options.addOption(Option.builder("host").hasArg(true).desc("Hostname [localhost]").build());
     options.addOption(Option.builder("port").hasArg(true).desc("Port number [6379]").build());
@@ -149,6 +155,7 @@ private static RunConfiguration verifyOptions(CommandLine line) throws ParseExce
                         return Stream.of(
                             ClientName.JEDIS,
                             ClientName.JEDIS_ASYNC,
+                            ClientName.BABUSHKA,
                             ClientName.BABUSHKA_ASYNC,
                             ClientName.LETTUCE,
                             ClientName.LETTUCE_ASYNC);
@@ -158,7 +165,7 @@ private static RunConfiguration verifyOptions(CommandLine line) throws ParseExce
                             ClientName.BABUSHKA_ASYNC,
                             ClientName.LETTUCE_ASYNC);
                       case ALL_SYNC:
-                        return Stream.of(ClientName.JEDIS, ClientName.LETTUCE);
+                        return Stream.of(ClientName.JEDIS, ClientName.LETTUCE, ClientName.BABUSHKA);
                       default:
                         return Stream.of(e);
                     }
@@ -210,6 +217,7 @@ public enum ClientName {
     LETTUCE("Lettuce"),
     LETTUCE_ASYNC("Lettuce async"),
     BABUSHKA_ASYNC("Babushka async"),
+    BABUSHKA("Babushka"),
     ALL("All"),
     ALL_SYNC("All sync"),
     ALL_ASYNC("All async");
@@ -250,8 +258,9 @@ public RunConfiguration() {
       concurrentTasks = new int[] {100, 1000};
       clients =
           new ClientName[] {
-            // ClientName.BABUSHKA_ASYNC,
-            ClientName.LETTUCE_ASYNC
+            // ClientName.LETTUCE,
+            // ClientName.LETTUCE_ASYNC,
+            ClientName.BABUSHKA_ASYNC, ClientName.BABUSHKA,
           };
       host = "localhost";
       port = 6379;

diff --git a/java/benchmarks/src/main/java/javababushka/benchmarks/clients/AsyncClient.java b/java/benchmarks/src/main/java/javababushka/benchmarks/clients/AsyncClient.java
@@ -0,0 +1,29 @@
+package javababushka.benchmarks.clients;
+
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import javababushka.benchmarks.utils.ConnectionSettings;
+
+/** A Redis client with async capabilities */
+public interface AsyncClient<T> extends Client {
+
+  long DEFAULT_TIMEOUT_MILLISECOND = 1000;
+
+  Future<T> asyncConnectToRedis(ConnectionSettings connectionSettings);
+
+  Future<T> asyncSet(String key, String value);
+
+  Future<String> asyncGet(String key);
+
+  default <T> T waitForResult(Future<T> future) {
+    return waitForResult(future, DEFAULT_TIMEOUT_MILLISECOND);
+  }
+
+  default <T> T waitForResult(Future<T> future, long timeout) {
+    try {
+      return future.get(timeout, TimeUnit.MILLISECONDS);
+    } catch (Exception ignored) {
+      return null;
+    }
+  }
+}
diff --git a/.../java/javababushka/benchmarks/Client.java → ...vababushka/benchmarks/clients/Client.java b/.../java/javababushka/benchmarks/Client.java → ...vababushka/benchmarks/clients/Client.java
@@ -1,7 +1,8 @@
-package javababushka.benchmarks;
+package javababushka.benchmarks.clients;
 
 import javababushka.benchmarks.utils.ConnectionSettings;
 
+/** A Redis client interface */
 public interface Client {
   void connectToRedis();
 

diff --git a/...a/javababushka/benchmarks/SyncClient.java → ...bushka/benchmarks/clients/SyncClient.java b/...a/javababushka/benchmarks/SyncClient.java → ...bushka/benchmarks/clients/SyncClient.java
@@ -1,5 +1,6 @@
-package javababushka.benchmarks;
+package javababushka.benchmarks.clients;
 
+/** A Redis client with sync capabilities */
 public interface SyncClient extends Client {
   void set(String key, String value);
 

diff --git a/java/benchmarks/src/main/java/javababushka/benchmarks/clients/babushka/JniNettyClient.java b/java/benchmarks/src/main/java/javababushka/benchmarks/clients/babushka/JniNettyClient.java
@@ -0,0 +1,69 @@
+package javababushka.benchmarks.clients.babushka;
+
+import static response.ResponseOuterClass.Response;
+
+import java.util.concurrent.Future;
+import javababushka.Client;
+import javababushka.benchmarks.clients.AsyncClient;
+import javababushka.benchmarks.clients.SyncClient;
+import javababushka.benchmarks.utils.ConnectionSettings;
+
+public class JniNettyClient implements SyncClient, AsyncClient<Response> {
+
+  private final Client testClient;
+  private String name = "JNI Netty";
+
+  public JniNettyClient(boolean async) {
+    name += async ? " async" : " sync";
+    testClient = new Client();
+  }
+
+  @Override
+  public String getName() {
+    return name;
+  }
+
+  @Override
+  public void closeConnection() {
+    testClient.closeConnection();
+  }
+
+  @Override
+  public void connectToRedis() {
+    connectToRedis(new ConnectionSettings("localhost", 6379, false, false));
+  }
+
+  @Override
+  public void connectToRedis(ConnectionSettings connectionSettings) {
+    waitForResult(asyncConnectToRedis(connectionSettings));
+  }
+
+  @Override
+  public Future<Response> asyncConnectToRedis(ConnectionSettings connectionSettings) {
+    return testClient.asyncConnectToRedis(
+        connectionSettings.host,
+        connectionSettings.port,
+        connectionSettings.useSsl,
+        connectionSettings.clusterMode);
+  }
+
+  @Override
+  public Future<Response> asyncSet(String key, String value) {
+    return testClient.asyncSet(key, value);
+  }
+
+  @Override
+  public Future<String> asyncGet(String key) {
+    return testClient.asyncGet(key);
+  }
+
+  @Override
+  public void set(String key, String value) {
+    testClient.set(key, value);
+  }
+
+  @Override
+  public String get(String key) {
+    return testClient.get(key);
+  }
+}
diff --git a/...abushka/benchmarks/jedis/JedisClient.java → ...benchmarks/clients/jedis/JedisClient.java b/...abushka/benchmarks/jedis/JedisClient.java → ...benchmarks/clients/jedis/JedisClient.java
@@ -1,24 +1,18 @@
-/*
- * This Java source file was generated by the Gradle 'init' task.
- */
-package javababushka.benchmarks.jedis;
+package javababushka.benchmarks.clients.jedis;
 
-import javababushka.benchmarks.SyncClient;
+import javababushka.benchmarks.clients.SyncClient;
 import javababushka.benchmarks.utils.ConnectionSettings;
 import redis.clients.jedis.Jedis;
 import redis.clients.jedis.JedisPool;
 
+/** A Jedis client with sync capabilities. See: https://github.com/redis/jedis */
 public class JedisClient implements SyncClient {
 
   public static final String DEFAULT_HOST = "localhost";
   public static final int DEFAULT_PORT = 6379;
 
   protected Jedis jedisResource;
 
-  public boolean someLibraryMethod() {
-    return true;
-  }
-
   @Override
   public void connectToRedis() {
     JedisPool pool = new JedisPool(DEFAULT_HOST, DEFAULT_PORT);
@@ -43,6 +37,9 @@ public void connectToRedis(ConnectionSettings connectionSettings) {
     jedisResource =
         new Jedis(connectionSettings.host, connectionSettings.port, connectionSettings.useSsl);
     jedisResource.connect();
+    if (!jedisResource.isConnected()) {
+      throw new RuntimeException("failed to connect to jedis");
+    }
   }
 
   public String info() {

diff --git a/...chmarks/jedis/JedisPseudoAsyncClient.java → ...clients/jedis/JedisPseudoAsyncClient.java b/...chmarks/jedis/JedisPseudoAsyncClient.java → ...clients/jedis/JedisPseudoAsyncClient.java
@@ -1,13 +1,22 @@
-package javababushka.benchmarks.jedis;
+package javababushka.benchmarks.clients.jedis;
 
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import javababushka.benchmarks.AsyncClient;
+import javababushka.benchmarks.clients.AsyncClient;
+import javababushka.benchmarks.utils.ConnectionSettings;
 
-// Jedis doesn't provide async API
-// https://github.com/redis/jedis/issues/241
+/**
+ * A Jedis client with pseudo-async capabilities. Jedis doesn't provide async API
+ * https://github.com/redis/jedis/issues/241
+ *
+ * <p>See: https://github.com/redis/jedis
+ */
 public class JedisPseudoAsyncClient extends JedisClient implements AsyncClient {
+  @Override
+  public Future<?> asyncConnectToRedis(ConnectionSettings connectionSettings) {
+    return CompletableFuture.runAsync(() -> super.connectToRedis(connectionSettings));
+  }
+
   @Override
   public Future<?> asyncSet(String key, String value) {
     return CompletableFuture.runAsync(() -> super.set(key, value));
@@ -18,20 +27,6 @@ public Future<String> asyncGet(String key) {
     return CompletableFuture.supplyAsync(() -> super.get(key));
   }
 
-  @Override
-  public <T> T waitForResult(Future<T> future) {
-    return waitForResult(future, DEFAULT_TIMEOUT);
-  }
-
-  @Override
-  public <T> T waitForResult(Future<T> future, long timeout) {
-    try {
-      return future.get(timeout, TimeUnit.MILLISECONDS);
-    } catch (Exception ignored) {
-      return null;
-    }
-  }
-
   @Override
   public String getName() {
     return "Jedis pseudo-async";