Python POC: added async API using CFFI

Author: barshaul

benchmarks/python/all_async_100

@@ -0,0 +1 @@
+[{"client": "redispy", "loop": "asyncio", "num_of_tasks": 10, "data_size": 100, "tps": 12261, "client_count": 1, "is_cluster": true, "get_existing_p50_latency": 0.757, "get_existing_p90_latency": 0.851, "get_existing_p99_latency": 0.69, "get_existing_average_latency": 0.798, "get_existing_std_dev": 3.012, "get_non_existing_p50_latency": 0.742, "get_non_existing_p90_latency": 0.837, "get_non_existing_p99_latency": 0.679, "get_non_existing_average_latency": 0.757, "get_non_existing_std_dev": 0.145, "set_p50_latency": 0.755, "set_p90_latency": 0.848, "set_p99_latency": 0.687, "set_average_latency": 0.813, "set_std_dev": 4.049}, {"client": "glide_socket", "loop": "asyncio", "num_of_tasks": 10, "data_size": 100, "tps": 14454, "client_count": 1, "is_cluster": true, "get_existing_p50_latency": 0.601, "get_existing_p90_latency": 0.92, "get_existing_p99_latency": 0.538, "get_existing_average_latency": 0.689, "get_existing_std_dev": 0.27, "get_non_existing_p50_latency": 0.596, "get_non_existing_p90_latency": 0.916, "get_non_existing_p99_latency": 0.534, "get_non_existing_average_latency": 0.686, "get_non_existing_std_dev": 0.298, "set_p50_latency": 0.598, "set_p90_latency": 0.916, "set_p99_latency": 0.537, "set_average_latency": 0.686, "set_std_dev": 0.295}, {"client": "glide_ffi", "loop": "asyncio", "num_of_tasks": 10, "data_size": 100, "tps": 13885, "client_count": 1, "is_cluster": true, "get_existing_p50_latency": 0.704, "get_existing_p90_latency": 0.89, "get_existing_p99_latency": 0.512, "get_existing_average_latency": 0.721, "get_existing_std_dev": 0.427, "get_non_existing_p50_latency": 0.69, "get_non_existing_p90_latency": 0.872, "get_non_existing_p99_latency": 0.494, "get_non_existing_average_latency": 0.706, "get_non_existing_std_dev": 0.532, "set_p50_latency": 0.697, "set_p90_latency": 0.879, "set_p99_latency": 0.5, "set_average_latency": 0.711, "set_std_dev": 0.378}, {"client": "redispy", "loop": "asyncio", "num_of_tasks": 100, "data_size": 100, "tps": 8658, "client_count": 1, "is_cluster": true, "get_existing_p50_latency": 6.738, "get_existing_p90_latency": 6.858, "get_existing_p99_latency": 6.635, "get_existing_average_latency": 9.442, "get_existing_std_dev": 93.62, "get_non_existing_p50_latency": 6.729, "get_non_existing_p90_latency": 6.846, "get_non_existing_p99_latency": 6.626, "get_non_existing_average_latency": 7.898, "get_non_existing_std_dev": 60.307, "set_p50_latency": 6.738, "set_p90_latency": 6.856, "set_p99_latency": 6.634, "set_average_latency": 10.012, "set_std_dev": 101.823}, {"client": "glide_socket", "loop": "asyncio", "num_of_tasks": 100, "data_size": 100, "tps": 24707, "client_count": 1, "is_cluster": true, "get_existing_p50_latency": 3.972, "get_existing_p90_latency": 4.138, "get_existing_p99_latency": 3.923, "get_existing_average_latency": 4.042, "get_existing_std_dev": 0.513, "get_non_existing_p50_latency": 3.971, "get_non_existing_p90_latency": 4.149, "get_non_existing_p99_latency": 3.922, "get_non_existing_average_latency": 4.037, "get_non_existing_std_dev": 0.485, "set_p50_latency": 3.974, "set_p90_latency": 4.129, "set_p99_latency": 3.925, "set_average_latency": 4.042, "set_std_dev": 0.509}, {"client": "glide_ffi", "loop": "asyncio", "num_of_tasks": 100, "data_size": 100, "tps": 13857, "client_count": 1, "is_cluster": true, "get_existing_p50_latency": 7.273, "get_existing_p90_latency": 7.79, "get_existing_p99_latency": 6.2921, "get_existing_average_latency": 7.215, "get_existing_std_dev": 2.779, "get_non_existing_p50_latency": 7.269, "get_non_existing_p90_latency": 7.791, "get_non_existing_p99_latency": 6.2847, "get_non_existing_average_latency": 7.217, "get_non_existing_std_dev": 2.738, "set_p50_latency": 7.275, "set_p90_latency": 7.774, "set_p99_latency": 6.297, "set_average_latency": 7.167, "set_std_dev": 2.257}, {"client": "redispy", "loop": "asyncio", "num_of_tasks": 1000, "data_size": 100, "tps": 1730, "client_count": 1, "is_cluster": true, "get_existing_p50_latency": 91.572, "get_existing_p90_latency": 92.46, "get_existing_p99_latency": 90.881, "get_existing_average_latency": 349.823, "get_existing_std_dev": 2868.364, "get_non_existing_p50_latency": 91.556, "get_non_existing_p90_latency": 92.4182, "get_non_existing_p99_latency": 90.865, "get_non_existing_average_latency": 336.043, "get_non_existing_std_dev": 2829.902, "set_p50_latency": 91.577, "set_p90_latency": 92.475, "set_p99_latency": 90.89, "set_average_latency": 348.878, "set_std_dev": 2900.974}, {"client": "glide_socket", "loop": "asyncio", "num_of_tasks": 1000, "data_size": 100, "tps": 17042, "client_count": 1, "is_cluster": true, "get_existing_p50_latency": 60.046, "get_existing_p90_latency": 61.8933, "get_existing_p99_latency": 41.5114, "get_existing_average_latency": 58.463, "get_existing_std_dev": 10.693, "get_non_existing_p50_latency": 60.04, "get_non_existing_p90_latency": 61.867, "get_non_existing_p99_latency": 41.516, "get_non_existing_average_latency": 58.406, "get_non_existing_std_dev": 10.671, "set_p50_latency": 60.093, "set_p90_latency": 61.798, "set_p99_latency": 41.5268, "set_average_latency": 58.358, "set_std_dev": 10.445}, {"client": "glide_ffi", "loop": "asyncio", "num_of_tasks": 1000, "data_size": 100, "tps": 12173, "client_count": 1, "is_cluster": true, "get_existing_p50_latency": 81.869, "get_existing_p90_latency": 85.655, "get_existing_p99_latency": 74.679, "get_existing_average_latency": 81.529, "get_existing_std_dev": 7.444, "get_non_existing_p50_latency": 81.87, "get_non_existing_p90_latency": 85.671, "get_non_existing_p99_latency": 74.5847, "get_non_existing_average_latency": 81.507, "get_non_existing_std_dev": 7.602, "set_p50_latency": 81.885, "set_p90_latency": 85.669, "set_p99_latency": 74.5914, "set_average_latency": 81.529, "set_std_dev": 7.391}]

