forked from Deuxfleurs/garage
Make all requests continue in the background even after we got enough responses.
This commit is contained in:
parent
4fe8329847
commit
6ce14e2c9e
5 changed files with 31 additions and 19 deletions
|
@ -219,7 +219,7 @@ async fn put_block(garage: Arc<Garage>, hash: Hash, data: Vec<u8>) -> Result<(),
|
|||
rpc_try_call_many(
|
||||
garage.system.clone(),
|
||||
&who[..],
|
||||
&Message::PutBlock(PutBlockMessage { hash, data }),
|
||||
Message::PutBlock(PutBlockMessage { hash, data }),
|
||||
(garage.system.config.data_replication_factor + 1) / 2,
|
||||
BLOCK_RW_TIMEOUT,
|
||||
)
|
||||
|
@ -366,7 +366,7 @@ async fn get_block(garage: Arc<Garage>, hash: &Hash) -> Result<Vec<u8>, Error> {
|
|||
let resps = rpc_try_call_many(
|
||||
garage.system.clone(),
|
||||
&who[..],
|
||||
&Message::GetBlock(hash.clone()),
|
||||
Message::GetBlock(hash.clone()),
|
||||
1,
|
||||
BLOCK_RW_TIMEOUT,
|
||||
)
|
||||
|
|
|
@ -270,7 +270,7 @@ impl System {
|
|||
.filter(|x| **x != self.id)
|
||||
.cloned()
|
||||
.collect::<Vec<_>>();
|
||||
rpc_call_many(self.clone(), &to[..], &msg, timeout).await;
|
||||
rpc_call_many(self.clone(), &to[..], msg, timeout).await;
|
||||
}
|
||||
|
||||
pub async fn bootstrap(self: Arc<Self>) {
|
||||
|
|
|
@ -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(),
|
||||
|
|
|
@ -12,7 +12,7 @@ use tokio::net::{TcpListener, TcpStream};
|
|||
use tokio_rustls::server::TlsStream;
|
||||
use tokio_rustls::TlsAcceptor;
|
||||
|
||||
use crate::data::{rmp_to_vec_all_named, debug_serialize};
|
||||
use crate::data::*;
|
||||
use crate::error::Error;
|
||||
use crate::proto::Message;
|
||||
use crate::server::Garage;
|
||||
|
|
|
@ -280,7 +280,7 @@ impl<F: TableSchema + 'static> Table<F> {
|
|||
let resps = rpc_try_call_many(
|
||||
self.system.clone(),
|
||||
who,
|
||||
&rpc_msg,
|
||||
rpc_msg,
|
||||
quorum,
|
||||
self.param.timeout,
|
||||
)
|
||||
|
@ -384,6 +384,7 @@ impl<F: TableSchema + 'static> Table<F> {
|
|||
}
|
||||
|
||||
pub async fn delete_range(&self, begin: &Hash, end: &Hash) -> Result<(), Error> {
|
||||
eprintln!("({}) Deleting range {:?} - {:?}", self.name, begin, end);
|
||||
// TODO
|
||||
Ok(())
|
||||
}
|
||||
|
|
Loading…
Reference in a new issue