d3-core 0.1.3

A framework for server development -- the core layer
Documentation 
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
139

use super::*;
use crossbeam::utils::Backoff;

// Machines cooperate in a collective. There are several primary traits
// that are presented here. Combined, they form the contract between
// the machines, allowing them to interact with each other.
//
// The MachineImpl designates an instruction set, which machines implement.
// It carries with it, two adapters and an instruction set. The ReceiverAdapter
// is the primary means of expressing the machine. The SenderAdapter is
// ancilary, used for parking a sender, which would otherwise block the
// executor.
//
// To aid in construction, there's a derive macro, MachineImpl, which should
// be used to designate the instruction sets which machines implement.
// A machine may implement one or more of these.
//
// Examples
//
//     #[derive(Copy, Clone, Debug, Eq, MachineImpl, PartialEq, SmartDefault)]
//     #[allow(dead_code)]
//     pub enum ActivationCommands {
//         #[default]
//         Start,
//         Stop,
//     }
//

// Each enum, that is an instuction set, derives MachineImpl.
#[doc(hidden)]
pub trait MachineImpl: 'static + Send + Sync {
    type Adapter;
    type SenderAdapter;
    type InstructionSet: Send + Sync;

    // Park a sender. If the sender can't be parked, the instruction
    // is returned.
    fn park_sender(
        channel_id: usize, receiver_machine: Weak<self::tls::collective::MachineAdapter>,
        sender: crossbeam::channel::Sender<Self::InstructionSet>, instruction: Self::InstructionSet,
    ) -> Result<(), Self::InstructionSet>;
}

/// The machine is the common trait all machines must implement
/// and describes how instuctions are delivered to a machine, via
/// the receive method.
pub trait Machine<T>: Send + Sync
where
    T: 'static + Send + Sync,
{
    /// The receive method receives instructions sent to it by itself or other machines.
    fn receive(&self, cmd: T);
    /// The disconnected method is called to notify the machine that its receiver has become
    /// disconnect; it will no longer receive instructions.
    /// This could be a result of server shutdown, or all senders dropping their senders.
    fn disconnected(&self) {}
    /// The connected method is called once, before receive messages. It provides a notification that the
    /// machine has become connected and may receive instructions. It includes a Uuid for the machine,
    /// which may be used in logging. A machine implementing several instruction sets will receive a differnt
    /// Uuid for each instruction set implemented.
    fn connected(&self, _uuid: Uuid) {}
}

// Adding the machine implementation to Mutex
// makes it possible to hide the underlying wrapping.
impl<T, P> Machine<P> for Mutex<T>
where
    T: Machine<P>,
    P: MachineImpl,
{
    // fn receive(&self, cmd: P) { self.lock().unwrap().receive(cmd); }
    // fn disconnected(&self) { self.lock().unwrap().disconnected(); }
    // fn connected(&self, uuid: Uuid) { self.lock().unwrap().connected(uuid); }

    // In order to prevent lockup of an executor, a try_lock is attempted, while
    // it should always obtain the lock, there may be data-races where it can't.
    // In those rare cases, a warning is logged and backoff is used. Eventually,
    // it will give up and re-schedule. This same approach is used for the disconnected()
    // and connected() methods.
    fn receive(&self, cmd: P) {
        if let Some(ref mut mutex) = self.try_lock() {
            (*mutex).receive(cmd);
        } else {
            log::warn!("try_lock failed for receive, retrying");
            let backoff = Backoff::new();
            loop {
                if let Some(ref mut mutex) = self.try_lock() {
                    (*mutex).receive(cmd);
                    return;
                } else if backoff.is_completed() {
                    log::error!("try_lock failed for receive, giving up after multiple retries");
                    return;
                } else {
                    backoff.snooze();
                }
            }
        }
    }

    fn disconnected(&self) {
        if let Some(ref mut mutex) = self.try_lock() {
            (*mutex).disconnected();
        } else {
            log::warn!("try_lock failed for disconnected, retrying");
            let backoff = Backoff::new();
            loop {
                if let Some(ref mut mutex) = self.try_lock() {
                    (*mutex).disconnected();
                    return;
                } else if backoff.is_completed() {
                    log::error!("try_lock failed for disconnected, giving up after multiple retries");
                    return;
                } else {
                    backoff.snooze();
                }
            }
        }
    }

    fn connected(&self, uuid: Uuid) {
        if let Some(ref mut mutex) = self.try_lock() {
            (*mutex).connected(uuid);
        } else {
            log::warn!("try_lock failed for connected, retrying");
            let backoff = Backoff::new();
            loop {
                if let Some(ref mut mutex) = self.try_lock() {
                    (*mutex).connected(uuid);
                    return;
                } else if backoff.is_completed() {
                    log::error!("try_lock failed for connected, giving up after multiple retries");
                    return;
                } else {
                    backoff.snooze();
                }
            }
        }
    }
}