aboutsummaryrefslogtreecommitdiffstats
path: root/src/wireguard/router/pool.rs
blob: 98b11441d7dae35ff921996814d59df76dbc39ac (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
use arraydeque::ArrayDeque;
use spin::{Mutex, MutexGuard};
use std::sync::mpsc::Receiver;
use std::sync::Arc;
use std::mem;

use super::runq::{RunQueue, ToKey};

const INORDER_QUEUE_SIZE: usize = 64;

pub struct InnerJob<P, B> {
    // peer (used by worker to schedule/handle inorder queue),
    // when the peer is None, the job is complete
    peer: Option<P>,
    pub body: B,
}

pub struct Job<P, B> {
    inner: Arc<Mutex<InnerJob<P, B>>>,
}

impl<P, B> Clone for Job<P, B> {
    fn clone(&self) -> Job<P, B> {
        Job {
            inner: self.inner.clone(),
        }
    }
}

impl<P, B> Job<P, B> {
    pub fn new(peer: P, body: B) -> Job<P, B> {
        Job {
            inner: Arc::new(Mutex::new(InnerJob {
                peer: Some(peer),
                body,
            })),
        }
    }
}

impl<P, B> Job<P, B> {
    /// Returns a mutex guard to the inner job if complete
    pub fn complete(&self) -> Option<MutexGuard<InnerJob<P, B>>> {
        self.inner
            .try_lock()
            .and_then(|m| if m.peer.is_none() { Some(m) } else { None })
    }
}

pub struct InorderQueue<P, B> {
    queue: Mutex<ArrayDeque<[Job<P, B>; INORDER_QUEUE_SIZE]>>,
}

impl<P, B> InorderQueue<P, B> {
    pub fn send(&self, job: Job<P, B>) -> bool {
        self.queue.lock().push_back(job).is_ok()
    }

    pub fn new() -> InorderQueue<P, B> {
        InorderQueue {
            queue: Mutex::new(ArrayDeque::new()),
        }
    }

    #[inline(always)]
    pub fn handle<F: Fn(&mut B)>(&self, f: F) {
        // take the mutex
        let mut queue = self.queue.lock();

        loop {
            // attempt to extract front element
            let front = queue.pop_front();
            let elem = match front {
                Some(elem) => elem,
                _ => {
                    return;
                }
            };
            
            // apply function if job complete
            let ret = if let Some(mut guard) = elem.complete() {
                mem::drop(queue);
                f(&mut guard.body);
                queue = self.queue.lock();
                false
            } else {
                true
            };

            // job not complete yet, return job to front
            if ret {
                queue.push_front(elem).unwrap();
                return;
            }
        }
    }
}

/// Allows easy construction of a semi-parallel worker.
/// Applicable for both decryption and encryption workers.
#[inline(always)]
pub fn worker_parallel<
    P : ToKey, // represents a peer (atomic reference counted pointer)
    B, // inner body type (message buffer, key material, ...)
    D, // device
    W: Fn(&P, &mut B),
    Q: Fn(&D) -> &RunQueue<P>,
>(
    device: D,
    queue: Q,
    receiver: Receiver<Job<P, B>>,
    work: W,
) {
    log::trace!("router worker started");
    loop {
        // handle new job
        let peer = {
            // get next job
            let job = match receiver.recv() {
                Ok(job) => job,
                _ => return,
            };

            // lock the job
            let mut job = job.inner.lock();

            // take the peer from the job
            let peer = job.peer.take().unwrap();

            // process job
            work(&peer, &mut job.body);
            peer
        };
        
        // process inorder jobs for peer
        queue(&device).insert(peer);
    }
}