benchmarks/python/python_benchmark.py

@@ -18,6 +18,7 @@
 from glide import (
     GlideClientConfiguration,
     GlideClusterClientConfiguration,
+    GlideAsync,
     GlideClient,
     GlideClusterClient,
     Logger,
@@ -197,7 +198,12 @@ def latency_results(prefix, latencies):
 
 
 async def create_clients(client_count, action):
-    return [await action() for _ in range(client_count)]
+    try:
+        return [await action() for _ in range(client_count)]
+    except Exception as e:
+        print(e)
+        return [action() for _ in range(client_count)]
+
 
 
 async def run_clients(
@@ -270,7 +276,7 @@ async def main(
         clients = await create_clients(
             client_count,
             lambda: client_class(
-                host=host, port=port, decode_responses=True, ssl=use_tls
+                host=host, port=port, decode_responses=False, ssl=use_tls
             ),
         )
 
@@ -285,7 +291,10 @@ async def main(
         )
 
         for client in clients:
-            await client.aclose()
+            try:
+                await client.aclose()
+            except Exception:
+                await client.close()
 
     if clients_to_run == "all" or clients_to_run == "glide":
         # Glide Socket
@@ -301,7 +310,28 @@ async def main(
         )
         await run_clients(
             clients,
-            "glide",
+            "glide_socket",
+            event_loop_name,
+            total_commands,
+            num_of_concurrent_tasks,
+            data_size,
+            is_cluster,
+        )
+    if clients_to_run == "all" or clients_to_run == "glide":
+        # Glide Socket
+        client_class = GlideAsync
+        config = GlideClusterClientConfiguration(
+            [NodeAddress(host=host, port=port)], use_tls=use_tls
+        ) if is_cluster else GlideClientConfiguration(
+            [NodeAddress(host=host, port=port)], use_tls=use_tls
+        )
+        clients = await create_clients(
+            client_count,
+            lambda: client_class(config),
+        )
+        await run_clients(
+            clients,
+            "glide_ffi",
             event_loop_name,
             total_commands,
             num_of_concurrent_tasks,
@@ -311,6 +341,7 @@ async def main(
 
 
 def number_of_iterations(num_of_concurrent_tasks):
+    return 100000
     return min(max(100000, num_of_concurrent_tasks * 10000), 5000000)
 
 

benchmarks/python/python_sync_benchmark.py

@@ -2,14 +2,19 @@
 
 #![deny(unsafe_op_in_unsafe_fn)]
 use glide_core::client::Client as GlideClient;
+use glide_core::command_request::SimpleRoutes;
+use glide_core::command_request::{Routes, SlotTypes};
 use glide_core::connection_request;
 use glide_core::errors;
 use glide_core::errors::RequestErrorType;
 use glide_core::request_type::RequestType;
 use glide_core::ConnectionRequest;
-use glide_core::client::{NodeAddress, TlsMode};
 use protobuf::Message;
-use redis::{RedisResult, Value};
+use redis::cluster_routing::{
+    MultipleNodeRoutingInfo, Route, RoutingInfo, SingleNodeRoutingInfo, SlotAddr,
+};
+use redis::cluster_routing::{ResponsePolicy, Routable};
+use redis::{Cmd, RedisResult, Value};
 use std::slice::from_raw_parts;
 use std::{
     ffi::{c_void, CString},
@@ -113,27 +118,34 @@ pub type FailureCallback = unsafe extern "C" fn(
     error_type: RequestErrorType,
 ) -> ();
 
-    /// The connection response.
-    ///
-    /// It contains either a connection or an error. It is represented as a struct instead of a union for ease of use in the wrapper language.
-    ///
-    /// The struct is freed by the external caller by using `free_connection_response` to avoid memory leaks.
-    #[repr(C)]
-    pub struct ConnectionResponse {
-        conn_ptr: *const c_void,
-        connection_error_message: *const c_char,
-    }
+/// The connection response.
+///
+/// It contains either a connection or an error. It is represented as a struct instead of a union for ease of use in the wrapper language.
+///
+/// The struct is freed by the external caller by using `free_connection_response` to avoid memory leaks.
+#[repr(C)]
+pub struct ConnectionResponse {
+    conn_ptr: *const c_void,
+    connection_error_message: *const c_char,
+}
 
 /// A `GlideClient` adapter.
 // TODO: Remove allow(dead_code) once connection logic is implemented
 #[allow(dead_code)]
 pub struct ClientAdapter {
     client: GlideClient,
+    success_callback: SuccessCallback,
+    failure_callback: FailureCallback,
     runtime: Runtime,
 }
 
 fn create_client_internal(
+    connection_request_bytes: &[u8],
+    success_callback: SuccessCallback,
+    failure_callback: FailureCallback,
 ) -> Result<ClientAdapter, String> {
+    let request = connection_request::ConnectionRequest::parse_from_bytes(connection_request_bytes)
+        .map_err(|err| err.to_string())?;
     // TODO: optimize this using multiple threads instead of a single worker thread (e.g. by pinning each go thread to a rust thread)
     let runtime = Builder::new_multi_thread()
         .enable_all()
@@ -144,17 +156,13 @@ fn create_client_internal(
             let redis_error = err.into();
             errors::error_message(&redis_error)
         })?;
-    let addresses = vec![NodeAddress { host: "localhost".to_string(), port: 6379}];
-    let use_tls = true;
     let client = runtime
-        .block_on(GlideClient::new(ConnectionRequest {addresses, tls_mode: if use_tls {
-            Some(TlsMode::SecureTls)
-        } else {
-            Some(TlsMode::NoTls)
-        }, cluster_mode_enabled: true, ..Default::default()}, None))
+        .block_on(GlideClient::new(ConnectionRequest::from(request), None))
         .map_err(|err| err.to_string())?;
     Ok(ClientAdapter {
         client,
+        success_callback,
+        failure_callback,
         runtime,
     })
 }
@@ -177,10 +185,15 @@ fn create_client_internal(
 /// * Both the `success_callback` and `failure_callback` function pointers need to live while the client is open/active. The caller is responsible for freeing both callbacks.
 // TODO: Consider making this async
 #[no_mangle]
-pub unsafe extern "C" fn create_client() -> *const ConnectionResponse {
-    // let request_bytes =
-    //     unsafe { std::slice::from_raw_parts(connection_request_bytes, connection_request_len) };
-    let response = match create_client_internal() {
+pub unsafe extern "C" fn create_client(
+    connection_request_bytes: *const u8,
+    connection_request_len: usize,
+    success_callback: SuccessCallback,
+    failure_callback: FailureCallback,
+) -> *const ConnectionResponse {
+    let request_bytes =
+        unsafe { std::slice::from_raw_parts(connection_request_bytes, connection_request_len) };
+    let response = match create_client_internal(request_bytes, success_callback, failure_callback) {
         Err(err) => ConnectionResponse {
             conn_ptr: std::ptr::null(),
             connection_error_message: CString::into_raw(
@@ -508,46 +521,131 @@ pub unsafe extern "C" fn command(
     arg_count: c_ulong,
     args: *const usize,
     args_len: *const c_ulong,
-) -> *mut CommandResponse {
+    route_bytes: *const u8,
+    route_bytes_len: usize,
+) {
     let client_adapter =
         unsafe { Box::leak(Box::from_raw(client_adapter_ptr as *mut ClientAdapter)) };
+    // The safety of this needs to be ensured by the calling code. Cannot dispose of the pointer before
+    // all operations have completed.
+    let ptr_address = client_adapter_ptr as usize;
 
-    // Ensure the arguments are converted properly
     let arg_vec =
         unsafe { convert_double_pointer_to_vec(args as *const *const c_void, arg_count, args_len) };
 
     let mut client_clone = client_adapter.client.clone();
 
     // Create the command outside of the task to ensure that the command arguments passed
-    // from the caller are still valid
+    // from "go" are still valid
     let mut cmd = command_type
         .get_command()
         .expect("Couldn't fetch command type");
-
     for command_arg in arg_vec {
         cmd.arg(command_arg);
     }
 
-    // Block on the async task to execute the command
-    let result = client_adapter.runtime.block_on(async move {
-        client_clone.send_command(&cmd, None).await
-    });
+    let r_bytes = unsafe { std::slice::from_raw_parts(route_bytes, route_bytes_len) };
+
+    let route = Routes::parse_from_bytes(r_bytes).unwrap();
+
+    client_adapter.runtime.spawn(async move {
+        let result = client_clone
+            .send_command(&cmd, get_route(route, Some(&cmd)))
+            .await;
+        let client_adapter = unsafe { Box::leak(Box::from_raw(ptr_address as *mut ClientAdapter)) };
+        let value = match result {
+            Ok(value) => value,
+            Err(err) => {
+                let message = errors::error_message(&err);
+                let error_type = errors::error_type(&err);
+
+                let c_err_str = CString::into_raw(
+                    CString::new(message).expect("Couldn't convert error message to CString"),
+                );
+                unsafe { (client_adapter.failure_callback)(channel, c_err_str, error_type) };
+                return;
+            }
+        };
+
+        let result: RedisResult<CommandResponse> = valkey_value_to_command_response(value);
 
-    match result {
-        Ok(value) => {
-            // Convert the value to a CommandResponse
-            match valkey_value_to_command_response(value) {
-                Ok(command_response) => Box::into_raw(Box::new(command_response)), // Return a pointer to the CommandResponse
+        unsafe {
+            match result {
+                Ok(message) => {
+                    (client_adapter.success_callback)(channel, Box::into_raw(Box::new(message)))
+                }
                 Err(err) => {
-                    eprintln!("Error converting value to CommandResponse: {:?}", err);
-                    std::ptr::null_mut()
+                    let message = errors::error_message(&err);
+                    let error_type = errors::error_type(&err);
+
+                    let c_err_str = CString::into_raw(
+                        CString::new(message).expect("Couldn't convert error message to CString"),
+                    );
+                    (client_adapter.failure_callback)(channel, c_err_str, error_type);
                 }
-            }
+            };
         }
-        Err(err) => {
-            // Handle the error case
-            eprintln!("Error executing command: {:?}", err);
-            std::ptr::null_mut()
+    });
+}
+
+fn get_route(route: Routes, cmd: Option<&Cmd>) -> Option<RoutingInfo> {
+    use glide_core::command_request::routes::Value;
+    let route = route.value?;
+    let get_response_policy = |cmd: Option<&Cmd>| {
+        cmd.and_then(|cmd| {
+            cmd.command()
+                .and_then(|cmd| ResponsePolicy::for_command(&cmd))
+        })
+    };
+    match route {
+        Value::SimpleRoutes(simple_route) => {
+            let simple_route = simple_route.enum_value().unwrap();
+            match simple_route {
+                SimpleRoutes::AllNodes => Some(RoutingInfo::MultiNode((
+                    MultipleNodeRoutingInfo::AllNodes,
+                    get_response_policy(cmd),
+                ))),
+                SimpleRoutes::AllPrimaries => Some(RoutingInfo::MultiNode((
+                    MultipleNodeRoutingInfo::AllMasters,
+                    get_response_policy(cmd),
+                ))),
+                SimpleRoutes::Random => {
+                    Some(RoutingInfo::SingleNode(SingleNodeRoutingInfo::Random))
+                }
+            }
         }
+        Value::SlotKeyRoute(slot_key_route) => Some(RoutingInfo::SingleNode(
+            SingleNodeRoutingInfo::SpecificNode(Route::new(
+                redis::cluster_topology::get_slot(slot_key_route.slot_key.as_bytes()),
+                get_slot_addr(&slot_key_route.slot_type),
+            )),
+        )),
+        Value::SlotIdRoute(slot_id_route) => Some(RoutingInfo::SingleNode(
+            SingleNodeRoutingInfo::SpecificNode(Route::new(
+                slot_id_route.slot_id as u16,
+                get_slot_addr(&slot_id_route.slot_type),
+            )),
+        )),
+        Value::ByAddressRoute(by_address_route) => match u16::try_from(by_address_route.port) {
+            Ok(port) => Some(RoutingInfo::SingleNode(SingleNodeRoutingInfo::ByAddress {
+                host: by_address_route.host.to_string(),
+                port,
+            })),
+            Err(_) => {
+                // TODO: Handle error propagation.
+                None
+            }
+        },
+        _ => panic!("unknown route type"),
     }
 }
+
+fn get_slot_addr(slot_type: &protobuf::EnumOrUnknown<SlotTypes>) -> SlotAddr {
+    slot_type
+        .enum_value()
+        .map(|slot_type| match slot_type {
+            SlotTypes::Primary => SlotAddr::Master,
+            SlotTypes::Replica => SlotAddr::ReplicaRequired,
+        })
+        .expect("Received unexpected slot id type")
+}