Java client jni netty

.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

benchmarks/utilities/csv_exporter.py

@@ -1,3 +1,5 @@
+#!/bin/python3
+
 import csv
 import json
 import os

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

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'
 
@@ -20,6 +23,9 @@ dependencies {
     implementation group: 'org.apache.commons', name: 'commons-lang3', version: '3.13.0'
     implementation group: 'com.google.code.gson', name: 'gson', version: '2.10.1'
 
+    //testImplementation group: 'org.slf4j', name: 'slf4j-reload4j', version: '2.0.9'
+    //testImplementation group: 'org.slf4j', name: 'slf4j-api', version: '2.0.9'
+
     compileOnly 'org.projectlombok:lombok:1.18.30'
     annotationProcessor 'org.projectlombok:lombok:1.18.30'
 }
@@ -34,12 +40,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"
 }

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");
@@ -232,9 +240,9 @@ public boolean isEqual(String other) {
 
   public static class RunConfiguration {
     public String configuration;
-    public Optional<String> resultsFile;
     public int[] dataSize;
     public int[] concurrentTasks;
+    public Optional<String> resultsFile;
     public ClientName[] clients;
     public String host;
     public int port;
@@ -251,7 +259,10 @@ public RunConfiguration() {
       clients =
           new ClientName[] {
             // ClientName.BABUSHKA_ASYNC,
-            ClientName.LETTUCE_ASYNC
+            // ClientName.JEDIS, ClientName.JEDIS_ASYNC, ClientName.LETTUCE,
+            // ClientName.LETTUCE_ASYNC
+            ClientName.BABUSHKA_ASYNC,
+            ClientName.BABUSHKA // , ClientName.LETTUCE, ClientName.LETTUCE_ASYNC
           };
       host = "localhost";
       port = 6379;

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;
+    }
+  }
+}

java/benchmarks/src/main/java/javababushka/benchmarks/Client.java → java/benchmarks/src/main/java/javababushka/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();
 

java/benchmarks/src/main/java/javababushka/benchmarks/SyncClient.java → java/benchmarks/src/main/java/javababushka/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);
 

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);
+  }
+}

java/benchmarks/src/main/java/javababushka/benchmarks/jedis/JedisClient.java → java/benchmarks/src/main/java/javababushka/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() {