forked from Deuxfleurs/garage
fix clippy warnings on util and rpc
This commit is contained in:
parent
88925ebe22
commit
f05bb111c2
9 changed files with 45 additions and 51 deletions
|
@ -1,3 +1,4 @@
|
||||||
|
#![allow(clippy::upper_case_acronyms)]
|
||||||
//! Crate containing rpc related functions and types used in Garage
|
//! Crate containing rpc related functions and types used in Garage
|
||||||
|
|
||||||
#[macro_use]
|
#[macro_use]
|
||||||
|
|
|
@ -3,7 +3,7 @@ use std::collections::HashMap;
|
||||||
use std::fmt::Write as FmtWrite;
|
use std::fmt::Write as FmtWrite;
|
||||||
use std::io::{Read, Write};
|
use std::io::{Read, Write};
|
||||||
use std::net::{IpAddr, SocketAddr};
|
use std::net::{IpAddr, SocketAddr};
|
||||||
use std::path::PathBuf;
|
use std::path::{Path, PathBuf};
|
||||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
@ -198,15 +198,15 @@ impl Status {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn gen_node_id(metadata_dir: &PathBuf) -> Result<UUID, Error> {
|
fn gen_node_id(metadata_dir: &Path) -> Result<UUID, Error> {
|
||||||
let mut id_file = metadata_dir.clone();
|
let mut id_file = metadata_dir.to_path_buf();
|
||||||
id_file.push("node_id");
|
id_file.push("node_id");
|
||||||
if id_file.as_path().exists() {
|
if id_file.as_path().exists() {
|
||||||
let mut f = std::fs::File::open(id_file.as_path())?;
|
let mut f = std::fs::File::open(id_file.as_path())?;
|
||||||
let mut d = vec![];
|
let mut d = vec![];
|
||||||
f.read_to_end(&mut d)?;
|
f.read_to_end(&mut d)?;
|
||||||
if d.len() != 32 {
|
if d.len() != 32 {
|
||||||
return Err(Error::Message(format!("Corrupt node_id file")));
|
return Err(Error::Message("Corrupt node_id file".to_string()));
|
||||||
}
|
}
|
||||||
|
|
||||||
let mut id = [0u8; 32];
|
let mut id = [0u8; 32];
|
||||||
|
@ -256,7 +256,7 @@ impl System {
|
||||||
let state_info = StateInfo {
|
let state_info = StateInfo {
|
||||||
hostname: gethostname::gethostname()
|
hostname: gethostname::gethostname()
|
||||||
.into_string()
|
.into_string()
|
||||||
.unwrap_or("<invalid utf-8>".to_string()),
|
.unwrap_or_else(|_| "<invalid utf-8>".to_string()),
|
||||||
};
|
};
|
||||||
|
|
||||||
let ring = Ring::new(net_config);
|
let ring = Ring::new(net_config);
|
||||||
|
@ -296,12 +296,12 @@ impl System {
|
||||||
match msg {
|
match msg {
|
||||||
Message::Ping(ping) => self2.handle_ping(&addr, &ping).await,
|
Message::Ping(ping) => self2.handle_ping(&addr, &ping).await,
|
||||||
|
|
||||||
Message::PullStatus => self2.handle_pull_status(),
|
Message::PullStatus => Ok(self2.handle_pull_status()),
|
||||||
Message::PullConfig => self2.handle_pull_config(),
|
Message::PullConfig => Ok(self2.handle_pull_config()),
|
||||||
Message::AdvertiseNodesUp(adv) => self2.handle_advertise_nodes_up(&adv).await,
|
Message::AdvertiseNodesUp(adv) => self2.handle_advertise_nodes_up(&adv).await,
|
||||||
Message::AdvertiseConfig(adv) => self2.handle_advertise_config(&adv).await,
|
Message::AdvertiseConfig(adv) => self2.handle_advertise_config(&adv).await,
|
||||||
|
|
||||||
_ => Err(Error::BadRPC(format!("Unexpected RPC message"))),
|
_ => Err(Error::BadRPC("Unexpected RPC message".to_string())),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
@ -358,13 +358,13 @@ impl System {
|
||||||
) {
|
) {
|
||||||
let self2 = self.clone();
|
let self2 = self.clone();
|
||||||
self.background
|
self.background
|
||||||
.spawn_worker(format!("discovery loop"), |stop_signal| {
|
.spawn_worker("discovery loop".to_string(), |stop_signal| {
|
||||||
self2.discovery_loop(peers, consul_host, consul_service_name, stop_signal)
|
self2.discovery_loop(peers, consul_host, consul_service_name, stop_signal)
|
||||||
});
|
});
|
||||||
|
|
||||||
let self2 = self.clone();
|
let self2 = self.clone();
|
||||||
self.background
|
self.background
|
||||||
.spawn_worker(format!("ping loop"), |stop_signal| {
|
.spawn_worker("ping loop".to_string(), |stop_signal| {
|
||||||
self2.ping_loop(stop_signal)
|
self2.ping_loop(stop_signal)
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
@ -424,7 +424,6 @@ impl System {
|
||||||
warn!("Node {:?} seems to be down.", id);
|
warn!("Node {:?} seems to be down.", id);
|
||||||
if !ring.config.members.contains_key(id) {
|
if !ring.config.members.contains_key(id) {
|
||||||
info!("Removing node {:?} from status (not in config and not responding to pings anymore)", id);
|
info!("Removing node {:?} from status (not in config and not responding to pings anymore)", id);
|
||||||
drop(st);
|
|
||||||
status.nodes.remove(&id);
|
status.nodes.remove(&id);
|
||||||
has_changes = true;
|
has_changes = true;
|
||||||
}
|
}
|
||||||
|
@ -438,7 +437,7 @@ impl System {
|
||||||
self.update_status(&update_locked, status).await;
|
self.update_status(&update_locked, status).await;
|
||||||
drop(update_locked);
|
drop(update_locked);
|
||||||
|
|
||||||
if to_advertise.len() > 0 {
|
if !to_advertise.is_empty() {
|
||||||
self.broadcast(Message::AdvertiseNodesUp(to_advertise), PING_TIMEOUT)
|
self.broadcast(Message::AdvertiseNodesUp(to_advertise), PING_TIMEOUT)
|
||||||
.await;
|
.await;
|
||||||
}
|
}
|
||||||
|
@ -474,15 +473,13 @@ impl System {
|
||||||
Ok(self.make_ping())
|
Ok(self.make_ping())
|
||||||
}
|
}
|
||||||
|
|
||||||
fn handle_pull_status(&self) -> Result<Message, Error> {
|
fn handle_pull_status(&self) -> Message {
|
||||||
Ok(Message::AdvertiseNodesUp(
|
Message::AdvertiseNodesUp(self.status.borrow().to_serializable_membership(self))
|
||||||
self.status.borrow().to_serializable_membership(self),
|
|
||||||
))
|
|
||||||
}
|
}
|
||||||
|
|
||||||
fn handle_pull_config(&self) -> Result<Message, Error> {
|
fn handle_pull_config(&self) -> Message {
|
||||||
let ring = self.ring.borrow().clone();
|
let ring = self.ring.borrow().clone();
|
||||||
Ok(Message::AdvertiseConfig(ring.config.clone()))
|
Message::AdvertiseConfig(ring.config.clone())
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn handle_advertise_nodes_up(
|
async fn handle_advertise_nodes_up(
|
||||||
|
@ -530,7 +527,7 @@ impl System {
|
||||||
self.update_status(&update_lock, status).await;
|
self.update_status(&update_lock, status).await;
|
||||||
drop(update_lock);
|
drop(update_lock);
|
||||||
|
|
||||||
if to_ping.len() > 0 {
|
if !to_ping.is_empty() {
|
||||||
self.background
|
self.background
|
||||||
.spawn_cancellable(self.clone().ping_nodes(to_ping).map(Ok));
|
.spawn_cancellable(self.clone().ping_nodes(to_ping).map(Ok));
|
||||||
}
|
}
|
||||||
|
@ -576,8 +573,8 @@ impl System {
|
||||||
self.clone().ping_nodes(ping_addrs).await;
|
self.clone().ping_nodes(ping_addrs).await;
|
||||||
|
|
||||||
select! {
|
select! {
|
||||||
_ = restart_at.fuse() => (),
|
_ = restart_at.fuse() => {},
|
||||||
_ = stop_signal.changed().fuse() => (),
|
_ = stop_signal.changed().fuse() => {},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -595,7 +592,7 @@ impl System {
|
||||||
};
|
};
|
||||||
|
|
||||||
while !*stop_signal.borrow() {
|
while !*stop_signal.borrow() {
|
||||||
let not_configured = self.ring.borrow().config.members.len() == 0;
|
let not_configured = self.ring.borrow().config.members.is_empty();
|
||||||
let no_peers = self.status.borrow().nodes.len() < 3;
|
let no_peers = self.status.borrow().nodes.len() < 3;
|
||||||
let bad_peers = self
|
let bad_peers = self
|
||||||
.status
|
.status
|
||||||
|
@ -613,11 +610,8 @@ impl System {
|
||||||
.map(|ip| (*ip, None))
|
.map(|ip| (*ip, None))
|
||||||
.collect::<Vec<_>>();
|
.collect::<Vec<_>>();
|
||||||
|
|
||||||
match self.persist_status.load_async().await {
|
if let Ok(peers) = self.persist_status.load_async().await {
|
||||||
Ok(peers) => {
|
ping_list.extend(peers.iter().map(|x| (x.addr, Some(x.id))));
|
||||||
ping_list.extend(peers.iter().map(|x| (x.addr, Some(x.id))));
|
|
||||||
}
|
|
||||||
_ => (),
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if let Some((consul_host, consul_service_name)) = &consul_config {
|
if let Some((consul_host, consul_service_name)) = &consul_config {
|
||||||
|
@ -636,12 +630,14 @@ impl System {
|
||||||
|
|
||||||
let restart_at = tokio::time::sleep(DISCOVERY_INTERVAL);
|
let restart_at = tokio::time::sleep(DISCOVERY_INTERVAL);
|
||||||
select! {
|
select! {
|
||||||
_ = restart_at.fuse() => (),
|
_ = restart_at.fuse() => {},
|
||||||
_ = stop_signal.changed().fuse() => (),
|
_ = stop_signal.changed().fuse() => {},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// for some reason fixing this is causing compilation error, see https://github.com/rust-lang/rust-clippy/issues/7052
|
||||||
|
#[allow(clippy::manual_async_fn)]
|
||||||
fn pull_status(
|
fn pull_status(
|
||||||
self: Arc<Self>,
|
self: Arc<Self>,
|
||||||
peer: UUID,
|
peer: UUID,
|
||||||
|
@ -672,18 +668,15 @@ impl System {
|
||||||
let mut list = status.to_serializable_membership(&self);
|
let mut list = status.to_serializable_membership(&self);
|
||||||
|
|
||||||
// Combine with old peer list to make sure no peer is lost
|
// Combine with old peer list to make sure no peer is lost
|
||||||
match self.persist_status.load_async().await {
|
if let Ok(old_list) = self.persist_status.load_async().await {
|
||||||
Ok(old_list) => {
|
for pp in old_list {
|
||||||
for pp in old_list {
|
if !list.iter().any(|np| pp.id == np.id) {
|
||||||
if !list.iter().any(|np| pp.id == np.id) {
|
list.push(pp);
|
||||||
list.push(pp);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
_ => (),
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if list.len() > 0 {
|
if !list.is_empty() {
|
||||||
info!("Persisting new peer list ({} peers)", list.len());
|
info!("Persisting new peer list ({} peers)", list.len());
|
||||||
self.persist_status
|
self.persist_status
|
||||||
.save_async(&list)
|
.save_async(&list)
|
||||||
|
|
|
@ -141,8 +141,7 @@ impl Ring {
|
||||||
if i_round >= node_info.capacity {
|
if i_round >= node_info.capacity {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
for pos2 in *pos..q.len() {
|
for (pos2, &qv) in q.iter().enumerate().skip(*pos) {
|
||||||
let qv = q[pos2];
|
|
||||||
if partitions[qv].len() != rep {
|
if partitions[qv].len() != rep {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
@ -205,7 +204,7 @@ impl Ring {
|
||||||
for (i, entry) in self.ring.iter().enumerate() {
|
for (i, entry) in self.ring.iter().enumerate() {
|
||||||
ret.push((i as u16, entry.location));
|
ret.push((i as u16, entry.location));
|
||||||
}
|
}
|
||||||
if ret.len() > 0 {
|
if !ret.is_empty() {
|
||||||
assert_eq!(ret[0].1, [0u8; 32].into());
|
assert_eq!(ret[0].1, [0u8; 32].into());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -234,6 +233,6 @@ impl Ring {
|
||||||
assert_eq!(partition_top & PARTITION_MASK_U16, top & PARTITION_MASK_U16);
|
assert_eq!(partition_top & PARTITION_MASK_U16, top & PARTITION_MASK_U16);
|
||||||
|
|
||||||
assert!(n <= partition.nodes.len());
|
assert!(n <= partition.nodes.len());
|
||||||
partition.nodes[..n].iter().cloned().collect::<Vec<_>>()
|
partition.nodes[..n].to_vec()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -240,7 +240,7 @@ impl<M: RpcMessage> RpcAddrClient<M> {
|
||||||
pub fn new(http_client: Arc<RpcHttpClient>, path: String) -> Self {
|
pub fn new(http_client: Arc<RpcHttpClient>, path: String) -> Self {
|
||||||
Self {
|
Self {
|
||||||
phantom: PhantomData::default(),
|
phantom: PhantomData::default(),
|
||||||
http_client: http_client,
|
http_client,
|
||||||
path,
|
path,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -57,7 +57,7 @@ where
|
||||||
trace!(
|
trace!(
|
||||||
"Request message: {}",
|
"Request message: {}",
|
||||||
serde_json::to_string(&msg)
|
serde_json::to_string(&msg)
|
||||||
.unwrap_or("<json error>".into())
|
.unwrap_or_else(|_| "<json error>".into())
|
||||||
.chars()
|
.chars()
|
||||||
.take(100)
|
.take(100)
|
||||||
.collect::<String>()
|
.collect::<String>()
|
||||||
|
@ -123,7 +123,7 @@ impl RpcServer {
|
||||||
req: Request<Body>,
|
req: Request<Body>,
|
||||||
addr: SocketAddr,
|
addr: SocketAddr,
|
||||||
) -> Result<Response<Body>, Error> {
|
) -> Result<Response<Body>, Error> {
|
||||||
if req.method() != &Method::POST {
|
if req.method() != Method::POST {
|
||||||
let mut bad_request = Response::default();
|
let mut bad_request = Response::default();
|
||||||
*bad_request.status_mut() = StatusCode::BAD_REQUEST;
|
*bad_request.status_mut() = StatusCode::BAD_REQUEST;
|
||||||
return Ok(bad_request);
|
return Ok(bad_request);
|
||||||
|
@ -201,7 +201,7 @@ impl RpcServer {
|
||||||
.get_ref()
|
.get_ref()
|
||||||
.0
|
.0
|
||||||
.peer_addr()
|
.peer_addr()
|
||||||
.unwrap_or(([0, 0, 0, 0], 0).into());
|
.unwrap_or_else(|_| ([0, 0, 0, 0], 0).into());
|
||||||
let self_arc = self_arc.clone();
|
let self_arc = self_arc.clone();
|
||||||
async move {
|
async move {
|
||||||
Ok::<_, Error>(service_fn(move |req: Request<Body>| {
|
Ok::<_, Error>(service_fn(move |req: Request<Body>| {
|
||||||
|
|
|
@ -72,7 +72,7 @@ impl FixedBytes32 {
|
||||||
&mut self.0[..]
|
&mut self.0[..]
|
||||||
}
|
}
|
||||||
/// Copy to a slice
|
/// Copy to a slice
|
||||||
pub fn to_vec(&self) -> Vec<u8> {
|
pub fn to_vec(self) -> Vec<u8> {
|
||||||
self.0.to_vec()
|
self.0.to_vec()
|
||||||
}
|
}
|
||||||
/// Try building a FixedBytes32 from a slice
|
/// Try building a FixedBytes32 from a slice
|
||||||
|
|
|
@ -93,12 +93,12 @@ impl From<sled::transaction::TransactionError<Error>> for Error {
|
||||||
|
|
||||||
impl<T> From<tokio::sync::watch::error::SendError<T>> for Error {
|
impl<T> From<tokio::sync::watch::error::SendError<T>> for Error {
|
||||||
fn from(_e: tokio::sync::watch::error::SendError<T>) -> Error {
|
fn from(_e: tokio::sync::watch::error::SendError<T>) -> Error {
|
||||||
Error::Message(format!("Watch send error"))
|
Error::Message("Watch send error".to_string())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<T> From<tokio::sync::mpsc::error::SendError<T>> for Error {
|
impl<T> From<tokio::sync::mpsc::error::SendError<T>> for Error {
|
||||||
fn from(_e: tokio::sync::mpsc::error::SendError<T>) -> Error {
|
fn from(_e: tokio::sync::mpsc::error::SendError<T>) -> Error {
|
||||||
Error::Message(format!("MPSC send error"))
|
Error::Message("MPSC send error".to_string())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,3 +1,4 @@
|
||||||
|
#![allow(clippy::upper_case_acronyms)]
|
||||||
//! Crate containing common functions and types used in Garage
|
//! Crate containing common functions and types used in Garage
|
||||||
|
|
||||||
#[macro_use]
|
#[macro_use]
|
||||||
|
|
|
@ -1,5 +1,5 @@
|
||||||
use std::io::{Read, Write};
|
use std::io::{Read, Write};
|
||||||
use std::path::PathBuf;
|
use std::path::{Path, PathBuf};
|
||||||
|
|
||||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||||
|
|
||||||
|
@ -18,8 +18,8 @@ impl<T> Persister<T>
|
||||||
where
|
where
|
||||||
T: Serialize + for<'de> Deserialize<'de>,
|
T: Serialize + for<'de> Deserialize<'de>,
|
||||||
{
|
{
|
||||||
pub fn new(base_dir: &PathBuf, file_name: &str) -> Self {
|
pub fn new(base_dir: &Path, file_name: &str) -> Self {
|
||||||
let mut path = base_dir.clone();
|
let mut path = base_dir.to_path_buf();
|
||||||
path.push(file_name);
|
path.push(file_name);
|
||||||
Self {
|
Self {
|
||||||
path,
|
path,
|
||||||
|
|
Loading…
Reference in a new issue