bin/ircd: new implement for PrivmsgBuffer

bin/ircd/Cargo.toml

@@ -34,6 +34,7 @@ simplelog = "0.12.0"
 fxhash = "0.2.1"
 ctrlc = { version = "3.2.3", features = ["termination"] }
 url = "2.2.2"
+chrono = "0.4.22"
 
 # Encoding and parsing
 serde_json = "1.0.85"

bin/ircd/src/buffers.rs

@@ -1,22 +1,26 @@
 use async_std::sync::{Arc, Mutex};
-use std::collections::VecDeque;
+use std::{cmp::Ordering, collections::VecDeque};
 
-use crate::{Privmsg, SIZE_OF_MSGS_BUFFER};
+use chrono::Utc;
 
+use crate::Privmsg;
+
+pub const SIZE_OF_MSGS_BUFFER: usize = 8191;
+pub const LIFETIME_FOR_ORPHAN: i64 = 600;
+
+#[derive(Clone)]
 pub struct RingBuffer<T> {
     pub items: VecDeque<T>,
-    pub size: usize,
 }
 
-impl<T: Eq + PartialEq> RingBuffer<T> {
+impl<T: Eq + PartialEq + Clone> RingBuffer<T> {
     pub fn new(capacity: usize) -> Self {
         let items = VecDeque::with_capacity(capacity);
-        let size = items.capacity();
-        Self { items, size }
+        Self { items }
     }
 
     pub fn push(&mut self, val: T) {
-        if self.items.len() == self.size {
+        if self.items.len() == self.items.capacity() {
             self.items.pop_front();
         }
         self.items.push_back(val);
@@ -25,38 +29,132 @@ impl<T: Eq + PartialEq> RingBuffer<T> {
     pub fn contains(&self, val: &T) -> bool {
         self.items.contains(val)
     }
+
+    pub fn len(&self) -> usize {
+        self.items.len()
+    }
+
+    pub fn as_slice(&mut self) -> &mut [T] {
+        self.items.make_contiguous()
+    }
+
+    pub fn iter(&self) -> impl Iterator<Item = &T> {
+        self.items.iter()
+    }
+
+    pub fn remove(&mut self, val: &T) -> Option<T> {
+        if let Some(index) = self.items.iter().position(|v| v == val) {
+            self.items.remove(index)
+        } else {
+            None
+        }
+    }
 }
 
 pub type SeenMsgIds = Arc<Mutex<RingBuffer<u64>>>;
 
 pub type ArcPrivmsgsBuffer = Arc<Mutex<PrivmsgsBuffer>>;
 
-pub struct PrivmsgsBuffer(RingBuffer<Privmsg>);
+pub struct PrivmsgsBuffer {
+    buffer: RingBuffer<Privmsg>,
+    orphans: RingBuffer<Orphan>,
+}
 
 impl PrivmsgsBuffer {
     pub fn new() -> ArcPrivmsgsBuffer {
-        Arc::new(Mutex::new(Self(RingBuffer::new(SIZE_OF_MSGS_BUFFER))))
+        Arc::new(Mutex::new(Self {
+            buffer: RingBuffer::new(SIZE_OF_MSGS_BUFFER),
+            orphans: RingBuffer::new(SIZE_OF_MSGS_BUFFER),
+        }))
+    }
+
+    pub fn push(&mut self, privmsg: &Privmsg) {
+        if self.buffer.contains(privmsg) {
+            return
+        }
+
+        match privmsg.term.cmp(&(self.last_term() + 1)) {
+            Ordering::Equal => self.buffer.push(privmsg.clone()),
+            Ordering::Greater => self.orphans.push(Orphan::new(privmsg)),
+            Ordering::Less => {
+                if !self.term_exist(privmsg.term) {
+                    self.orphans.push(Orphan::new(privmsg))
+                }
+            }
+        }
     }
 
-    pub fn push(&mut self, _privmsg: &Privmsg) {
-        // TODO
+    pub fn iter(&self) -> impl Iterator<Item = &Privmsg> {
+        self.buffer.iter()
     }
 
     pub fn last_term(&self) -> u64 {
-        match self.0.items.len() {
+        match self.buffer.len() {
             0 => 0,
-            n => self.0.items[n - 1].term,
+            n => self.buffer.items[n - 1].term,
         }
     }
 
-    pub fn to_vec(&self) -> Vec<Privmsg> {
-        self.0.items.clone().into()
+    pub fn update(&mut self) {
+        self.sort_orphans();
+        self.push_orphans();
+        self.sort();
+    }
+
+    fn term_exist(&self, term: u64) -> bool {
+        self.buffer.items.iter().find(|p| p.term == term).is_some()
+    }
+
+    fn sort(&mut self) {
+        self.buffer.as_slice().sort_by(|a, b| a.term.cmp(&b.term));
+    }
+
+    fn sort_orphans(&mut self) {
+        self.orphans.as_slice().sort_by(|a, b| a.msg.term.cmp(&b.msg.term));
+    }
+
+    fn push_orphans(&mut self) {
+        for orphan in self.orphans.clone().iter() {
+            let privmsg = orphan.msg.clone();
+            match privmsg.term.cmp(&(self.last_term() + 1)) {
+                Ordering::Equal => {
+                    self.buffer.push(privmsg.clone());
+                    self.orphans.remove(orphan);
+                }
+                Ordering::Less => {
+                    if !self.term_exist(privmsg.term) {
+                        self.buffer.push(privmsg.clone());
+                    }
+                    self.orphans.remove(orphan);
+                }
+                Ordering::Greater => {
+                    if (orphan.timestamp + LIFETIME_FOR_ORPHAN) < Utc::now().timestamp() {
+                        self.orphans.remove(orphan);
+                    }
+                }
+            }
+        }
+    }
+}
+
+#[derive(Clone, PartialEq, Eq)]
+struct Orphan {
+    msg: Privmsg,
+    timestamp: i64,
+}
+
+impl Orphan {
+    fn new(privmsg: &Privmsg) -> Self {
+        Self { msg: privmsg.clone(), timestamp: Utc::now().timestamp() }
     }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::Privmsg;
+    use rand::{seq::SliceRandom, thread_rng};
+
     #[test]
     fn test_ring_buffer() {
         let mut b = RingBuffer::<&str>::new(3);
@@ -64,9 +162,100 @@ mod tests {
         b.push("h2");
         b.push("h3");
         assert_eq!(b.items, vec!["h1", "h2", "h3"]);
-        assert_eq!(b.items.capacity(), b.size);
+        assert_eq!(b.items.capacity(), 3);
         b.push("h4");
         assert_eq!(b.items, vec!["h2", "h3", "h4"]);
-        assert_eq!(b.items.capacity(), b.size);
+        assert_eq!(b.items.capacity(), 3);
+        b.push("h5");
+        b.push("h6");
+        b.push("h7");
+        b.push("h8");
+        b.push("h9");
+        assert_eq!(b.len(), 3);
+        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),
+        };
+
+        //
+        // Fill the buffer with random generated terms in range 0..3001
+        //
+        let mut terms: Vec<u64> = (1..3001).collect();
+        terms.shuffle(&mut thread_rng());
+
+        for term in terms {
+            let privmsg = Privmsg::new("nick", "#dev", &format!("message_{}", term), term);
+            pms.push(&privmsg);
+        }
+
+        pms.update();
+
+        assert_eq!(pms.buffer.len(), 3000);
+        assert_eq!(pms.last_term(), 3000);
+        assert_eq!(pms.orphans.len(), 0);
+
+        //
+        // Fill the buffer with random generated terms in range 2000..4001
+        // Since the buffer len now is 3000 it will take only the terms from
+        // 3001 to 4000 without overwriting
+        //
+        let mut terms: Vec<u64> = (2000..4001).collect();
+        terms.shuffle(&mut thread_rng());
+
+        for term in terms {
+            let privmsg = Privmsg::new("nick", "#dev", &format!("message_{}", term), term);
+            pms.push(&privmsg);
+        }
+
+        pms.update();
+
+        assert_eq!(pms.buffer.len(), 4000);
+        assert_eq!(pms.last_term(), 4000);
+        assert_eq!(pms.orphans.len(), 0);
+
+        //
+        // Fill the buffer with random generated terms in range 4000..9001
+        // Since the buffer max size is SIZE_OF_MSGS_BUFFER it has to remove the old msges
+        //
+        let mut terms: Vec<u64> = (4001..9001).collect();
+        terms.shuffle(&mut thread_rng());
+
+        for term in terms {
+            let privmsg = Privmsg::new("nick", "#dev", &format!("message_{}", term), term);
+            pms.push(&privmsg);
+        }
+
+        pms.update();
+
+        assert_eq!(pms.buffer.len(), SIZE_OF_MSGS_BUFFER);
+        assert_eq!(pms.last_term(), 9000);
+        assert_eq!(pms.orphans.len(), 0);
+
+        //
+        // Fill the buffer with random generated terms in range 9001..11001
+        // This will occasionally update the buffer
+        // At the end, the messages in the buffer have to be in correct order
+        //
+        let mut terms: Vec<u64> = (9001..11001).collect();
+        terms.shuffle(&mut thread_rng());
+
+        for term in terms {
+            let privmsg = Privmsg::new("nick", "#dev", &format!("message_{}", term), term);
+            pms.push(&privmsg);
+            if rand::random() {
+                pms.update();
+            }
+        }
+
+        pms.update();
+
+        assert_eq!(pms.buffer.len(), SIZE_OF_MSGS_BUFFER);
+        assert_eq!(pms.last_term(), 11000);
+        assert_eq!(pms.orphans.len(), 0);
     }
 }

bin/ircd/src/main.rs

@@ -49,7 +49,6 @@ use crate::{
 };
 
 const SIZE_OF_MSG_IDSS_BUFFER: usize = 65536;
-pub const SIZE_OF_MSGS_BUFFER: usize = 4096;
 pub const MAXIMUM_LENGTH_OF_MESSAGE: usize = 1024;
 pub const MAXIMUM_LENGTH_OF_NICKNAME: usize = 32;
 

bin/ircd/src/privmsg.rs

@@ -18,10 +18,17 @@ pub struct Privmsg {
 }
 
 impl Privmsg {
-    pub fn new(nickname: String, target: String, message: String, term: u64) -> Self {
+    pub fn new(nickname: &str, target: &str, message: &str, term: u64) -> Self {
         let id = OsRng.next_u64();
         let timestamp = Timestamp::current_time();
-        Self { id, nickname, target, message, timestamp, term }
+        Self {
+            id,
+            nickname: nickname.to_string(),
+            target: target.to_string(),
+            message: message.to_string(),
+            timestamp,
+            term,
+        }
     }
 }
 

bin/ircd/src/protocol_privmsg.rs

@@ -51,12 +51,12 @@ impl ProtocolPrivmsg {
         let exclude_list = vec![self.channel.address()];
 
         // once a channel get started
-        let msgs_buffer = self.msgs.lock().await;
-        let msgs = msgs_buffer.to_vec();
-        drop(msgs_buffer);
-        for m in msgs {
+        let mut msgs_buffer = self.msgs.lock().await;
+        msgs_buffer.update();
+        for m in msgs_buffer.iter() {
             self.channel.send(m.clone()).await?;
         }
+        drop(msgs_buffer);
 
         loop {
             let msg = self.msg_sub.receive().await?;

bin/ircd/src/server/command.rs

@@ -206,7 +206,7 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcServerConnection<C>
 
         info!("(Plain) PRIVMSG {} :{}", target, message);
 
-        let mut privmsg = Privmsg::new(self.nickname.clone(), target.to_string(), message, 0);
+        let mut privmsg = Privmsg::new(&self.nickname, target, &message, 0);
 
         if target.starts_with('#') {
             if !self.configured_chans.contains_key(target) {
@@ -274,9 +274,9 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcServerConnection<C>
 
             // Send messages in buffer
             if !self.capabilities.get("no-history").unwrap() {
-                for msg in self.privmsgs_buffer.lock().await.to_vec() {
+                for msg in self.privmsgs_buffer.lock().await.iter() {
                     if msg.target == *chan {
-                        self.senders.notify_by_id(msg, self.subscriber_id).await;
+                        self.senders.notify_by_id(msg.clone(), self.subscriber_id).await;
                     }
                 }
             }

bin/ircd/src/server/mod.rs

@@ -186,14 +186,17 @@ impl<C: AsyncRead + AsyncWrite + Send + Unpin + 'static> IrcServerConnection<C>
             }
 
             // Send dm messages in buffer
-            for msg in self.privmsgs_buffer.lock().await.to_vec() {
+            let mut privmsgs_buffer = self.privmsgs_buffer.lock().await;
+            privmsgs_buffer.update();
+            for msg in privmsgs_buffer.iter() {
                 let is_dm = msg.target == self.nickname ||
                     (msg.nickname == self.nickname && !msg.target.starts_with('#'));
 
                 if is_dm {
-                    self.senders.notify_by_id(msg, self.subscriber_id).await;
+                    self.senders.notify_by_id(msg.clone(), self.subscriber_id).await;
                 }
             }
+            drop(privmsgs_buffer);
         }
         Ok(())
     }