Implement add_keypair semantics

src/router/device.rs

@@ -1,24 +1,33 @@
-use arraydeque::{ArrayDeque, Wrapping};
+use arraydeque::{ArrayDeque, Saturating, Wrapping};
+use lifeguard::{Pool, Recycled};
 use treebitmap::IpLookupTable;
 
-use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
-use std::sync::atomic::{AtomicPtr, AtomicU64};
-use std::sync::{Arc, Mutex};
-use std::time::Instant;
+use std::collections::HashMap;
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
+use std::sync::atomic::{AtomicPtr, AtomicU64, Ordering};
+use std::sync::{Arc, Mutex, Weak};
+use std::time::{Duration, Instant};
+
+use spin::RwLock;
 
 use super::super::types::KeyPair;
 use super::anti_replay::AntiReplay;
 
+use std::u64;
+
+const REJECT_AFTER_MESSAGES: u64 = u64::MAX - (1 << 4);
 const MAX_STAGED_PACKETS: usize = 128;
 
-pub struct Device {
-    ipv4: IpLookupTable<Ipv4Addr, Arc<Peer>>,
-    ipv6: IpLookupTable<Ipv6Addr, Arc<Peer>>,
+pub struct Device<'a> {
+    recv: RwLock<HashMap<u32, Arc<Peer<'a>>>>, // map receiver id -> peer
+    ipv4: IpLookupTable<Ipv4Addr, Arc<Peer<'a>>>, // ipv4 trie
+    ipv6: IpLookupTable<Ipv6Addr, Arc<Peer<'a>>>, // ipv6 trie
+    pool: Pool<Vec<u8>>,                       // message buffer pool
 }
 
 struct KeyState(KeyPair, AntiReplay);
 
