Merge pull request 'replaced mutex with rwlock' (#3) from feature-rwlock into develop

Reviewed-on: #3

Cargo.toml

@@ -21,5 +21,5 @@ regex = "1.7"
 reqwest = { version = "0.11", default-features = false, features = ["json","rustls-tls"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
-tokio = { version = "1.23", features = ["full"] }
+tokio = { version = "1.23", features = ["full", "sync"] }
 zmq = "0.10"

src/api.rs

@@ -5,11 +5,11 @@ use std::io;
 use std::sync::Arc;
 use tokio::io::AsyncWriteExt;
 use tokio::net::TcpSocket;
-use tokio::sync::Mutex;
+use tokio::sync::RwLock;
 
-pub async fn apiserver(ctxarc: &Arc<Mutex<Context>>) -> io::Result<()> {
+pub async fn apiserver(ctxarc: &Arc<RwLock<Context>>) -> io::Result<()> {
     let ctxclone = Arc::clone(ctxarc);
-    let ctx = ctxclone.lock().await;
+    let ctx = ctxclone.read().await;
     let addr = ctx.cfg.api.parse().unwrap();
     drop(ctx);
 
@@ -26,7 +26,7 @@ pub async fn apiserver(ctxarc: &Arc<Mutex<Context>>) -> io::Result<()> {
                     //let mut buf = [0; 1024];
                     let data;
                     {
-                        let ctx = ctxclone.lock().await;
+                        let ctx = ctxclone.read().await;
                         data = serde_json::to_string(&ctx.blocklist);
                     }
 

src/ipblc.rs

@@ -13,7 +13,7 @@ use nix::sys::inotify::InotifyEvent;
 use std::collections::HashMap;
 use std::sync::Arc;
 use tokio::sync::mpsc::{channel, Receiver, Sender};
-use tokio::sync::Mutex;
+use tokio::sync::RwLock;
 
 pub const PKG_NAME: &str = env!("CARGO_PKG_NAME");
 const BL_CHAN_SIZE: usize = 32;
@@ -21,7 +21,7 @@ const ZMQ_CHAN_SIZE: usize = 64;
 const API_CHAN_SIZE: usize = 64;
 
 pub async fn run() {
-    let ctxarc = Arc::new(Mutex::new(Context::new().await));
+    let ctxarc = Arc::new(RwLock::new(Context::new().await));
 
     let pkgversion = format!("{}@{}", env!("CARGO_PKG_VERSION"), GIT_VERSION);
 
@@ -72,7 +72,7 @@ pub async fn run() {
             val = ipdata_wait => {
                 let mut received_ip = val.unwrap();
 
-                let mut ctx = ctxclone.lock().await;
+                let mut ctx = ctxclone.write().await;
 
                 if received_ip.ip == "".to_string() && received_ip.mode == "init".to_string() {
                     for ip_to_send in &mut ctx.get_blocklist_toblock().await {
@@ -97,7 +97,7 @@ pub async fn run() {
 
         let toblock;
         {
-            let mut ctx = ctxarc.lock().await;
+            let mut ctx = ctxarc.write().await;
             ctx.gc_blocklist().await;
             toblock = ctx.get_blocklist_toblock().await;
         }
@@ -118,7 +118,7 @@ pub async fn run() {
             let now_cfg_reload = Local::now().trunc_subsecs(0);
             if (now_cfg_reload - last_cfg_reload) > Duration::seconds(5) {
                 // reload configuration from the server
-                let mut ctx = ctxclone.lock().await;
+                let mut ctx = ctxclone.write().await;
                 match ctx.load().await {
                     Ok(_) => {
                         last_cfg_reload = Local::now().trunc_subsecs(0);
@@ -133,7 +133,7 @@ pub async fn run() {
     }
 }
 
-async fn watchfiles(ctxarc: &Arc<Mutex<Context>>) -> Receiver<FileEvent> {
+async fn watchfiles(ctxarc: &Arc<RwLock<Context>>) -> Receiver<FileEvent> {
     let (bltx, blrx): (Sender<FileEvent>, Receiver<FileEvent>) = channel(BL_CHAN_SIZE);
     let ctxclone = Arc::clone(ctxarc);
     tokio::spawn(async move {
@@ -141,7 +141,7 @@ async fn watchfiles(ctxarc: &Arc<Mutex<Context>>) -> Receiver<FileEvent> {
             let events;
             let instance;
             {
-                let ctx = ctxclone.lock().await;
+                let ctx = ctxclone.read().await;
                 instance = ctx.instance.clone();
             }
 
@@ -169,7 +169,7 @@ async fn get_last_file_size(w: &mut HashMap<String, u64>, path: &str) -> (u64, b
 }
 
 async fn compare_files_changes(
-    ctxarc: &Arc<Mutex<Context>>,
+    ctxarc: &Arc<RwLock<Context>>,
     inrx: &mut Receiver<FileEvent>,
     ipdatatx: &Sender<IpData>,
 ) {
@@ -181,7 +181,7 @@ async fn compare_files_changes(
         let sask;
         let sas;
         {
-            let ctx = ctxarc.lock().await;
+            let ctx = ctxarc.read().await;
             sas = ctx.clone().sas;
             sask = sas.keys();
             tnets = ctx.cfg.build_trustnets();
@@ -204,7 +204,7 @@ async fn compare_files_changes(
 
                         let (filesize, sizechanged);
                         {
-                            let mut ctx = ctxarc.lock().await;
+                            let mut ctx = ctxarc.write().await;
                             let sa = ctx.sas.get_mut(sak).unwrap();
                             (filesize, sizechanged) =
                                 get_last_file_size(&mut sa.watchedfiles, &handle).await;

src/zmqcom.rs

@@ -4,7 +4,7 @@ use crate::utils::gethostname;
 
 use std::sync::Arc;
 use tokio::sync::mpsc::Sender;
-use tokio::sync::Mutex;
+use tokio::sync::RwLock;
 
 const ZMQPROTO: &str = "tcp";
 
@@ -21,13 +21,13 @@ pub async fn zconnect<'a>(
     Ok(socket)
 }
 
-pub async fn zmqinit(ctx: &Arc<Mutex<Context>>, ipdatatx: &Sender<IpData>) -> zmq::Socket {
+pub async fn zmqinit(ctx: &Arc<RwLock<Context>>, ipdatatx: &Sender<IpData>) -> zmq::Socket {
     let ctxarc = Arc::clone(&ctx);
 
     let zmqreqsocket;
     let zmqsubsocket;
     {
-        let zmqctx = ctxarc.lock().await;
+        let zmqctx = ctxarc.read().await;
         zmqreqsocket = zconnect(&zmqctx.cfg.zmq.get("reqrep").unwrap(), zmq::REQ)
             .await
             .unwrap();
@@ -40,10 +40,10 @@ pub async fn zmqinit(ctx: &Arc<Mutex<Context>>, ipdatatx: &Sender<IpData>) -> zm
     return zmqreqsocket;
 }
 
-async fn listenpubsub(ctx: &Arc<Mutex<Context>>, txpubsub: Sender<IpData>, socket: zmq::Socket) {
+async fn listenpubsub(ctx: &Arc<RwLock<Context>>, txpubsub: Sender<IpData>, socket: zmq::Socket) {
     let prefix;
     {
-        let ctx = ctx.lock().await;
+        let ctx = ctx.read().await;
         prefix = format!(
             "{sub} ",
             sub = ctx.cfg.zmq.get("pubsub").unwrap().subscription