bin/ircd: more optimization for the buffer code and clean up

bin/ircd/src/buffers.rs

@@ -7,68 +7,23 @@ use std::{
 use chrono::Utc;
 use ripemd::{Digest, Ripemd160};
 
-use crate::Privmsg;
+use crate::{settings, Privmsg};
 
-pub const SIZE_OF_MSGS_BUFFER: usize = 4095;
-pub const SIZE_OF_MSG_IDSS_BUFFER: usize = 65536;
-pub const LIFETIME_FOR_ORPHAN: i64 = 600;
-pub const TERM_MAX_TIME_DIFFERENCE: i64 = 180;
-
-pub type InvSeenIds = Arc<Mutex<RingBuffer<u64>>>;
-pub type SeenIds = Mutex<RingBuffer<u64>>;
-pub type MutexPrivmsgsBuffer = Mutex<PrivmsgsBuffer>;
-pub type UnreadMsgs = Mutex<UMsgs>;
 pub type Buffers = Arc<Msgs>;
 
 pub struct Msgs {
-    pub privmsgs: MutexPrivmsgsBuffer,
-    pub unread_msgs: UnreadMsgs,
+    pub privmsgs: PrivmsgsBuffer,
+    pub unread_msgs: UMsgs,
     pub seen_ids: SeenIds,
 }
 
 pub fn create_buffers() -> Buffers {
-    let seen_ids = Mutex::new(RingBuffer::new(SIZE_OF_MSG_IDSS_BUFFER));
+    let seen_ids = SeenIds::new();
     let privmsgs = PrivmsgsBuffer::new();
-    let unread_msgs = Mutex::new(UMsgs::new());
+    let unread_msgs = UMsgs::new();
     Arc::new(Msgs { privmsgs, unread_msgs, seen_ids })
 }
 
-#[derive(Clone)]
-pub struct UMsgs {
-    pub msgs: BTreeMap<String, Privmsg>,
-    capacity: usize,
-}
-
-impl UMsgs {
-    pub fn new() -> Self {
-        Self { msgs: BTreeMap::new(), capacity: SIZE_OF_MSGS_BUFFER }
-    }
-
-    pub fn insert(&mut self, msg: &Privmsg) -> String {
-        let mut hasher = Ripemd160::new();
-        hasher.update(msg.to_string());
-        let key = hex::encode(hasher.finalize());
-
-        if self.msgs.len() == self.capacity {
-            self.pop_front();
-        }
-
-        self.msgs.insert(key.clone(), msg.clone());
-        key
-    }
-
-    fn pop_front(&mut self) {
-        let first_key = self.msgs.iter().next_back().unwrap().0.clone();
-        self.msgs.remove(&first_key);
-    }
-}
-
-impl Default for UMsgs {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
 #[derive(Clone)]
 pub struct RingBuffer<T> {
     pub items: VecDeque<T>,
@@ -116,18 +71,61 @@ impl<T: Eq + PartialEq + Clone> RingBuffer<T> {
     }
 }
 
-#[derive(Clone)]
 pub struct PrivmsgsBuffer {
+    msgs: Mutex<OrderingAlgo>,
+}
+
+impl PrivmsgsBuffer {
+    pub fn new() -> Self {
+        Self { msgs: Mutex::new(OrderingAlgo::new()) }
+    }
+
+    pub async fn push(&self, privmsg: &Privmsg) {
+        self.msgs.lock().await.push(privmsg);
+    }
+
+    pub async fn load(&self) -> Vec<Privmsg> {
+        self.msgs.lock().await.load()
+    }
+
+    pub async fn get_msg_by_term(&self, term: u64) -> Option<Privmsg> {
+        self.msgs.lock().await.get_msg_by_term(term)
+    }
+
+    pub async fn len(&self) -> usize {
+        self.msgs.lock().await.len()
+    }
+
+    pub async fn is_empty(&self) -> bool {
+        self.msgs.lock().await.is_empty()
+    }
+
+    pub async fn last_term(&self) -> u64 {
+        self.msgs.lock().await.last_term()
+    }
+
+    pub async fn fetch_msgs(&self, term: u64) -> Vec<Privmsg> {
+        self.msgs.lock().await.fetch_msgs(term)
+    }
+}
+
+pub struct OrderingAlgo {
     buffer: RingBuffer<Privmsg>,
     orphans: RingBuffer<Orphan>,
 }
 
-impl PrivmsgsBuffer {
-    pub fn new() -> MutexPrivmsgsBuffer {
-        Mutex::new(Self {
-            buffer: RingBuffer::new(SIZE_OF_MSGS_BUFFER),
-            orphans: RingBuffer::new(SIZE_OF_MSGS_BUFFER),
-        })
+impl Default for OrderingAlgo {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl OrderingAlgo {
+    pub fn new() -> Self {
+        Self {
+            buffer: RingBuffer::new(settings::SIZE_OF_MSGS_BUFFER),
+            orphans: RingBuffer::new(settings::SIZE_OF_MSGS_BUFFER),
+        }
     }
 
     pub fn push(&mut self, privmsg: &Privmsg) {
@@ -135,7 +133,7 @@ impl PrivmsgsBuffer {
             Ordering::Equal => self.buffer.push(privmsg.clone()),
             Ordering::Less => {
                 if let Some(msg) = self.get_msg_by_term(privmsg.term) {
-                    if (msg.timestamp - privmsg.timestamp) <= TERM_MAX_TIME_DIFFERENCE {
+                    if (msg.timestamp - privmsg.timestamp) <= settings::TERM_MAX_TIME_DIFFERENCE {
                         self.buffer.push(privmsg.clone());
                     }
                 } else {
@@ -147,8 +145,8 @@ impl PrivmsgsBuffer {
         self.update();
     }
 
-    pub fn iter(&self) -> impl Iterator<Item = &Privmsg> + DoubleEndedIterator {
-        self.buffer.iter()
+    pub fn load(&self) -> Vec<Privmsg> {
+        self.buffer.iter().cloned().collect::<Vec<Privmsg>>()
     }
 
     pub fn get_msg_by_term(&self, term: u64) -> Option<Privmsg> {
@@ -160,7 +158,7 @@ impl PrivmsgsBuffer {
     }
 
     pub fn is_empty(&self) -> bool {
-        self.len() == 0
+        self.buffer.is_empty()
     }
 
     pub fn last_term(&self) -> u64 {
@@ -194,15 +192,15 @@ impl PrivmsgsBuffer {
         });
     }
 
-    fn oprhan_is_valid(&mut self, orphan: &Orphan) -> bool {
-        (orphan.timestamp + LIFETIME_FOR_ORPHAN) > Utc::now().timestamp()
+    fn oprhan_is_valid(orphan: &Orphan) -> bool {
+        (orphan.timestamp + settings::LIFETIME_FOR_ORPHAN) > Utc::now().timestamp()
     }
 
     fn update_orphans(&mut self) {
         for orphan in self.orphans.clone().iter() {
             let privmsg = orphan.msg.clone();
 
-            if !self.oprhan_is_valid(orphan) {
+            if !Self::oprhan_is_valid(orphan) {
                 self.orphans.remove(orphan);
                 continue
             }
@@ -214,7 +212,8 @@ impl PrivmsgsBuffer {
                 }
                 Ordering::Less => {
                     if let Some(msg) = self.get_msg_by_term(privmsg.term) {
-                        if (msg.timestamp - privmsg.timestamp) <= TERM_MAX_TIME_DIFFERENCE {
+                        if (msg.timestamp - privmsg.timestamp) <= settings::TERM_MAX_TIME_DIFFERENCE
+                        {
                             self.buffer.push(privmsg.clone());
                         }
                     } else {
@@ -240,6 +239,90 @@ impl Orphan {
     }
 }
 
+pub struct SeenIds {
+    ids: Mutex<RingBuffer<u64>>,
+}
+
+impl Default for SeenIds {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl SeenIds {
+    pub fn new() -> Self {
+        Self { ids: Mutex::new(RingBuffer::new(settings::SIZE_OF_IDSS_BUFFER)) }
+    }
+
+    pub async fn push(&self, id: u64) -> bool {
+        let ids = &mut self.ids.lock().await;
+        if !ids.contains(&id) {
+            ids.push(id);
+            return true
+        }
+        false
+    }
+}
+
+pub struct UMsgs {
+    msgs: Mutex<BTreeMap<String, Privmsg>>,
+}
+
+impl Default for UMsgs {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl UMsgs {
+    pub fn new() -> Self {
+        Self { msgs: Mutex::new(BTreeMap::new()) }
+    }
+
+    pub async fn len(&self) -> usize {
+        self.msgs.lock().await.len()
+    }
+
+    pub async fn contains(&self, key: &str) -> bool {
+        self.msgs.lock().await.contains_key(key)
+    }
+
+    pub async fn remove(&self, key: &str) -> Option<Privmsg> {
+        self.msgs.lock().await.remove(key)
+    }
+
+    pub async fn get(&self, key: &str) -> Option<Privmsg> {
+        self.msgs.lock().await.get(key).cloned()
+    }
+
+    pub async fn load(&self) -> BTreeMap<String, Privmsg> {
+        self.msgs.lock().await.clone()
+    }
+
+    pub async fn inc_read_confirms(&self, key: &str) -> bool {
+        if let Some(msg) = self.msgs.lock().await.get_mut(key) {
+            msg.read_confirms += 1;
+            return true
+        }
+        false
+    }
+
+    pub async fn insert(&self, msg: &Privmsg) -> String {
+        let mut hasher = Ripemd160::new();
+        hasher.update(msg.to_string() + &msg.term.to_string());
+        let key = hex::encode(hasher.finalize());
+
+        let msgs = &mut self.msgs.lock().await;
+        if msgs.len() == settings::SIZE_OF_MSGS_BUFFER {
+            let first_key = msgs.iter().next_back().unwrap().0.clone();
+            msgs.remove(&first_key);
+        }
+
+        msgs.insert(key.clone(), msg.clone());
+        key
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -266,12 +349,9 @@ mod tests {
         assert_eq!(b.iter().last().unwrap(), &"h9");
     }
 
-    #[test]
-    fn test_privmsgs_buffer() {
-        let mut pms = PrivmsgsBuffer {
-            buffer: RingBuffer::new(SIZE_OF_MSGS_BUFFER),
-            orphans: RingBuffer::new(SIZE_OF_MSGS_BUFFER),
-        };
+    #[async_std::test]
+    async fn test_privmsgs_buffer() {
+        let pms = PrivmsgsBuffer::new();
 
         //
         // Fill the buffer with random generated terms in range 0..3001
@@ -281,12 +361,11 @@ mod tests {
 
         for term in terms {
             let privmsg = Privmsg::new("nick", "#dev", &format!("message_{}", term), term);
-            pms.push(&privmsg);
+            pms.push(&privmsg).await;
         }
 
-        assert_eq!(pms.buffer.len(), 3000);
-        assert_eq!(pms.last_term(), 3000);
-        assert_eq!(pms.orphans.len(), 0);
+        assert_eq!(pms.len().await, 3000);
+        assert_eq!(pms.last_term().await, 3000);
 
         //
         // Fill the buffer with random generated terms in range 2000..4001
@@ -298,12 +377,11 @@ mod tests {
 
         for term in terms {
             let privmsg = Privmsg::new("nick", "#dev", &format!("message_{}", term), term);
-            pms.push(&privmsg);
+            pms.push(&privmsg).await;
         }
 
-        assert_eq!(pms.buffer.len(), SIZE_OF_MSGS_BUFFER);
-        assert_eq!(pms.last_term(), 4000);
-        assert_eq!(pms.orphans.len(), 0);
+        assert_eq!(pms.len().await, settings::SIZE_OF_MSGS_BUFFER);
+        assert_eq!(pms.last_term().await, 4000);
 
         //
         // Fill the buffer with random generated terms in range 4000..7001
@@ -314,11 +392,44 @@ mod tests {
 
         for term in terms {
             let privmsg = Privmsg::new("nick", "#dev", &format!("message_{}", term), term);
-            pms.push(&privmsg);
+            pms.push(&privmsg).await;
         }
 
-        assert_eq!(pms.buffer.len(), SIZE_OF_MSGS_BUFFER);
-        assert_eq!(pms.last_term(), 7000);
-        assert_eq!(pms.orphans.len(), 0);
+        assert_eq!(pms.len().await, settings::SIZE_OF_MSGS_BUFFER);
+        assert_eq!(pms.last_term().await, 7000);
+    }
+
+    #[async_std::test]
+    async fn test_seen_ids() {
+        let seen_ids = SeenIds::default();
+        assert!(seen_ids.push(3000).await);
+        assert!(seen_ids.push(3001).await);
+        assert!(!seen_ids.push(3000).await);
+    }
+
+    #[async_std::test]
+    async fn test_unread_msgs() {
+        let unread_msgs = UMsgs::default();
+
+        let p = Privmsg::new("nick", "#dev", &format!("message_{}", 0), 0);
+        let p_k = unread_msgs.insert(&p).await;
+
+        let p2 = Privmsg::new("nick", "#dev", &format!("message_{}", 0), 1);
+        let p2_k = unread_msgs.insert(&p2).await;
+
+        let p3 = Privmsg::new("nick", "#dev", &format!("message_{}", 0), 2);
+        let p3_k = unread_msgs.insert(&p3).await;
+
+        assert_eq!(unread_msgs.len().await, 3);
+
+        assert_eq!(unread_msgs.get(&p_k).await, Some(p.clone()));
+        assert_eq!(unread_msgs.get(&p2_k).await, Some(p2));
+        assert_eq!(unread_msgs.get(&p3_k).await, Some(p3));
+
+        assert!(unread_msgs.inc_read_confirms(&p_k).await);
+        assert!(!unread_msgs.inc_read_confirms("NONE KEY").await);
+
+        assert_ne!(unread_msgs.get(&p_k).await, Some(p));
+        assert_eq!(unread_msgs.get(&p_k).await.unwrap().read_confirms, 1);
     }
 }

bin/ircd/src/irc/client.rs

@@ -16,17 +16,13 @@ use darkfi::{
 use crate::{
     buffers::Buffers,
     crypto::{decrypt_privmsg, decrypt_target, encrypt_privmsg},
-    privmsg::{MAXIMUM_LENGTH_OF_MESSAGE, MAXIMUM_LENGTH_OF_NICKNAME},
+    settings,
+    settings::RPL,
     ChannelInfo, Privmsg,
 };
 
 use super::IrcConfig;
 
-const RPL_NOTOPIC: u32 = 331;
-const RPL_TOPIC: u32 = 332;
-const RPL_NAMEREPLY: u32 = 353;
-const RPL_ENDOFNAMES: u32 = 366;
-
 pub struct IrcClient<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> {
     // network stream
     write_stream: WriteHalf<C>,
@@ -159,7 +155,7 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcClient<C> {
     }
 
     async fn update(&mut self, line: String) -> Result<()> {
-        if line.len() > MAXIMUM_LENGTH_OF_MESSAGE {
+        if line.len() > settings::MAXIMUM_LENGTH_OF_MESSAGE {
             return Err(Error::MalformedPacket)
         }
 
@@ -211,9 +207,8 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcClient<C> {
             }
 
             // Send dm messages in buffer
-            let privmsgs = self.buffers.privmsgs.lock().await.clone();
-            for msg in privmsgs.iter() {
-                self.process_msg(msg).await?;
+            for msg in self.buffers.privmsgs.load().await {
+                self.process_msg(&msg).await?;
             }
         }
         Ok(())
@@ -255,7 +250,7 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcClient<C> {
     }
 
     async fn on_receive_nick(&mut self, nickname: &str) -> Result<()> {
-        if nickname.len() > MAXIMUM_LENGTH_OF_NICKNAME {
+        if nickname.len() > settings::MAXIMUM_LENGTH_OF_NICKNAME {
             return Ok(())
         }
 
@@ -300,10 +295,22 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcClient<C> {
             // Client is asking or the topic
             let chan_info = self.irc_config.configured_chans.get(channel).unwrap();
             let topic_reply = if let Some(topic) = &chan_info.topic {
-                format!("{} {} {} :{}\r\n", RPL_TOPIC, self.irc_config.nickname, channel, topic)
+                format!(
+                    "{} {} {} :{}\r\n",
+                    RPL::Topic as u32,
+                    self.irc_config.nickname,
+                    channel,
+                    topic
+                )
             } else {
                 const TOPIC: &str = "No topic is set";
-                format!("{} {} {} :{}\r\n", RPL_NOTOPIC, self.irc_config.nickname, channel, TOPIC)
+                format!(
+                    "{} {} {} :{}\r\n",
+                    RPL::NoTopic as u32,
+                    self.irc_config.nickname,
+                    channel,
+                    TOPIC
+                )
             };
             self.reply(&topic_reply).await?;
         }
@@ -408,7 +415,7 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcClient<C> {
                 let names_reply = format!(
                     ":{}!anon@dark.fi {} = {} : {}\r\n",
                     self.irc_config.nickname,
-                    RPL_NAMEREPLY,
+                    RPL::NameReply as u32,
                     chan,
                     chan_info.names.join(" ")
                 );
@@ -417,7 +424,9 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcClient<C> {
 
                 let end_of_names = format!(
                     ":DarkFi {:03} {} {} :End of NAMES list\r\n",
-                    RPL_ENDOFNAMES, self.irc_config.nickname, chan
+                    RPL::EndOfNames as u32,
+                    self.irc_config.nickname,
+                    chan
                 );
 
                 self.reply(&end_of_names).await?;
@@ -437,9 +446,7 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcClient<C> {
 
         info!("[CLIENT {}] (Plain) PRIVMSG {} :{}", self.address, target, message,);
 
-        let privmsgs_buffer = self.buffers.privmsgs.lock().await;
-        let last_term = privmsgs_buffer.last_term() + 1;
-        drop(privmsgs_buffer);
+        let last_term = self.buffers.privmsgs.last_term().await + 1;
 
         let mut privmsg = Privmsg::new(&self.irc_config.nickname, target, &message, last_term);
 
@@ -470,10 +477,8 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcClient<C> {
             }
         }
 
-        {
-            (*self.buffers.seen_ids.lock().await).push(privmsg.id);
-            (*self.buffers.privmsgs.lock().await).push(&privmsg);
-        }
+        self.buffers.seen_ids.push(privmsg.id).await;
+        self.buffers.privmsgs.push(&privmsg).await;
 
         self.notify_clients
             .notify_with_exclude(privmsg.clone(), &[self.subscription.get_id()])

bin/ircd/src/main.rs

@@ -1,4 +1,4 @@
-use async_std::sync::{Arc, Mutex};
+use async_std::sync::Arc;
 use std::fmt;
 
 use async_channel::Receiver;
@@ -33,7 +33,7 @@ pub mod rpc;
 pub mod settings;
 
 use crate::{
-    buffers::{create_buffers, Buffers, RingBuffer, SIZE_OF_MSG_IDSS_BUFFER},
+    buffers::{create_buffers, Buffers},
     irc::IrcServer,
     privmsg::Privmsg,
     protocol_privmsg::{LastTerm, ProtocolPrivmsg},
@@ -41,9 +41,6 @@ use crate::{
     settings::{Args, ChannelInfo, CONFIG_FILE, CONFIG_FILE_CONTENTS},
 };
 
-const TIMEOUT_FOR_RESEND: u64 = 240;
-const SEND_LAST_TERM_MSG: u64 = 4;
-
 #[derive(serde::Serialize)]
 struct KeyPair {
     private_key: String,
@@ -58,9 +55,9 @@ impl fmt::Display for KeyPair {
 
 async fn resend_unread_msgs(p2p: P2pPtr, buffers: Buffers) -> Result<()> {
     loop {
-        sleep(TIMEOUT_FOR_RESEND).await;
+        sleep(settings::TIMEOUT_FOR_RESEND_UNREAD_MSGS).await;
 
-        for msg in buffers.unread_msgs.lock().await.msgs.values() {
+        for msg in buffers.unread_msgs.load().await.values() {
             p2p.broadcast(msg.clone()).await?;
         }
     }
@@ -68,9 +65,9 @@ async fn resend_unread_msgs(p2p: P2pPtr, buffers: Buffers) -> Result<()> {
 
 async fn send_last_term(p2p: P2pPtr, buffers: Buffers) -> Result<()> {
     loop {
-        sleep(SEND_LAST_TERM_MSG).await;
+        sleep(settings::BROADCAST_LAST_TERM_MSG).await;
 
-        let term = buffers.privmsgs.lock().await.last_term();
+        let term = buffers.privmsgs.last_term().await;
         p2p.broadcast(LastTerm { term }).await?;
     }
 }
@@ -122,7 +119,6 @@ impl Ircd {
 
 async_daemonize!(realmain);
 async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
-    let seen_inv_ids = Arc::new(Mutex::new(RingBuffer::new(SIZE_OF_MSG_IDSS_BUFFER)));
     let buffers = create_buffers();
 
     if settings.gen_secret {
@@ -163,16 +159,11 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
     let registry = p2p.protocol_registry();
 
     let buffers_cloned = buffers.clone();
-    let seen_inv_ids_cloned = seen_inv_ids.clone();
     registry
         .register(net::SESSION_ALL, move |channel, p2p| {
             let sender = p2p_send_channel.clone();
-            let seen_inv_ids_cloned = seen_inv_ids_cloned.clone();
             let buffers_cloned = buffers_cloned.clone();
-            async move {
-                ProtocolPrivmsg::init(channel, sender, p2p, seen_inv_ids_cloned, buffers_cloned)
-                    .await
-            }
+            async move { ProtocolPrivmsg::init(channel, sender, p2p, buffers_cloned).await }
         })
         .await;
 

bin/ircd/src/privmsg.rs

@@ -5,9 +5,6 @@ use darkfi::util::serial::{SerialDecodable, SerialEncodable};
 
 pub type PrivmsgId = u64;
 
-pub const MAXIMUM_LENGTH_OF_MESSAGE: usize = 1024;
-pub const MAXIMUM_LENGTH_OF_NICKNAME: usize = 32;
-
 #[derive(Debug, Clone, SerialEncodable, SerialDecodable, Eq, PartialEq)]
 pub struct Privmsg {
     pub id: PrivmsgId,

bin/ircd/src/protocol_privmsg.rs

@@ -13,13 +13,7 @@ use darkfi::{
     Result,
 };
 
-use crate::{
-    buffers::{Buffers, InvSeenIds},
-    Privmsg,
-};
-
-const MAX_CONFIRM: u8 = 4;
-const UNREAD_MSG_EXPIRE_TIME: i64 = 18000;
+use crate::{buffers::Buffers, settings, Privmsg};
 
 #[derive(SerialDecodable, SerialEncodable, Clone, Debug)]
 struct Inv {
@@ -62,7 +56,6 @@ pub struct ProtocolPrivmsg {
     last_term_sub: net::MessageSubscription<LastTerm>,
     p2p: net::P2pPtr,
     channel: net::ChannelPtr,
-    inv_ids: InvSeenIds,
     buffers: Buffers,
 }
 
@@ -71,7 +64,6 @@ impl ProtocolPrivmsg {
         channel: net::ChannelPtr,
         notify: async_channel::Sender<Privmsg>,
         p2p: net::P2pPtr,
-        inv_ids: InvSeenIds,
         buffers: Buffers,
     ) -> net::ProtocolBasePtr {
         let message_subsytem = channel.get_message_subsystem();
@@ -103,7 +95,6 @@ impl ProtocolPrivmsg {
             jobsman: net::ProtocolJobsManager::new("ProtocolPrivmsg", channel.clone()),
             p2p,
             channel,
-            inv_ids,
             buffers,
         })
     }
@@ -115,19 +106,13 @@ impl ProtocolPrivmsg {
             let inv = self.inv_sub.receive().await?;
             let inv = (*inv).to_owned();
 
-            let mut inv_ids = self.inv_ids.lock().await;
-            if inv_ids.contains(&inv.id) {
+            if !self.buffers.seen_ids.push(inv.id).await {
                 continue
             }
-            inv_ids.push(inv.id);
-            drop(inv_ids);
 
             let mut inv_requested = vec![];
             for inv_object in inv.invs.iter() {
-                let msgs = &mut self.buffers.unread_msgs.lock().await.msgs;
-                if let Some(msg) = msgs.get_mut(&inv_object.0) {
-                    msg.read_confirms += 1;
-                } else {
+                if !self.buffers.unread_msgs.inc_read_confirms(&inv_object.0).await {
                     inv_requested.push(inv_object.clone());
                 }
             }
@@ -149,14 +134,11 @@ impl ProtocolPrivmsg {
             let msg = self.msg_sub.receive().await?;
             let mut msg = (*msg).to_owned();
 
-            let mut msg_ids = self.buffers.seen_ids.lock().await;
-            if msg_ids.contains(&msg.id) {
+            if !self.buffers.seen_ids.push(msg.id).await {
                 continue
             }
-            msg_ids.push(msg.id);
-            drop(msg_ids);
 
-            if msg.read_confirms >= MAX_CONFIRM {
+            if msg.read_confirms >= settings::MAX_CONFIRM {
                 self.add_to_msgs(&msg).await?;
             } else {
                 msg.read_confirms += 1;
@@ -177,12 +159,9 @@ impl ProtocolPrivmsg {
 
             self.update_unread_msgs().await?;
 
-            let privmsgs = self.buffers.privmsgs.lock().await;
-            let self_last_term = privmsgs.last_term();
-
-            match self_last_term.cmp(&last_term) {
+            match self.buffers.privmsgs.last_term().await.cmp(&last_term) {
                 Ordering::Less => {
-                    for msg in privmsgs.fetch_msgs(last_term) {
+                    for msg in self.buffers.privmsgs.fetch_msgs(last_term).await {
                         self.channel.send(msg).await?;
                     }
                 }
@@ -197,9 +176,8 @@ impl ProtocolPrivmsg {
             let getdata = self.getdata_sub.receive().await?;
             let getdata = (*getdata).to_owned();
 
-            let msgs = &self.buffers.unread_msgs.lock().await.msgs;
             for inv in getdata.invs {
-                if let Some(msg) = msgs.get(&inv.0) {
+                if let Some(msg) = self.buffers.unread_msgs.get(&inv.0).await {
                     self.channel.send(msg.clone()).await?;
                 }
             }
@@ -207,26 +185,25 @@ impl ProtocolPrivmsg {
     }
 
     async fn add_to_unread_msgs(&self, msg: &Privmsg) -> String {
-        self.buffers.unread_msgs.lock().await.insert(msg)
+        self.buffers.unread_msgs.insert(msg).await
     }
 
     async fn update_unread_msgs(&self) -> Result<()> {
-        let msgs = &mut self.buffers.unread_msgs.lock().await.msgs;
-        for (hash, msg) in msgs.clone() {
-            if msg.timestamp + UNREAD_MSG_EXPIRE_TIME < Utc::now().timestamp() {
-                msgs.remove(&hash);
+        for (hash, msg) in self.buffers.unread_msgs.load().await {
+            if msg.timestamp + settings::UNREAD_MSG_EXPIRE_TIME < Utc::now().timestamp() {
+                self.buffers.unread_msgs.remove(&hash).await;
                 continue
             }
-            if msg.read_confirms >= MAX_CONFIRM {
+            if msg.read_confirms >= settings::MAX_CONFIRM {
                 self.add_to_msgs(&msg).await?;
-                msgs.remove(&hash);
+                self.buffers.unread_msgs.remove(&hash).await;
             }
         }
         Ok(())
     }
 
     async fn add_to_msgs(&self, msg: &Privmsg) -> Result<()> {
-        self.buffers.privmsgs.lock().await.push(msg);
+        self.buffers.privmsgs.push(msg).await;
         self.notify.send(msg.clone()).await?;
         Ok(())
     }
@@ -239,11 +216,9 @@ impl net::ProtocolBase for ProtocolPrivmsg {
     /// waits for pong reply. Waits for ping and replies with a pong.
     async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
         // once a channel get started
-        let msgs_buffer = self.buffers.privmsgs.lock().await;
-        for m in msgs_buffer.iter() {
-            self.channel.send(m.clone()).await?;
+        for m in self.buffers.privmsgs.load().await {
+            self.channel.send(m).await?;
         }
-        drop(msgs_buffer);
 
         debug!(target: "ircd", "ProtocolPrivmsg::start() [START]");
         self.jobsman.clone().start(executor.clone());

bin/ircd/src/settings.rs

@@ -9,9 +9,34 @@ use url::Url;
 
 use darkfi::{net::settings::SettingsOpt, Result};
 
+// Location for config file
 pub const CONFIG_FILE: &str = "ircd_config.toml";
 pub const CONFIG_FILE_CONTENTS: &str = include_str!("../ircd_config.toml");
 
+// Buffers and ordering configuration
+pub const SIZE_OF_MSGS_BUFFER: usize = 4095;
+pub const SIZE_OF_IDSS_BUFFER: usize = 16384;
+pub const LIFETIME_FOR_ORPHAN: i64 = 600;
+pub const TERM_MAX_TIME_DIFFERENCE: i64 = 180;
+pub const BROADCAST_LAST_TERM_MSG: u64 = 4;
+
+// Msg config
+pub const MAXIMUM_LENGTH_OF_MESSAGE: usize = 1024;
+pub const MAXIMUM_LENGTH_OF_NICKNAME: usize = 32;
+
+// Protocol config
+pub const MAX_CONFIRM: u8 = 4;
+pub const UNREAD_MSG_EXPIRE_TIME: i64 = 18000;
+pub const TIMEOUT_FOR_RESEND_UNREAD_MSGS: u64 = 240;
+
+// IRC Client
+pub enum RPL {
+    NoTopic = 331,
+    Topic = 332,
+    NameReply = 353,
+    EndOfNames = 366,
+}
+
 /// ircd cli
 #[derive(Clone, Debug, Deserialize, StructOpt, StructOptToml)]
 #[serde(default)]