aboutsummaryrefslogtreecommitdiff
path: root/src/rpc/rpc_helper.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/rpc/rpc_helper.rs')
-rw-r--r--src/rpc/rpc_helper.rs492
1 files changed, 361 insertions, 131 deletions
diff --git a/src/rpc/rpc_helper.rs b/src/rpc/rpc_helper.rs
index c46e577f..977c6ed8 100644
--- a/src/rpc/rpc_helper.rs
+++ b/src/rpc/rpc_helper.rs
@@ -1,12 +1,12 @@
//! Contain structs related to making RPCs
-use std::sync::Arc;
+use std::collections::HashMap;
+use std::sync::{Arc, RwLock};
use std::time::Duration;
use futures::future::join_all;
use futures::stream::futures_unordered::FuturesUnordered;
use futures::stream::StreamExt;
use tokio::select;
-use tokio::sync::watch;
use opentelemetry::KeyValue;
use opentelemetry::{
@@ -26,8 +26,8 @@ use garage_util::data::*;
use garage_util::error::Error;
use garage_util::metrics::RecordDuration;
+use crate::layout::{LayoutHelper, LayoutHistory};
use crate::metrics::RpcMetrics;
-use crate::ring::Ring;
// Default RPC timeout = 5 minutes
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(300);
@@ -36,11 +36,11 @@ const DEFAULT_TIMEOUT: Duration = Duration::from_secs(300);
#[derive(Copy, Clone)]
pub struct RequestStrategy {
/// Min number of response to consider the request successful
- pub rs_quorum: Option<usize>,
- /// Should requests be dropped after enough response are received
- pub rs_interrupt_after_quorum: bool,
+ rs_quorum: Option<usize>,
+ /// Send all requests at once
+ rs_send_all_at_once: Option<bool>,
/// Request priority
- pub rs_priority: RequestPriority,
+ rs_priority: RequestPriority,
/// Custom timeout for this request
rs_timeout: Timeout,
}
@@ -57,7 +57,7 @@ impl RequestStrategy {
pub fn with_priority(prio: RequestPriority) -> Self {
RequestStrategy {
rs_quorum: None,
- rs_interrupt_after_quorum: false,
+ rs_send_all_at_once: None,
rs_priority: prio,
rs_timeout: Timeout::Default,
}
@@ -67,10 +67,9 @@ impl RequestStrategy {
self.rs_quorum = Some(quorum);
self
}
- /// Set if requests can be dropped after quorum has been reached
- /// In general true for read requests, and false for write
- pub fn interrupt_after_quorum(mut self, interrupt: bool) -> Self {
- self.rs_interrupt_after_quorum = interrupt;
+ /// Set quorum to be reached for request
+ pub fn send_all_at_once(mut self, value: bool) -> Self {
+ self.rs_send_all_at_once = Some(value);
self
}
/// Deactivate timeout for this request
@@ -91,7 +90,7 @@ pub struct RpcHelper(Arc<RpcHelperInner>);
struct RpcHelperInner {
our_node_id: Uuid,
peering: Arc<PeeringManager>,
- ring: watch::Receiver<Arc<Ring>>,
+ layout: Arc<RwLock<LayoutHelper>>,
metrics: RpcMetrics,
rpc_timeout: Duration,
}
@@ -100,7 +99,7 @@ impl RpcHelper {
pub(crate) fn new(
our_node_id: Uuid,
peering: Arc<PeeringManager>,
- ring: watch::Receiver<Arc<Ring>>,
+ layout: Arc<RwLock<LayoutHelper>>,
rpc_timeout: Option<Duration>,
) -> Self {
let metrics = RpcMetrics::new();
@@ -108,7 +107,7 @@ impl RpcHelper {
Self(Arc::new(RpcHelperInner {
our_node_id,
peering,
- ring,
+ layout,
metrics,
rpc_timeout: rpc_timeout.unwrap_or(DEFAULT_TIMEOUT),
}))
@@ -130,6 +129,12 @@ impl RpcHelper {
N: IntoReq<M> + Send,
H: StreamingEndpointHandler<M>,
{
+ let tracer = opentelemetry::global::tracer("garage");
+ let span_name = format!("RPC [{}] to {:?}", endpoint.path(), to);
+ let mut span = tracer.start(span_name);
+ span.set_attribute(KeyValue::new("from", format!("{:?}", self.0.our_node_id)));
+ span.set_attribute(KeyValue::new("to", format!("{:?}", to)));
+
let metric_tags = [
KeyValue::new("rpc_endpoint", endpoint.path().to_string()),
KeyValue::new("from", format!("{:?}", self.0.our_node_id)),
@@ -141,6 +146,7 @@ impl RpcHelper {
let node_id = to.into();
let rpc_call = endpoint
.call_streaming(&node_id, msg, strat.rs_priority)
+ .with_context(Context::current_with_span(span))
.record_duration(&self.0.metrics.rpc_duration, &metric_tags);
let timeout = async {
@@ -183,12 +189,17 @@ impl RpcHelper {
N: IntoReq<M>,
H: StreamingEndpointHandler<M>,
{
+ let tracer = opentelemetry::global::tracer("garage");
+ let span_name = format!("RPC [{}] call_many {} nodes", endpoint.path(), to.len());
+ let span = tracer.start(span_name);
+
let msg = msg.into_req().map_err(garage_net::error::Error::from)?;
let resps = join_all(
to.iter()
.map(|to| self.call(endpoint, *to, msg.clone(), strat)),
)
+ .with_context(Context::current_with_span(span))
.await;
Ok(to
.iter()
@@ -220,6 +231,22 @@ impl RpcHelper {
/// Make a RPC call to multiple servers, returning either a Vec of responses,
/// or an error if quorum could not be reached due to too many errors
+ ///
+ /// If RequestStrategy has send_all_at_once set, then all requests will be
+ /// sent at once, and `try_call_many` will return as soon as a quorum of
+ /// responses is achieved, dropping and cancelling the remaining requests.
+ ///
+ /// Otherwise, `quorum` requests will be sent at the same time, and if an
+ /// error response is received, a new request will be sent to replace it.
+ /// The ordering of nodes to which requests are sent is determined by
+ /// the `RpcHelper::request_order` function, which takes into account
+ /// parameters such as node zones and measured ping values.
+ ///
+ /// In both cases, the basic contract of this function is that even in the
+ /// absence of failures, the RPC call might not be driven to completion
+ /// on all of the specified nodes. It is therefore unfit for broadcast
+ /// write operations where we expect all nodes to successfully store
+ /// the written date.
pub async fn try_call_many<M, N, H, S>(
&self,
endpoint: &Arc<Endpoint<M, H>>,
@@ -236,31 +263,24 @@ impl RpcHelper {
let quorum = strategy.rs_quorum.unwrap_or(to.len());
let tracer = opentelemetry::global::tracer("garage");
- let span_name = if strategy.rs_interrupt_after_quorum {
- format!("RPC {} to {} of {}", endpoint.path(), quorum, to.len())
- } else {
- format!(
- "RPC {} to {} (quorum {})",
- endpoint.path(),
- to.len(),
- quorum
- )
- };
+ let span_name = format!(
+ "RPC [{}] try_call_many (quorum {}/{})",
+ endpoint.path(),
+ quorum,
+ to.len()
+ );
+
let mut span = tracer.start(span_name);
span.set_attribute(KeyValue::new("from", format!("{:?}", self.0.our_node_id)));
span.set_attribute(KeyValue::new("to", format!("{:?}", to)));
span.set_attribute(KeyValue::new("quorum", quorum as i64));
- span.set_attribute(KeyValue::new(
- "interrupt_after_quorum",
- strategy.rs_interrupt_after_quorum.to_string(),
- ));
- self.try_call_many_internal(endpoint, to, msg, strategy, quorum)
+ self.try_call_many_inner(endpoint, to, msg, strategy, quorum)
.with_context(Context::current_with_span(span))
.await
}
- async fn try_call_many_internal<M, N, H, S>(
+ async fn try_call_many_inner<M, N, H, S>(
&self,
endpoint: &Arc<Endpoint<M, H>>,
to: &[Uuid],
@@ -274,129 +294,238 @@ impl RpcHelper {
H: StreamingEndpointHandler<M> + 'static,
S: Send + 'static,
{
- let msg = msg.into_req().map_err(garage_net::error::Error::from)?;
+ // Once quorum is reached, other requests don't matter.
+ // What we do here is only send the required number of requests
+ // to reach a quorum, priorizing nodes with the lowest latency.
+ // When there are errors, we start new requests to compensate.
+
+ // TODO: this could be made more aggressive, e.g. if after 2x the
+ // average ping of a given request, the response is not yet received,
+ // preemptively send an additional request to any remaining nodes.
+
+ // Reorder requests to priorize closeness / low latency
+ let request_order = self.request_order(&self.0.layout.read().unwrap(), to.iter().copied());
+ let send_all_at_once = strategy.rs_send_all_at_once.unwrap_or(false);
// Build future for each request
// They are not started now: they are added below in a FuturesUnordered
// object that will take care of polling them (see below)
- let requests = to.iter().cloned().map(|to| {
+ let msg = msg.into_req().map_err(garage_net::error::Error::from)?;
+ let mut requests = request_order.into_iter().map(|to| {
let self2 = self.clone();
let msg = msg.clone();
let endpoint2 = endpoint.clone();
- (to, async move {
- self2.call(&endpoint2, to, msg, strategy).await
- })
+ async move { self2.call(&endpoint2, to, msg, strategy).await }
});
// Vectors in which success results and errors will be collected
let mut successes = vec![];
let mut errors = vec![];
- if strategy.rs_interrupt_after_quorum {
- // Case 1: once quorum is reached, other requests don't matter.
- // What we do here is only send the required number of requests
- // to reach a quorum, priorizing nodes with the lowest latency.
- // When there are errors, we start new requests to compensate.
-
- // Reorder requests to priorize closeness / low latency
- let request_order = self.request_order(to);
- let mut ord_requests = vec![(); request_order.len()]
- .into_iter()
- .map(|_| None)
- .collect::<Vec<_>>();
- for (to, fut) in requests {
- let i = request_order.iter().position(|x| *x == to).unwrap();
- ord_requests[i] = Some((to, fut));
+ // resp_stream will contain all of the requests that are currently in flight.
+ // (for the moment none, they will be added in the loop below)
+ let mut resp_stream = FuturesUnordered::new();
+
+ // Do some requests and collect results
+ while successes.len() < quorum {
+ // If the current set of requests that are running is not enough to possibly
+ // reach quorum, start some new requests.
+ while send_all_at_once || successes.len() + resp_stream.len() < quorum {
+ if let Some(fut) = requests.next() {
+ resp_stream.push(fut)
+ } else {
+ break;
+ }
+ }
+
+ if successes.len() + resp_stream.len() < quorum {
+ // We know we won't ever reach quorum
+ break;
}
- // Make an iterator to take requests in their sorted order
- let mut requests = ord_requests.into_iter().map(Option::unwrap);
-
- // resp_stream will contain all of the requests that are currently in flight.
- // (for the moment none, they will be added in the loop below)
- let mut resp_stream = FuturesUnordered::new();
-
- // Do some requests and collect results
- 'request_loop: while successes.len() < quorum {
- // If the current set of requests that are running is not enough to possibly
- // reach quorum, start some new requests.
- while successes.len() + resp_stream.len() < quorum {
- if let Some((req_to, fut)) = requests.next() {
- let tracer = opentelemetry::global::tracer("garage");
- let span = tracer.start(format!("RPC to {:?}", req_to));
- resp_stream.push(tokio::spawn(
- fut.with_context(Context::current_with_span(span)),
- ));
- } else {
- // If we have no request to add, we know that we won't ever
- // reach quorum: bail out now.
- break 'request_loop;
- }
+ // Wait for one request to terminate
+ match resp_stream.next().await.unwrap() {
+ Ok(msg) => {
+ successes.push(msg);
}
- assert!(!resp_stream.is_empty()); // because of loop invariants
-
- // Wait for one request to terminate
- match resp_stream.next().await.unwrap().unwrap() {
- Ok(msg) => {
- successes.push(msg);
- }
- Err(e) => {
- errors.push(e);
- }
+ Err(e) => {
+ errors.push(e);
}
}
+ }
+
+ if successes.len() >= quorum {
+ Ok(successes)
} else {
- // Case 2: all of the requests need to be sent in all cases,
- // and need to terminate. (this is the case for writes that
- // must be spread to n nodes)
- // Just start all the requests in parallel and return as soon
- // as the quorum is reached.
- let mut resp_stream = requests
- .map(|(_, fut)| fut)
- .collect::<FuturesUnordered<_>>();
-
- while let Some(resp) = resp_stream.next().await {
- match resp {
- Ok(msg) => {
- successes.push(msg);
- if successes.len() >= quorum {
- break;
- }
- }
- Err(e) => {
- errors.push(e);
- }
- }
- }
+ let errors = errors.iter().map(|e| format!("{}", e)).collect::<Vec<_>>();
+ Err(Error::Quorum(
+ quorum,
+ None,
+ successes.len(),
+ to.len(),
+ errors,
+ ))
+ }
+ }
+
+ /// Make a RPC call to multiple servers, returning either a Vec of responses,
+ /// or an error if quorum could not be reached due to too many errors
+ ///
+ /// Contrary to try_call_many, this fuction is especially made for broadcast
+ /// write operations. In particular:
+ ///
+ /// - The request are sent to all specified nodes as soon as `try_write_many_sets`
+ /// is invoked.
+ ///
+ /// - When `try_write_many_sets` returns, all remaining requests that haven't
+ /// completed move to a background task so that they have a chance to
+ /// complete successfully if there are no failures.
+ ///
+ /// In addition, the nodes to which requests should be sent are divided in
+ /// "quorum sets", and `try_write_many_sets` only returns once a quorum
+ /// has been validated in each set. This is used in the case of cluster layout
+ /// changes, where data has to be written both in the old layout and in the
+ /// new one as long as all nodes have not successfully tranisitionned and
+ /// moved all data to the new layout.
+ pub async fn try_write_many_sets<M, N, H, S>(
+ &self,
+ endpoint: &Arc<Endpoint<M, H>>,
+ to_sets: &[Vec<Uuid>],
+ msg: N,
+ strategy: RequestStrategy,
+ ) -> Result<Vec<S>, Error>
+ where
+ M: Rpc<Response = Result<S, Error>> + 'static,
+ N: IntoReq<M>,
+ H: StreamingEndpointHandler<M> + 'static,
+ S: Send + 'static,
+ {
+ let quorum = strategy
+ .rs_quorum
+ .expect("internal error: missing quorum value in try_write_many_sets");
+
+ let tracer = opentelemetry::global::tracer("garage");
+ let span_name = format!(
+ "RPC [{}] try_write_many_sets (quorum {} in {} sets)",
+ endpoint.path(),
+ quorum,
+ to_sets.len()
+ );
+
+ let mut span = tracer.start(span_name);
+ span.set_attribute(KeyValue::new("from", format!("{:?}", self.0.our_node_id)));
+ span.set_attribute(KeyValue::new("to", format!("{:?}", to_sets)));
+ span.set_attribute(KeyValue::new("quorum", quorum as i64));
+
+ self.try_write_many_sets_inner(endpoint, to_sets, msg, strategy, quorum)
+ .with_context(Context::current_with_span(span))
+ .await
+ }
+
+ async fn try_write_many_sets_inner<M, N, H, S>(
+ &self,
+ endpoint: &Arc<Endpoint<M, H>>,
+ to_sets: &[Vec<Uuid>],
+ msg: N,
+ strategy: RequestStrategy,
+ quorum: usize,
+ ) -> Result<Vec<S>, Error>
+ where
+ M: Rpc<Response = Result<S, Error>> + 'static,
+ N: IntoReq<M>,
+ H: StreamingEndpointHandler<M> + 'static,
+ S: Send + 'static,
+ {
+ // Peers may appear in many quorum sets. Here, build a list of peers,
+ // mapping to the index of the quorum sets in which they appear.
+ let mut result_tracker = QuorumSetResultTracker::new(to_sets, quorum);
+
+ // Send one request to each peer of the quorum sets
+ let msg = msg.into_req().map_err(garage_net::error::Error::from)?;
+ let requests = result_tracker.nodes.keys().map(|peer| {
+ let self2 = self.clone();
+ let msg = msg.clone();
+ let endpoint2 = endpoint.clone();
+ let to = *peer;
+ async move { (to, self2.call(&endpoint2, to, msg, strategy).await) }
+ });
+ let mut resp_stream = requests.collect::<FuturesUnordered<_>>();
+
+ // Drive requests to completion
+ while let Some((node, resp)) = resp_stream.next().await {
+ // Store the response in the correct vector and increment the
+ // appropriate counters
+ result_tracker.register_result(node, resp);
- if !resp_stream.is_empty() {
- // Continue remaining requests in background.
- // Note: these requests can get interrupted on process shutdown,
- // we must not count on them being executed for certain.
- // For all background things that have to happen with certainty,
- // they have to be put in a proper queue that is persisted to disk.
+ // If we have a quorum of ok in all quorum sets, then it's a success!
+ if result_tracker.all_quorums_ok() {
+ // Continue all other requets in background
tokio::spawn(async move {
- resp_stream.collect::<Vec<Result<_, _>>>().await;
+ resp_stream.collect::<Vec<(Uuid, Result<_, _>)>>().await;
});
+
+ return Ok(result_tracker.success_values());
+ }
+
+ // If there is a quorum set for which too many errors were received,
+ // we know it's impossible to get a quorum, so return immediately.
+ if result_tracker.too_many_failures() {
+ break;
}
}
- if successes.len() >= quorum {
- Ok(successes)
- } else {
- let errors = errors.iter().map(|e| format!("{}", e)).collect::<Vec<_>>();
- Err(Error::Quorum(quorum, successes.len(), to.len(), errors))
+ // At this point, there is no quorum and we know that a quorum
+ // will never be achieved. Currently, we drop all remaining requests.
+ // Should we still move them to background so that they can continue
+ // for non-failed nodes? Not doing so has no impact on correctness,
+ // but it means that more cancellation messages will be sent. Idk.
+ // (When an in-progress request future is dropped, Netapp automatically
+ // sends a cancellation message to the remote node to inform it that
+ // the result is no longer needed. In turn, if the remote node receives
+ // the cancellation message in time, it interrupts the task of the
+ // running request handler.)
+
+ // Failure, could not get quorum
+ Err(result_tracker.quorum_error())
+ }
+
+ // ---- functions not related to MAKING RPCs, but just determining to what nodes
+ // they should be made and in which order ----
+
+ pub fn block_read_nodes_of(&self, position: &Hash, rpc_helper: &RpcHelper) -> Vec<Uuid> {
+ let layout = self.0.layout.read().unwrap();
+
+ let mut ret = Vec::with_capacity(12);
+ let ver_iter = layout
+ .versions
+ .iter()
+ .rev()
+ .chain(layout.old_versions.iter().rev());
+ for ver in ver_iter {
+ if ver.version > layout.sync_map_min() {
+ continue;
+ }
+ let nodes = ver.nodes_of(position, ver.replication_factor);
+ for node in rpc_helper.request_order(&layout, nodes) {
+ if !ret.contains(&node) {
+ ret.push(node);
+ }
+ }
}
+ ret
}
- pub fn request_order(&self, nodes: &[Uuid]) -> Vec<Uuid> {
+ fn request_order(
+ &self,
+ layout: &LayoutHistory,
+ nodes: impl Iterator<Item = Uuid>,
+ ) -> Vec<Uuid> {
// Retrieve some status variables that we will use to sort requests
let peer_list = self.0.peering.get_peer_list();
- let ring: Arc<Ring> = self.0.ring.borrow().clone();
- let our_zone = match ring.layout.node_role(&self.0.our_node_id) {
- Some(pc) => &pc.zone,
- None => "",
- };
+ let our_zone = layout
+ .current()
+ .get_node_zone(&self.0.our_node_id)
+ .unwrap_or("");
// Augment requests with some information used to sort them.
// The tuples are as follows:
@@ -405,22 +534,18 @@ impl RpcHelper {
// By sorting this vec, we priorize ourself, then nodes in the same zone,
// and within a same zone we priorize nodes with the lowest latency.
let mut nodes = nodes
- .iter()
.map(|to| {
- let peer_zone = match ring.layout.node_role(to) {
- Some(pc) => &pc.zone,
- None => "",
- };
+ let peer_zone = layout.current().get_node_zone(&to).unwrap_or("");
let peer_avg_ping = peer_list
.iter()
.find(|x| x.id.as_ref() == to.as_slice())
.and_then(|pi| pi.avg_ping)
.unwrap_or_else(|| Duration::from_secs(10));
(
- *to != self.0.our_node_id,
+ to != self.0.our_node_id,
peer_zone != our_zone,
peer_avg_ping,
- *to,
+ to,
)
})
.collect::<Vec<_>>();
@@ -434,3 +559,108 @@ impl RpcHelper {
.collect::<Vec<_>>()
}
}
+
+// ------- utility for tracking successes/errors among write sets --------
+
+pub struct QuorumSetResultTracker<S, E> {
+ /// The set of nodes and the index of the quorum sets they belong to
+ pub nodes: HashMap<Uuid, Vec<usize>>,
+ /// The quorum value, i.e. number of success responses to await in each set
+ pub quorum: usize,
+
+ /// The success responses received
+ pub successes: Vec<(Uuid, S)>,
+ /// The error responses received
+ pub failures: Vec<(Uuid, E)>,
+
+ /// The counters for successes in each set
+ pub success_counters: Box<[usize]>,
+ /// The counters for failures in each set
+ pub failure_counters: Box<[usize]>,
+ /// The total number of nodes in each set
+ pub set_lens: Box<[usize]>,
+}
+
+impl<S, E> QuorumSetResultTracker<S, E>
+where
+ E: std::fmt::Display,
+{
+ pub fn new<A>(sets: &[A], quorum: usize) -> Self
+ where
+ A: AsRef<[Uuid]>,
+ {
+ let mut nodes = HashMap::<Uuid, Vec<usize>>::new();
+ for (i, set) in sets.iter().enumerate() {
+ for node in set.as_ref().iter() {
+ nodes.entry(*node).or_default().push(i);
+ }
+ }
+
+ let num_nodes = nodes.len();
+ Self {
+ nodes,
+ quorum,
+ successes: Vec::with_capacity(num_nodes),
+ failures: vec![],
+ success_counters: vec![0; sets.len()].into_boxed_slice(),
+ failure_counters: vec![0; sets.len()].into_boxed_slice(),
+ set_lens: sets
+ .iter()
+ .map(|x| x.as_ref().len())
+ .collect::<Vec<_>>()
+ .into_boxed_slice(),
+ }
+ }
+
+ pub fn register_result(&mut self, node: Uuid, result: Result<S, E>) {
+ match result {
+ Ok(s) => {
+ self.successes.push((node, s));
+ for set in self.nodes.get(&node).unwrap().iter() {
+ self.success_counters[*set] += 1;
+ }
+ }
+ Err(e) => {
+ self.failures.push((node, e));
+ for set in self.nodes.get(&node).unwrap().iter() {
+ self.failure_counters[*set] += 1;
+ }
+ }
+ }
+ }
+
+ pub fn all_quorums_ok(&self) -> bool {
+ self.success_counters
+ .iter()
+ .all(|ok_cnt| *ok_cnt >= self.quorum)
+ }
+
+ pub fn too_many_failures(&self) -> bool {
+ self.failure_counters
+ .iter()
+ .zip(self.set_lens.iter())
+ .any(|(err_cnt, set_len)| *err_cnt + self.quorum > *set_len)
+ }
+
+ pub fn success_values(self) -> Vec<S> {
+ self.successes
+ .into_iter()
+ .map(|(_, x)| x)
+ .collect::<Vec<_>>()
+ }
+
+ pub fn quorum_error(self) -> Error {
+ let errors = self
+ .failures
+ .iter()
+ .map(|(n, e)| format!("{:?}: {}", n, e))
+ .collect::<Vec<_>>();
+ Error::Quorum(
+ self.quorum,
+ Some(self.set_lens.len()),
+ self.successes.len(),
+ self.nodes.len(),
+ errors,
+ )
+ }
+}
generated by cgit v1.2.3 (git 2.25.1) at 2024-09-19 22:30:16 +0000