Merge pull request redis-rs#195 from shachlanAmazon/cluster-benchmark

Add cluster option to node benchmarks.

benchmarks/csharp/Program.cs

@@ -227,6 +227,7 @@ int num_of_concurrent_tasks
             {"data_size", data_size},
             {"tps", tps},
             {"clientCount", clients.Length},
+            {"is_cluster", "false"}
         };
         result = result
             .Concat(get_existing_latency_results)
@@ -349,7 +350,7 @@ await run_clients(
 
     private static int number_of_iterations(int num_of_concurrent_tasks)
     {
-        return Math.Max(100000, num_of_concurrent_tasks * 10000);
+        return Math.min(Math.Max(100000, num_of_concurrent_tasks * 10000), 10000000);
     }
 
     public static async Task Main(string[] args)

benchmarks/install_and_test.sh

@@ -57,7 +57,7 @@ function runNodeBenchmark(){
   cd ${BENCH_FOLDER}/node
   npm install
   npx tsc
-  npm run bench -- --resultsFile=../$1 --dataSize $2 --concurrentTasks $concurrentTasks --clients $chosenClients --host $host --clientCount $clientCount $tlsFlag
+  npm run bench -- --resultsFile=../$1 --dataSize $2 --concurrentTasks $concurrentTasks --clients $chosenClients --host $host --clientCount $clientCount $tlsFlag $clusterFlag
 }
 
 function runCSharpBenchmark(){
@@ -70,13 +70,13 @@ function runCSharpBenchmark(){
 function flushDB() {
   cd $utilitiesDir
   npm install
-  npm run flush -- --host $host $tlsFlag
+  npm run flush -- --host $host $tlsFlag $clusterFlag
 }
 
 function fillDB(){
   flushDB
   cd $utilitiesDir
-  npm run fill -- --dataSize $1 --host $host $tlsFlag
+  npm run fill -- --dataSize $1 --host $host $tlsFlag $clusterFlag
 }
 
 utilitiesDir=`pwd`/utilities
@@ -117,6 +117,7 @@ function Help() {
     echo Pass -only-ffi to only run Babushka FFI based clients.
     echo Pass -only-socket to only run Babushka socket based clients.
     echo Pass -only-babushka to only run Babushk clients.
+    echo Pass -is-cluster if the host is a CME server.
     echo The benchmark will connect to the server using transport level security \(TLS\) by default. Pass -no-tls to connect to server without TLS.
     echo By default, the benchmark runs against localhost. Pass -host and then the address of the requested Redis server in order to connect to a different server.
 }
@@ -185,6 +186,9 @@ do
         -no-tls) 
             tlsFlag=
             ;;
+        -is-cluster) 
+            clusterFlag="--clusterModeEnabled"
+            ;;            
     esac
     shift
 done

benchmarks/node/node_benchmark.ts

@@ -1,9 +1,14 @@
-import percentile from "percentile";
-import { stdev } from "stats-lite";
-import { createClient } from "redis";
-import { AsyncClient, SocketConnection, setLoggerConfig } from "babushka-rs";
+import {
+    AsyncClient,
+    ClusterSocketConnection,
+    SocketConnection,
+    setLoggerConfig,
+} from "babushka-rs";
 import commandLineArgs from "command-line-args";
 import { writeFileSync } from "fs";
+import percentile from "percentile";
+import { RedisClientType, createClient, createCluster } from "redis";
+import { stdev } from "stats-lite";
 
 enum ChosenAction {
     GET_NON_EXISTING,
@@ -13,8 +18,8 @@ enum ChosenAction {
 // Demo - Setting the internal logger to log every log that has a level of info and above, and save the logs to the first.log file.
 setLoggerConfig("info", "first.log");
 
+const PORT = 6379;
 function getAddress(host: string, port?: number): string {
-    const PORT = 6379;
     return `${host}:${port === undefined ? PORT : port}`;
 }
 
@@ -23,7 +28,6 @@ function getAddressWithProtocol(
     useTLS: boolean,
     port?: number
 ): string {
-    const PORT = 6379;
     const protocol = useTLS ? "rediss" : "redis";
     return `${protocol}://${getAddress(host, port ?? PORT)}`;
 }
@@ -144,13 +148,15 @@ async function run_clients(
     total_commands: number,
     num_of_concurrent_tasks: number,
     data_size: number,
-    data: string
+    data: string,
+    clientDisposal: (client: IAsyncClient) => void,
+    is_cluster: boolean
 ) {
     const now = new Date();
     console.log(
         `Starting ${client_name} data size: ${data_size} concurrency: ${num_of_concurrent_tasks} client count: ${
             clients.length
-        } ${now.toLocaleTimeString()}`
+        } is_cluster: ${is_cluster} ${now.toLocaleTimeString()}`
     );
     const action_latencies = {
         [ChosenAction.SET]: [],
@@ -189,11 +195,14 @@ async function run_clients(
         data_size,
         tps,
         clientCount: clients.length,
+        is_cluster,
         ...set_latency_results,
         ...get_existing_latency_results,
         ...get_non_existing_latency_results,
     };
     bench_json_results.push(json_res);
+
+    Promise.all(clients.map((client) => clientDisposal(client)));
 }
 
 function createClients(
@@ -213,32 +222,34 @@ async function main(
     clients_to_run: "all" | "ffi" | "socket" | "babushka",
     host: string,
     clientCount: number,
-    useTLS: boolean
+    useTLS: boolean,
+    clusterModeEnabled: boolean
 ) {
     const data = generate_value(data_size);
     if (
-        clients_to_run == "ffi" ||
-        clients_to_run == "all" ||
-        clients_to_run == "babushka"
+        !clusterModeEnabled && // no FFI support for CME yet
+        (clients_to_run == "ffi" ||
+            clients_to_run == "all" ||
+            clients_to_run == "babushka")
     ) {
-        const clients = await createClients(
-            clientCount,
-            () =>
-                new Promise((resolve) =>
-                    resolve(
-                        AsyncClient.CreateConnection(
-                            getAddressWithProtocol(host, useTLS)
-                        )
-                    )
+        const clients = await createClients(clientCount, () =>
+            Promise.resolve(
+                AsyncClient.CreateConnection(
+                    getAddressWithProtocol(host, useTLS)
                 )
+            )
         );
         await run_clients(
             clients,
             "babushka FFI",
             total_commands,
             num_of_concurrent_tasks,
             data_size,
-            data
+            data,
+            () => {
+                /* disposed by GC */
+            },
+            clusterModeEnabled
         );
     }
 
@@ -247,8 +258,11 @@ async function main(
         clients_to_run == "all" ||
         clients_to_run == "babushka"
     ) {
+        const clientClass = clusterModeEnabled
+            ? ClusterSocketConnection
+            : SocketConnection;
         const clients = await createClients(clientCount, () =>
-            SocketConnection.CreateConnection({
+            clientClass.CreateConnection({
                 addresses: [{ host }],
                 useTLS,
             })
@@ -259,27 +273,49 @@ async function main(
             total_commands,
             num_of_concurrent_tasks,
             data_size,
-            data
+            data,
+            (client) => {
+                (client as SocketConnection).dispose();
+            },
+            clusterModeEnabled
         );
-        clients.forEach((client) => (client as SocketConnection).dispose());
         await new Promise((resolve) => setTimeout(resolve, 100));
     }
 
     if (clients_to_run == "all") {
+        const clients = await createClients(clientCount, async () => {
+            const node = {
+                url: getAddressWithProtocol(host, useTLS),
+            };
+            const node_redis_client = clusterModeEnabled
+                ? createCluster({
+                      rootNodes: [
+                          { socket: { host, port: PORT, tls: useTLS } },
+                      ],
+                      defaults: {
+                          socket: {
+                              tls: useTLS,
+                          },
+                      },
+                      useReplicas: true,
+                  })
+                : createClient(node);
+            await node_redis_client.connect();
+            return node_redis_client;
+        });
         await run_clients(
-            await createClients(clientCount, async () => {
-                const node_redis_client = createClient({
-                    url: getAddressWithProtocol(host, useTLS),
-                });
-                await node_redis_client.connect();
-                return node_redis_client;
-            }),
+            clients,
             "node_redis",
             total_commands,
             num_of_concurrent_tasks,
             data_size,
-            data
+            data,
+            (client) => {
+                (client as RedisClientType).disconnect();
+            },
+            clusterModeEnabled
         );
+        await new Promise((resolve) => setTimeout(resolve, 100));
     }
 }
 
@@ -290,7 +326,8 @@ const optionDefinitions = [
     { name: "clients", type: String },
     { name: "host", type: String },
     { name: "clientCount", type: String, multiple: true },
-    { name: "tls", type: Boolean },
+    { name: "tls", type: Boolean, defaultValue: false },
+    { name: "clusterModeEnabled", type: Boolean, defaultValue: false },
 ];
 const receivedOptions = commandLineArgs(optionDefinitions);
 
@@ -329,7 +366,8 @@ Promise.resolve() // just added to clean the indentation of the rest of the call
                 clients_to_run,
                 receivedOptions.host,
                 clientCount,
-                receivedOptions.tls
+                receivedOptions.tls,
+                receivedOptions.clusterModeEnabled
             );
         }
 

benchmarks/node/package.json

@@ -13,7 +13,7 @@
         "@types/stats-lite": "^2.2.0",
         "babushka-rs": "file:../../node",
         "command-line-args": "^5.2.1",
-        "redis": "^4.2.0",
+        "redis": "^4.6.2",
         "stats-lite": "^2.2.0",
         "percentile": "^1.6.0"
     },

benchmarks/python/python_benchmark.py

@@ -7,15 +7,16 @@
 from datetime import datetime
 from enum import Enum
 from statistics import mean
+
 import numpy as np
 import redis.asyncio as redispy
 from pybushka import (
+    AddressInfo,
     ClientConfiguration,
+    LogLevel,
     RedisAsyncFFIClient,
     RedisAsyncSocketClient,
-    LogLevel,
     set_logger_config,
-    AddressInfo
 )
 
 
@@ -211,6 +212,7 @@ async def run_clients(
             "data_size": data_size,
             "tps": tps,
             "clientCount": len(clients),
+            "is_cluster": "false",
         },
         **get_existing_latency_results,
         **get_non_existing_latency_results,
@@ -256,7 +258,9 @@ async def main(
         or clients_to_run == "babushka"
     ):
         # Babushka FFI
-        config = ClientConfiguration([AddressInfo(host=host, port=PORT)], use_tls=use_tls)
+        config = ClientConfiguration(
+            [AddressInfo(host=host, port=PORT)], use_tls=use_tls
+        )
         clients = await create_clients(
             client_count,
             lambda: RedisAsyncFFIClient.create(config),
@@ -276,7 +280,9 @@ async def main(
         or clients_to_run == "babushka"
     ):
         # Babushka Socket
-        config = ClientConfiguration([AddressInfo(host=host, port=PORT)], use_tls=use_tls)
+        config = ClientConfiguration(
+            [AddressInfo(host=host, port=PORT)], use_tls=use_tls
+        )
         clients = await create_clients(
             client_count,
             lambda: RedisAsyncSocketClient.create(config),
@@ -292,7 +298,7 @@ async def main(
 
 
 def number_of_iterations(num_of_concurrent_tasks):
-    return max(100000, num_of_concurrent_tasks * 10000)
+    return min(max(100000, num_of_concurrent_tasks * 10000), 10000000)
 
 
 if __name__ == "__main__":

benchmarks/utilities/csv_exporter.py

@@ -9,6 +9,7 @@
     base_fields = [
         "language",
         "client",
+        "is_cluster",
         "num_of_tasks",
         "data_size",
         "clientCount",