Use a RwLock instead of a Mutex for the peers

This commit is contained in:
Nils Andreas Svee committed 2022-09-28 00:43:48 +02:00
1 parent e3b7b7e9ba
commit 9219ac690a
4 files changed
+16 -16

No files matched your search

+4 -4
View File
@@ -21,7 +21,7 @@ use std::net::IpAddr;
use std::process::ExitCode; use std::process::ExitCode;
use std::str::FromStr; use std::str::FromStr;
use std::sync::mpsc::channel; use std::sync::mpsc::channel;
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex, RwLock};
use std::thread::sleep; use std::thread::sleep;
use std::time::Duration; use std::time::Duration;
use std::{panic, process, thread}; use std::{panic, process, thread};
@@ -68,7 +68,7 @@ fn main() -> ExitCode {
// Create data structures shared by different threads // Create data structures shared by different threads
let owd_baseline = Arc::new(Mutex::new(HashMap::<IpAddr, ReflectorStats>::new())); let owd_baseline = Arc::new(Mutex::new(HashMap::<IpAddr, ReflectorStats>::new()));
let owd_recent = Arc::new(Mutex::new(HashMap::<IpAddr, ReflectorStats>::new())); let owd_recent = Arc::new(Mutex::new(HashMap::<IpAddr, ReflectorStats>::new()));
let reflector_peers_lock = Arc::new(Mutex::new(Vec::<IpAddr>::new())); let reflector_peers_lock = Arc::new(RwLock::new(Vec::<IpAddr>::new()));
let mut reflector_pool = Vec::<IpAddr>::new(); let mut reflector_pool = Vec::<IpAddr>::new();
let reflector_pool_size = reflectors.len(); let reflector_pool_size = reflectors.len();
@@ -83,12 +83,12 @@ fn main() -> ExitCode {
match reflector_pool_size > 5 { match reflector_pool_size > 5 {
true => { true => {
let mut peers = reflector_peers_lock.lock().unwrap(); let mut peers = reflector_peers_lock.write().unwrap();
peers.append(default_reflectors.to_vec().as_mut()); peers.append(default_reflectors.to_vec().as_mut());
reflector_pool.append(reflectors.as_mut()); reflector_pool.append(reflectors.as_mut());
} }
false => { false => {
let mut peers = reflector_peers_lock.lock().unwrap(); let mut peers = reflector_peers_lock.write().unwrap();
peers.append(default_reflectors.to_vec().as_mut()); peers.append(default_reflectors.to_vec().as_mut());
} }
} }
+5 -5
View File
@@ -8,7 +8,7 @@ use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddrV4, SocketAddrV6};
use std::os::unix::io::RawFd; use std::os::unix::io::RawFd;
use std::str::FromStr; use std::str::FromStr;
use std::sync::mpsc::Sender; use std::sync::mpsc::Sender;
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex, RwLock};
use std::thread; use std::thread;
use std::time::Duration; use std::time::Duration;
@@ -59,7 +59,7 @@ pub trait PingListener {
fn listen( fn listen(
&mut self, &mut self,
type_: SocketType, type_: SocketType,
reflectors_lock: Arc<Mutex<Vec<IpAddr>>>, reflectors_lock: Arc<RwLock<Vec<IpAddr>>>,
stats_sender: Sender<PingReply>, stats_sender: Sender<PingReply>,
) -> Result<(), Box<dyn Error>> { ) -> Result<(), Box<dyn Error>> {
let socket = open_socket(type_)?; let socket = open_socket(type_)?;
@@ -80,7 +80,7 @@ pub trait PingListener {
let addr_bytes = sender.expect("Should be an address here").ip(); let addr_bytes = sender.expect("Should be an address here").ip();
let addr = IpAddr::from(addr_bytes.to_be_bytes()); let addr = IpAddr::from(addr_bytes.to_be_bytes());
let reflectors = reflectors_lock.lock().unwrap(); let reflectors = reflectors_lock.read().unwrap();
if !reflectors.contains(&addr) { if !reflectors.contains(&addr) {
continue; continue;
} }
@@ -112,7 +112,7 @@ pub trait PingSender {
fn send( fn send(
&mut self, &mut self,
type_: SocketType, type_: SocketType,
reflectors_lock: Arc<Mutex<Vec<IpAddr>>>, reflectors_lock: Arc<RwLock<Vec<IpAddr>>>,
) -> Result<(), Box<dyn Error>> { ) -> Result<(), Box<dyn Error>> {
let socket = open_socket(type_)?; let socket = open_socket(type_)?;
@@ -121,7 +121,7 @@ pub trait PingSender {
let sleep_duration = Duration::from_millis(tick_duration_ms as u64); let sleep_duration = Duration::from_millis(tick_duration_ms as u64);
loop { loop {
let reflectors_unlocked = reflectors_lock.lock().unwrap(); let reflectors_unlocked = reflectors_lock.read().unwrap();
let reflectors = reflectors_unlocked.clone(); let reflectors = reflectors_unlocked.clone();
drop(reflectors_unlocked); drop(reflectors_unlocked);
+3 -3
View File
@@ -9,7 +9,7 @@ use std::fs::File;
use std::io::Write; use std::io::Write;
use std::net::IpAddr; use std::net::IpAddr;
use std::sync::mpsc::Sender; use std::sync::mpsc::Sender;
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex, RwLock};
use std::thread::sleep; use std::thread::sleep;
use std::time::Duration; use std::time::Duration;
@@ -65,7 +65,7 @@ pub struct Ratecontroller {
pub(crate) config: Config, pub(crate) config: Config,
pub(crate) owd_baseline: Arc<Mutex<HashMap<IpAddr, ReflectorStats>>>, pub(crate) owd_baseline: Arc<Mutex<HashMap<IpAddr, ReflectorStats>>>,
pub(crate) owd_recent: Arc<Mutex<HashMap<IpAddr, ReflectorStats>>>, pub(crate) owd_recent: Arc<Mutex<HashMap<IpAddr, ReflectorStats>>>,
pub(crate) reflectors_lock: Arc<Mutex<Vec<IpAddr>>>, pub(crate) reflectors_lock: Arc<RwLock<Vec<IpAddr>>>,
pub(crate) reselect_trigger: Sender<bool>, pub(crate) reselect_trigger: Sender<bool>,
} }
@@ -164,7 +164,7 @@ impl Ratecontroller {
// change speeds here // change speeds here
let owd_baseline = self.owd_baseline.lock().unwrap(); let owd_baseline = self.owd_baseline.lock().unwrap();
let owd_recent = self.owd_recent.lock().unwrap(); let owd_recent = self.owd_recent.lock().unwrap();
let reflectors = self.reflectors_lock.lock().unwrap(); let reflectors = self.reflectors_lock.read().unwrap();
let mut down_delta_stat: f64; let mut down_delta_stat: f64;
let mut up_delta_stat: f64; let mut up_delta_stat: f64;
+4 -4
View File
@@ -5,14 +5,14 @@ use rand::{thread_rng, Rng};
use std::collections::HashMap; use std::collections::HashMap;
use std::net::IpAddr; use std::net::IpAddr;
use std::sync::mpsc::Receiver; use std::sync::mpsc::Receiver;
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex, RwLock};
use std::thread::sleep; use std::thread::sleep;
use std::time::Duration; use std::time::Duration;
pub struct ReflectorSelector { pub struct ReflectorSelector {
pub(crate) config: Config, pub(crate) config: Config,
pub(crate) owd_recent: Arc<Mutex<HashMap<IpAddr, ReflectorStats>>>, pub(crate) owd_recent: Arc<Mutex<HashMap<IpAddr, ReflectorStats>>>,
pub(crate) reflector_peers_lock: Arc<Mutex<Vec<IpAddr>>>, pub(crate) reflector_peers_lock: Arc<RwLock<Vec<IpAddr>>>,
pub(crate) reflector_pool: Vec<IpAddr>, pub(crate) reflector_pool: Vec<IpAddr>,
pub(crate) trigger_channel: Receiver<bool>, pub(crate) trigger_channel: Receiver<bool>,
} }
@@ -50,7 +50,7 @@ impl ReflectorSelector {
} }
let mut next_peers: Vec<IpAddr> = Vec::new(); let mut next_peers: Vec<IpAddr> = Vec::new();
let mut reflectors_peers = self.reflector_peers_lock.lock().unwrap(); let mut reflectors_peers = self.reflector_peers_lock.write().unwrap();
// Include all current peers // Include all current peers
for reflector in reflectors_peers.iter() { for reflector in reflectors_peers.iter() {
@@ -76,7 +76,7 @@ impl ReflectorSelector {
sleep(baseline_sleep_time); sleep(baseline_sleep_time);
// Re-acquire the lock when we wake up again // Re-acquire the lock when we wake up again
reflectors_peers = self.reflector_peers_lock.lock().unwrap(); reflectors_peers = self.reflector_peers_lock.write().unwrap();
reflectors_peers.len(); reflectors_peers.len();
let mut candidates = Vec::new(); let mut candidates = Vec::new();