feat: initial implementation for lock free queue

Cargo.toml

@@ -0,0 +1,6 @@
+[package]
+name = "concurrent-non-blocking-queue"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]

src/main.rs

@@ -0,0 +1,138 @@
+use std::ptr;
+use std::sync::atomic::{AtomicPtr, Ordering};
+
+#[derive(Debug)]
+struct Node<T> {
+    data: T,
+    next: AtomicPtr<Node<T>>,
+}
+
+#[derive(Debug)]
+pub struct LockFreeQueue<T> {
+    head: AtomicPtr<Node<T>>,
+    tail: AtomicPtr<Node<T>>
+}
+
+impl<T: std::default::Default> LockFreeQueue<T> {
+
+    pub fn new() -> Self {
+        let dummy_node = Box::into_raw(Box::new(Node {
+            data: Default::default(),
+            next: AtomicPtr::new(ptr::null_mut()),
+        }));
+
+        LockFreeQueue {
+            head: AtomicPtr::new(dummy_node),
+            tail: AtomicPtr::new(dummy_node),
+        }
+    }
+
+    pub fn offer(&self, data: T) {
+        let new_node = Box::into_raw(Box::new(Node {
+            data,
+            next: AtomicPtr::new(ptr::null_mut()),
+        }));
+        let mut tail = self.tail.load(Ordering::Relaxed);
+        let mut next;
+        loop {
+            unsafe {
+                next = (*tail).next.load(Ordering::Relaxed);
+
+                if next.is_null() {
+                    if (*tail)
+                        .next
+                        .compare_exchange(next, new_node, Ordering::Release, Ordering::Relaxed)
+                        .unwrap()
+                        == next
+                    {
+                        break;
+                    }
+                } else {
+                    self.tail.compare_exchange(tail, next, Ordering::Release, Ordering::Relaxed);
+                    tail = self.tail.load(Ordering::Relaxed);
+                }
+            }
+        }
+        self.tail.compare_exchange(tail, new_node, Ordering::Release, Ordering::Relaxed);
+    }
+
+    pub fn take(&self) -> Option<T> {
+        let mut head = self.head.load(Ordering::Relaxed);
+        let mut next;
+        loop {
+            unsafe {
+                next = (*head).next.load(Ordering::Relaxed);
+
+                if next.is_null() {
+                    return None;
+                }
+
+                if self
+                    .head
+                    .compare_and_swap(head, next, Ordering::Relaxed)
+                    == head
+                {
+                    let node = Box::from_raw(head);
+                    return Some(node.data);
+                }
+
+                head = self.head.load(Ordering::Relaxed);
+            }
+        }
+    }
+
+    pub fn is_empty(&self) -> bool {
+        let head = self.head.load(Ordering::Relaxed);
+        let next = unsafe { (*head).next.load(Ordering::Relaxed) };
+        next.is_null()
+    }
+}
+
+impl<T> Drop for LockFreeQueue<T> {
+
+    fn drop(&mut self) {
+        let mut node = self.head.load(Ordering::Relaxed);
+        while node != ptr::null_mut() {
+            let n = unsafe { Box::from_raw(node) };
+            node = n.next.load(Ordering::Relaxed);
+        }
+    }
+}
+
+
+#[cfg(test)]
+mod tests {
+    use crate::LockFreeQueue;
+
+    #[test]
+    fn test_is_empty() {
+        let queue = LockFreeQueue::new();
+
+        assert!(queue.is_empty());
+
+        queue.offer(1);
+        assert!(!queue.is_empty());
+
+        queue.take();
+        assert!(queue.is_empty());
+    }
+
+    #[test]
+    fn test_take_from_empty_queue() {
+        let queue: LockFreeQueue<i32> = LockFreeQueue::new();
+        assert_eq!(queue.take(), None);
+    }
+
+    #[test]
+    fn test_offer_take() {
+        let queue = LockFreeQueue::new();
+
+        queue.offer(1);
+        queue.offer(2);
+
+        assert_eq!(queue.take(), Some(0)); // TODO this dummy element
+        assert_eq!(queue.take(), Some(1));
+        assert_eq!(queue.take(), Some(2));
+        assert_eq!(queue.take(), None);
+    }
+}