Refactoring: rename config files, make modifications less invasive

1 files changed, 151 insertions, 137 deletions

diff --git a/src/rpc/rpc_helper.rs b/src/rpc/rpc_helper.rs
index 099c6429..4114724f 100644
--- a/src/rpc/rpc_helper.rs
+++ b/src/rpc/rpc_helper.rs
@@ -238,154 +238,168 @@ impl RpcHelper {
 		span.set_attribute(KeyValue::new("to", format!("{:?}", to)));
 		span.set_attribute(KeyValue::new("quorum", quorum as i64));
 
-		async {
-			let msg = Arc::new(msg);
-
-			// 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 self2 = self.clone();
-				let msg = msg.clone();
-				let endpoint2 = endpoint.clone();
-				(to, async move {
-					self2.call_arc(&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.
-
-				// Retrieve some status variables that we will use to sort requests
-				let peer_list = self.0.fullmesh.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 => "",
-				};
-
-				// Augment requests with some information used to sort them.
-				// The tuples are as follows:
-				//         (is another node?, is another zone?, latency, node ID, request future)
-				// We store all of these tuples in a vec that we can sort.
-				// 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 requests = requests
-					.map(|(to, fut)| {
-						let peer_zone = match ring.layout.node_role(&to) {
-							Some(pc) => &pc.zone,
-							None => "",
-						};
-						let peer_avg_ping = peer_list
-							.iter()
-							.find(|x| x.id.as_ref() == to.as_slice())
-							.map(|pi| pi.avg_ping)
-							.flatten()
-							.unwrap_or_else(|| Duration::from_secs(1));
-						(
-							to != self.0.our_node_id,
-							peer_zone != our_zone,
-							peer_avg_ping,
-							to,
-							fut,
-						)
-					})
-					.collect::<Vec<_>>();
-
-				// Sort requests by (priorize ourself, priorize same zone, priorize low latency)
-				requests.sort_by_key(|(diffnode, diffzone, ping, _to, _fut)| {
-					(*diffnode, *diffzone, *ping)
-				});
+		self.try_call_many_internal(endpoint, to, msg, strategy, quorum)
+			.with_context(Context::current_with_span(span))
+			.await
+	}
 
-				// Make an iterator to take requests in their sorted order
-				let mut requests = requests.into_iter();
-
-				// 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 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;
-						}
+	async fn try_call_many_internal<M, H, S>(
+		&self,
+		endpoint: &Arc<Endpoint<M, H>>,
+		to: &[Uuid],
+		msg: M,
+		strategy: RequestStrategy,
+		quorum: usize,
+	) -> Result<Vec<S>, Error>
+	where
+		M: Rpc<Response = Result<S, Error>> + 'static,
+		H: EndpointHandler<M> + 'static,
+		S: Send + 'static,
+	{
+		let msg = Arc::new(msg);
+
+		// 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 self2 = self.clone();
+			let msg = msg.clone();
+			let endpoint2 = endpoint.clone();
+			(to, async move {
+				self2.call_arc(&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.
+
+			// Retrieve some status variables that we will use to sort requests
+			let peer_list = self.0.fullmesh.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 => "",
+			};
+
+			// Augment requests with some information used to sort them.
+			// The tuples are as follows:
+			//         (is another node?, is another zone?, latency, node ID, request future)
+			// We store all of these tuples in a vec that we can sort.
+			// 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 requests = requests
+				.map(|(to, fut)| {
+					let peer_zone = match ring.layout.node_role(&to) {
+						Some(pc) => &pc.zone,
+						None => "",
+					};
+					let peer_avg_ping = peer_list
+						.iter()
+						.find(|x| x.id.as_ref() == to.as_slice())
+						.map(|pi| pi.avg_ping)
+						.flatten()
+						.unwrap_or_else(|| Duration::from_secs(1));
+					(
+						to != self.0.our_node_id,
+						peer_zone != our_zone,
+						peer_avg_ping,
+						to,
+						fut,
+					)
+				})
+				.collect::<Vec<_>>();
+
+			// Sort requests by (priorize ourself, priorize same zone, priorize low latency)
+			requests
+				.sort_by_key(|(diffnode, diffzone, ping, _to, _fut)| (*diffnode, *diffzone, *ping));
+
+			// Make an iterator to take requests in their sorted order
+			let mut requests = requests.into_iter();
+
+			// 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;
 					}
-					assert!(!resp_stream.is_empty()); // because of loop invariants
+				}
+				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);
-						}
+				// Wait for one request to terminate
+				match resp_stream.next().await.unwrap().unwrap() {
+					Ok(msg) => {
+						successes.push(msg);
+					}
+					Err(e) => {
+						errors.push(e);
 					}
 				}
-			} 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);
+			}
+		} 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;
 						}
 					}
-				}
-
-				if !resp_stream.is_empty() {
-					// Continue remaining requests in background.
-					// Continue the remaining requests immediately using tokio::spawn
-					// but enqueue a task in the background runner
-					// to ensure that the process won't exit until the requests are done
-					// (if we had just enqueued the resp_stream.collect directly in the background runner,
-					// the requests might have been put on hold in the background runner's queue,
-					// in which case they might timeout or otherwise fail)
-					let wait_finished_fut = tokio::spawn(async move {
-						resp_stream.collect::<Vec<Result<_, _>>>().await;
-					});
-					self.0.background.spawn(wait_finished_fut.map(|_| Ok(())));
+					Err(e) => {
+						errors.push(e);
+					}
 				}
 			}
 
-			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))
+			if !resp_stream.is_empty() {
+				// Continue remaining requests in background.
+				// Continue the remaining requests immediately using tokio::spawn
+				// but enqueue a task in the background runner
+				// to ensure that the process won't exit until the requests are done
+				// (if we had just enqueued the resp_stream.collect directly in the background runner,
+				// the requests might have been put on hold in the background runner's queue,
+				// in which case they might timeout or otherwise fail)
+				let wait_finished_fut = tokio::spawn(async move {
+					resp_stream.collect::<Vec<Result<_, _>>>().await;
+				});
+				self.0.background.spawn(wait_finished_fut.map(|_| Ok(())));
 			}
 		}
-		.with_context(Context::current_with_span(span))
-		.await
+
+		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))
+		}
 	}
 }