-struct EncryptState {
+struct EncryptionState {
     key: [u8; 32],    // encryption key
     id: u64,          // sender id
     nonce: AtomicU64, // next available nonce
@@ -32,28 +41,119 @@ struct KeyWheel {
     previous: AtomicPtr<Arc<Option<KeyState>>>, // old key state (used for decryption)
 }
 
-pub struct Peer {
+pub struct Peer<'a> {
+    inorder: Mutex<ArrayDeque<[Option<Recycled<'a, Vec<u8>>>; MAX_STAGED_PACKETS], Saturating>>, // inorder queue
     staged_packets: Mutex<ArrayDeque<[Vec<u8>; MAX_STAGED_PACKETS], Wrapping>>, // packets awaiting handshake
     rx_bytes: AtomicU64,                                                        // received bytes
     tx_bytes: AtomicU64,                                                        // transmitted bytes
     keys: KeyWheel,                                                             // key-wheel
-    encryption: AtomicPtr<Arc<EncryptState>>, // current encryption key (starts expired)
+    ekey: AtomicPtr<Arc<EncryptionState>>,                                      // encryption state
+    endpoint: AtomicPtr<Arc<Option<SocketAddr>>>,
 }
 
-pub struct PeerRef();
+impl<'a> Peer<'a> {
+    pub fn set_endpoint(&self, endpoint: SocketAddr) {
+        self.endpoint
+            .store(&mut Arc::new(Some(endpoint)), Ordering::Relaxed)
+    }
 
-impl Device {
-    pub fn new() -> Device {
-        unimplemented!();
+    pub fn add_keypair(&self, keypair: KeyPair) {
+        let confirmed = keypair.confirmed;
+        let mut st_new = Arc::new(Some(KeyState(keypair, AntiReplay::new())));
+        let st_previous = self.keys.previous.load(Ordering::Relaxed);
+        if confirmed {
+            // previous <- current
+            self.keys.previous.compare_and_swap(
+                st_previous,
+                self.keys.current.load(Ordering::Relaxed),
+                Ordering::Relaxed,
+            );
+
+            // current  <- new
+            self.keys.next.store(&mut st_new, Ordering::Relaxed)
+        } else {
+            // previous <- next
+            self.keys.previous.compare_and_swap(
+                st_previous,
+                self.keys.next.load(Ordering::Relaxed),
+                Ordering::Relaxed,
+            );
+
+            // next <- new
+            self.keys.next.store(&mut st_new, Ordering::Relaxed)
+        }
+    }
+
+    pub fn rx_bytes(&self) -> u64 {
+        self.rx_bytes.load(Ordering::Relaxed)
+    }
+
+    pub fn tx_bytes(&self) -> u64 {
+        self.tx_bytes.load(Ordering::Relaxed)
+    }
+}
+
+impl<'a> Device<'a> {
+    pub fn new() -> Device<'a> {
+        Device {
+            recv: RwLock::new(HashMap::new()),
+            ipv4: IpLookupTable::new(),
+            ipv6: IpLookupTable::new(),
+            pool: Pool::with_size_and_max(0, MAX_STAGED_PACKETS * 2),
+        }
+    }
+
+    pub fn subnets(&self, peer: Arc<Peer<'a>>) -> Vec<(IpAddr, u32)> {
+        let mut subnets = Vec::new();
+
+        // extract ipv4 entries
+        for subnet in self.ipv4.iter() {
+            let (ip, masklen, p) = subnet;
+            if Arc::ptr_eq(&peer, p) {
+                subnets.push((IpAddr::V4(ip), masklen))
+            }
+        }
+
+        // extract ipv6 entries
+        for subnet in self.ipv6.iter() {
+            let (ip, masklen, p) = subnet;
+            if Arc::ptr_eq(&peer, p) {
+                subnets.push((IpAddr::V6(ip), masklen))
+            }
+        }
+
+        subnets
     }
 
     /// Adds a new peer to the device
     ///
     /// # Returns
     ///
-    /// An opaque value representing the peer.
-    pub fn add(&self) -> PeerRef {
-        unimplemented!();
+    /// A atomic ref. counted peer (with liftime matching the device)
+    pub fn add(&mut self) -> Arc<Peer<'a>> {
+        Arc::new(Peer {
+            inorder: Mutex::new(ArrayDeque::new()),
+            staged_packets: Mutex::new(ArrayDeque::new()),
+            rx_bytes: AtomicU64::new(0),
+            tx_bytes: AtomicU64::new(0),
+            keys: KeyWheel {
+                next: AtomicPtr::new(&mut Arc::new(None)),
+                current: AtomicPtr::new(&mut Arc::new(None)),
+                previous: AtomicPtr::new(&mut Arc::new(None)),
+            },
+            // long expired encryption key
+            ekey: AtomicPtr::new(&mut Arc::new(EncryptionState {
+                key: [0u8; 32],
+                id: 0,
+                nonce: AtomicU64::new(REJECT_AFTER_MESSAGES),
+                death: Instant::now() - Duration::from_secs(31536000),
+            })),
+            endpoint: AtomicPtr::new(&mut Arc::new(None)),
+        })
+    }
+
+    pub fn get_buffer(&self) -> Recycled<Vec<u8>> {
+        self.pool.new()
     }
 
     /// Cryptkey routes and sends a plaintext message (IP packet)
@@ -68,7 +168,7 @@ impl Device {
     /// This indicates that a handshake should be initated (see the handshake module).
     /// If this occurs the packet is copied to an internal buffer
     /// and retransmission can be attempted using send_run_queue
-    pub fn send(&self, pt_msg: &mut [u8]) -> Option<PeerRef> {
+    pub fn send(&self, pt_msg: &mut [u8]) -> Arc<Peer> {
         unimplemented!();
     }
 
@@ -80,7 +180,7 @@ impl Device {
     ///
     /// - peer: Reference to the destination peer
     /// - msg: Message to transmit
-    pub fn send_raw(&self, peer: PeerRef, msg: &mut [u8]) {
+    pub fn send_raw(&self, peer: Arc<Peer>, msg: &mut [u8]) {
         unimplemented!();
     }
 
@@ -89,7 +189,7 @@ impl Device {
     /// # Arguments
     ///
     /// - peer: Reference for the peer to flush
-    pub fn flush_queue(&self, peer: PeerRef) {
+    pub fn flush_queue(&self, peer: Arc<Peer>) {
         unimplemented!();
     }
 
@@ -101,7 +201,7 @@ impl Device {
     ///
     /// A boolean indicating whether packages where sent.
     /// Note: This is used for implicit confirmation of handshakes.
-    pub fn send_run_queue(&self, peer: PeerRef) -> bool {
+    pub fn send_run_queue(&self, peer: Arc<Peer>) -> bool {
         unimplemented!();
     }
 
@@ -119,15 +219,15 @@ impl Device {
     /// # Arguments
     ///
     /// - peer: The peer to retrieve the endpoint for
-    pub fn get_endpoint(&self, peer: PeerRef) -> SocketAddr {
+    pub fn get_endpoint(&self, peer: Arc<Peer>) -> SocketAddr {
         unimplemented!();
     }
 
-    pub fn set_endpoint(&self, peer: PeerRef, endpoint: SocketAddr) {
+    pub fn set_endpoint(&self, peer: Arc<Peer>, endpoint: SocketAddr) {
         unimplemented!();
     }
 
-    pub fn new_keypair(&self, peer: PeerRef, keypair: KeyPair) {
+    pub fn new_keypair(&self, peer: Arc<Peer>, keypair: KeyPair) {
         unimplemented!();
     }
 }