aboutsummaryrefslogtreecommitdiff
path: root/src/rpc_client.rs
diff options
context:
space:
mode:
authorAlex Auvolat <alex@adnab.me>2020-04-16 23:13:15 +0200
committerAlex Auvolat <alex@adnab.me>2020-04-16 23:13:15 +0200
commit6ce14e2c9eb1ba81add3f61377a5a83854880b42 (patch)
treea920499c387af62fa98e87cf8ca942890e573487 /src/rpc_client.rs
parent4fe8329847fe5d39495e4ffda261373f46d2d547 (diff)
downloadgarage-6ce14e2c9eb1ba81add3f61377a5a83854880b42.tar.gz
garage-6ce14e2c9eb1ba81add3f61377a5a83854880b42.zip
Make all requests continue in the background even after we got enough responses.
Diffstat (limited to 'src/rpc_client.rs')
-rw-r--r--src/rpc_client.rs39
1 files changed, 25 insertions, 14 deletions
diff --git a/src/rpc_client.rs b/src/rpc_client.rs
index a1c5dde0..81d20966 100644
--- a/src/rpc_client.rs
+++ b/src/rpc_client.rs
@@ -1,6 +1,7 @@
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
+use std::borrow::Borrow;
use bytes::IntoBuf;
use futures::stream::futures_unordered::FuturesUnordered;
@@ -19,12 +20,13 @@ use crate::tls_util;
pub async fn rpc_call_many(
sys: Arc<System>,
to: &[UUID],
- msg: &Message,
+ msg: Message,
timeout: Duration,
) -> Vec<Result<Message, Error>> {
+ let msg = Arc::new(msg);
let mut resp_stream = to
.iter()
- .map(|to| rpc_call(sys.clone(), to, msg, timeout))
+ .map(|to| rpc_call(sys.clone(), to, msg.clone(), timeout))
.collect::<FuturesUnordered<_>>();
let mut results = vec![];
@@ -37,13 +39,15 @@ pub async fn rpc_call_many(
pub async fn rpc_try_call_many(
sys: Arc<System>,
to: &[UUID],
- msg: &Message,
+ msg: Message,
stop_after: usize,
timeout: Duration,
) -> Result<Vec<Message>, Error> {
- let mut resp_stream = to
- .iter()
- .map(|to| rpc_call(sys.clone(), to, msg, timeout))
+ let sys2 = sys.clone();
+ let msg = Arc::new(msg);
+ let mut resp_stream = to.to_vec()
+ .into_iter()
+ .map(move |to| rpc_call(sys2.clone(), to.clone(), msg.clone(), timeout))
.collect::<FuturesUnordered<_>>();
let mut results = vec![];
@@ -64,6 +68,13 @@ pub async fn rpc_try_call_many(
}
if results.len() >= stop_after {
+ // Continue requests in background
+ // TODO: make this optionnal (only usefull for write requests)
+ sys.background.spawn(async move {
+ resp_stream.collect::<Vec<_>>().await;
+ Ok(())
+ });
+
Ok(results)
} else {
let mut msg = "Too many failures:".to_string();
@@ -74,17 +85,17 @@ pub async fn rpc_try_call_many(
}
}
-pub async fn rpc_call(
+pub async fn rpc_call<M: Borrow<Message>, N: Borrow<UUID>>(
sys: Arc<System>,
- to: &UUID,
- msg: &Message,
+ to: N,
+ msg: M,
timeout: Duration,
) -> Result<Message, Error> {
let addr = {
let status = sys.status.borrow().clone();
- match status.nodes.get(to) {
+ match status.nodes.get(to.borrow()) {
Some(status) => status.addr.clone(),
- None => return Err(Error::Message(format!("Peer ID not found: {:?}", to))),
+ None => return Err(Error::Message(format!("Peer ID not found: {:?}", to.borrow()))),
}
};
sys.rpc_client.call(&addr, msg, timeout).await
@@ -119,10 +130,10 @@ impl RpcClient {
}
}
- pub async fn call(
+ pub async fn call<M: Borrow<Message>>(
&self,
to_addr: &SocketAddr,
- msg: &Message,
+ msg: M,
timeout: Duration,
) -> Result<Message, Error> {
let uri = match self {
@@ -133,7 +144,7 @@ impl RpcClient {
let req = Request::builder()
.method(Method::POST)
.uri(uri)
- .body(Body::from(rmp_to_vec_all_named(msg)?))?;
+ .body(Body::from(rmp_to_vec_all_named(msg.borrow())?))?;
let resp_fut = match self {
RpcClient::HTTP(client) => client.request(req).fuse(),