[5/5] Add Async MultiplexedClusterConnection

redis/Cargo.toml

@@ -41,14 +41,14 @@ tokio = { version = "1", features = ["rt", "net"], optional = true }
 
 # Only needed for the connection manager
 arc-swap = { version = "1.1.0", optional = true }
-futures = { version = "0.3.3", optional = true }
 
 # Only needed for the r2d2 feature
 r2d2 = { version = "0.8.8", optional = true }
 
 # Only needed for cluster
 crc16 = { version = "0.4", optional = true }
 rand = { version = "0.8", optional = true }
+
 # Only needed for async_std support
 async-std = { version = "1.8.0", optional = true}
 async-trait = { version = "0.1.24", optional = true }
@@ -73,7 +73,7 @@ async-std-comp = ["aio", "async-std"]
 async-std-tls-comp = ["async-std-comp", "async-native-tls", "tls"]
 tokio-comp = ["aio", "tokio", "tokio/net"]
 tokio-native-tls-comp = ["tls", "tokio-native-tls"]
-connection-manager = ["arc-swap", "futures", "aio"]
+connection-manager = ["arc-swap", "aio"]
 streams = []
 
 

redis/benches/bench_cluster.rs

@@ -1,14 +1,16 @@
 #![allow(clippy::unit_arg)] // want to allow this for `black_box()`
 #![cfg(feature = "cluster")]
 use criterion::{black_box, criterion_group, criterion_main, Criterion, Throughput};
+use futures::{prelude::*, stream};
 use redis::cluster::cluster_pipe;
+use redis::RedisError;
 
 use support::*;
 
 #[path = "../tests/support/mod.rs"]
 mod support;
 
-const PIPELINE_QUERIES: usize = 100;
+const PIPELINE_QUERIES: usize = 1_000;
 
 fn bench_set_get_and_del(c: &mut Criterion, con: &mut redis::cluster::ClusterConnection) {
     let key = "test_key";
@@ -76,6 +78,123 @@ fn bench_pipeline(c: &mut Criterion, con: &mut redis::cluster::ClusterConnection
     group.finish();
 }
 
+fn bench_set_get_and_del_async(c: &mut Criterion, client: &mut redis::cluster::ClusterClient) {
+    let key = "test_key";
+
+    let mut group = c.benchmark_group("cluster_async_basic");
+    let runtime = current_thread_runtime();
+    let con = client.get_async_connection();
+    let mut con = runtime.block_on(con).unwrap();
+
+    group.bench_function("set", |b| {
+        b.iter(|| {
+            runtime
+                .block_on(async {
+                    redis::cmd("SET")
+                        .arg(key)
+                        .arg(42)
+                        .query_async(&mut con)
+                        .await?;
+                    Ok::<(), RedisError>(())
+                })
+                .unwrap()
+        })
+    });
+
+    group.bench_function("get", |b| {
+        b.iter(|| {
+            runtime
+                .block_on(async {
+                    redis::cmd("GET").arg(key).query_async(&mut con).await?;
+                    Ok::<(), RedisError>(())
+                })
+                .unwrap()
+        })
+    });
+
+    group.bench_function("set_and_del", |b| {
+        b.iter(|| {
+            runtime
+                .block_on(async {
+                    redis::cmd("SET")
+                        .arg(key)
+                        .arg(42)
+                        .query_async(&mut con)
+                        .await?;
+                    redis::cmd("DEL").arg(key).query_async(&mut con).await?;
+                    Ok::<(), RedisError>(())
+                })
+                .unwrap()
+        })
+    });
+
+    group.finish();
+}
+
+fn bench_pipeline_async(c: &mut Criterion, client: &mut redis::cluster::ClusterClient) {
+    let mut group = c.benchmark_group("cluster_async_pipeline");
+    let runtime = current_thread_runtime();
+    let con = client.get_async_connection();
+    let mut con = runtime.block_on(con).unwrap();
+
+    group.throughput(Throughput::Elements(PIPELINE_QUERIES as u64));
+
+    let mut queries = Vec::new();
+    for i in 0..PIPELINE_QUERIES {
+        queries.push(format!("foo{}", i));
+    }
+
+    let build_pipeline = || {
+        let mut pipe = cluster_pipe();
+        for q in &queries {
+            pipe.set(q, "bar").ignore();
+        }
+    };
+    group.bench_function("build_pipeline", |b| {
+        b.iter(|| {
+            build_pipeline();
+            black_box(())
+        })
+    });
+
+    let mut pipe = cluster_pipe();
+    for q in &queries {
+        pipe.set(q, "bar").ignore();
+    }
+    group.bench_function("query_pipeline", |b| {
+        b.iter(|| {
+            runtime
+                .block_on(async { pipe.query_async::<_, ()>(&mut con).await })
+                .unwrap();
+        })
+    });
+
+    let cmds: Vec<_> = (0..PIPELINE_QUERIES)
+        .map(|i| redis::cmd("SET").arg(format!("foo{}", i)).arg(i).clone())
+        .collect();
+
+    let mut connections = (0..PIPELINE_QUERIES)
+        .map(|_| con.clone())
+        .collect::<Vec<_>>();
+
+    group.bench_function("query_implicit_pipeline", |b| {
+        b.iter(|| {
+            runtime
+                .block_on(async {
+                    cmds.iter()
+                        .zip(&mut connections)
+                        .map(|(cmd, con)| cmd.query_async(con))
+                        .collect::<stream::FuturesUnordered<_>>()
+                        .try_for_each(|()| async { Ok(()) })
+                        .await
+                })
+                .unwrap();
+        })
+    });
+
+    group.finish();
+}
+
 fn bench_cluster_setup(c: &mut Criterion) {
     let cluster = TestClusterContext::new(6, 1);
     cluster.wait_for_cluster_up();
@@ -85,6 +204,14 @@ fn bench_cluster_setup(c: &mut Criterion) {
     bench_pipeline(c, &mut con);
 }
 
+fn bench_cluster_async_setup(c: &mut Criterion) {
+    let mut cluster = TestClusterContext::new(6, 1);
+    cluster.wait_for_cluster_up();
+
+    bench_set_get_and_del_async(c, &mut cluster.client);
+    bench_pipeline_async(c, &mut cluster.client);
+}
+
 #[allow(dead_code)]
 fn bench_cluster_read_from_replicas_setup(c: &mut Criterion) {
     let cluster = TestClusterContext::new_with_cluster_client_builder(6, 1, |builder| {
@@ -97,9 +224,22 @@ fn bench_cluster_read_from_replicas_setup(c: &mut Criterion) {
     bench_pipeline(c, &mut con);
 }
 
+#[allow(dead_code)]
+fn bench_cluster_async_read_from_replicas_setup(c: &mut Criterion) {
+    let mut cluster = TestClusterContext::new_with_cluster_client_builder(6, 1, |builder| {
+        builder.read_from_replicas()
+    });
+    cluster.wait_for_cluster_up();
+
+    bench_set_get_and_del_async(c, &mut cluster.client);
+    bench_pipeline_async(c, &mut cluster.client);
+}
+
 criterion_group!(
     cluster_bench,
     bench_cluster_setup,
-    // bench_cluster_read_from_replicas_setup
+    // bench_cluster_read_from_replicas_setup,
+    bench_cluster_async_setup,
+    // bench_cluster_async_read_from_replicas_setup,
 );
 criterion_main!(cluster_bench);