d3_core/scheduler/
machine.rs

1#![allow(dead_code)]
2use super::*;
3
4use self::setup_teardown::*;
5use crate::channel::sender::*;
6
7/// The connect method creates a machine, implementing an instruction set.
8/// The machine has a bound communication channel of a default size receiving those instructions.
9pub fn connect<T, P>(
10    machine: T,
11) -> (
12    Arc<Mutex<T>>,
13    Sender<<<P as MachineImpl>::Adapter as MachineBuilder>::InstructionSet>,
14)
15where
16    T: 'static + Machine<P> + Machine<<<P as MachineImpl>::Adapter as MachineBuilder>::InstructionSet>,
17    P: MachineImpl,
18    <P as MachineImpl>::Adapter: MachineBuilder,
19{
20    let channel_max = default_channel_max.load();
21    let (machine, mut sender, collective_adapter) = <<P as MachineImpl>::Adapter as MachineBuilder>::build_raw(machine, channel_max);
22    let adapter = Arc::new(collective_adapter);
23    sender.bind(Arc::downgrade(&adapter));
24    Server::assign_machine(adapter);
25    (machine, sender)
26}
27
28/// The and_connect method adds an additional instruction set and communication channel to the machine.
29/// The communicate channel ib bound to a default size.
30pub fn and_connect<T, P>(machine: &Arc<Mutex<T>>) -> Sender<<<P as MachineImpl>::Adapter as MachineBuilder>::InstructionSet>
31where
32    T: 'static + Machine<P> + Machine<<<P as MachineImpl>::Adapter as MachineBuilder>::InstructionSet>,
33    P: MachineImpl,
34    <P as MachineImpl>::Adapter: MachineBuilder,
35{
36    let channel_max = default_channel_max.load();
37    let (mut sender, collective_adapter) = <<P as MachineImpl>::Adapter as MachineBuilder>::build_addition(machine, channel_max);
38    let adapter = Arc::new(collective_adapter);
39    sender.bind(Arc::downgrade(&adapter));
40    Server::assign_machine(adapter);
41    sender
42}
43
44/// The connect_with_capacity method creates a machine with a bounded queue of the specified size.
45pub fn connect_with_capacity<T, P>(
46    machine: T, capacity: usize,
47) -> (
48    Arc<Mutex<T>>,
49    Sender<<<P as MachineImpl>::Adapter as MachineBuilder>::InstructionSet>,
50)
51where
52    T: 'static + Machine<P> + Machine<<<P as MachineImpl>::Adapter as MachineBuilder>::InstructionSet>,
53    P: MachineImpl,
54    <P as MachineImpl>::Adapter: MachineBuilder,
55{
56    let (machine, mut sender, collective_adapter) = <<P as MachineImpl>::Adapter as MachineBuilder>::build_raw(machine, capacity);
57    let adapter = Arc::new(collective_adapter);
58    sender.bind(Arc::downgrade(&adapter));
59    Server::assign_machine(adapter);
60    (machine, sender)
61}
62
63/// The and_connect_with_capacity method adds an additional instruction set and sender to the machine.
64/// The communication channel is bound to the specified size.
65pub fn and_connect_with_capacity<T, P>(
66    machine: &Arc<Mutex<T>>, capacity: usize,
67) -> Sender<<<P as MachineImpl>::Adapter as MachineBuilder>::InstructionSet>
68where
69    T: 'static + Machine<P> + Machine<<<P as MachineImpl>::Adapter as MachineBuilder>::InstructionSet>,
70    P: MachineImpl,
71    <P as MachineImpl>::Adapter: MachineBuilder,
72{
73    let (mut sender, collective_adapter) = <<P as MachineImpl>::Adapter as MachineBuilder>::build_addition(machine, capacity);
74    let adapter = Arc::new(collective_adapter);
75    sender.bind(Arc::downgrade(&adapter));
76    Server::assign_machine(adapter);
77    sender
78}
79
80/// The connect_unbounded method creates a machine with an unbounded queue. It can result
81/// in a panic if system resources become exhausted.
82pub fn connect_unbounded<T, P>(
83    machine: T,
84) -> (
85    Arc<Mutex<T>>,
86    Sender<<<P as MachineImpl>::Adapter as MachineBuilder>::InstructionSet>,
87)
88where
89    T: 'static + Machine<P> + Machine<<<P as MachineImpl>::Adapter as MachineBuilder>::InstructionSet>,
90    P: MachineImpl,
91    <P as MachineImpl>::Adapter: MachineBuilder,
92{
93    let (machine, mut sender, collective_adapter) = <<P as MachineImpl>::Adapter as MachineBuilder>::build_unbounded(machine);
94    let adapter = Arc::new(collective_adapter);
95    sender.bind(Arc::downgrade(&adapter));
96    Server::assign_machine(adapter);
97    (machine, sender)
98}
99
100/// The and_connect_unbounded method adds an additional instruction set and sender to the machine.
101/// The communication channel is unbound.
102pub fn and_connect_unbounded<T, P>(machine: &Arc<Mutex<T>>) -> Sender<<<P as MachineImpl>::Adapter as MachineBuilder>::InstructionSet>
103where
104    T: 'static + Machine<P> + Machine<<<P as MachineImpl>::Adapter as MachineBuilder>::InstructionSet>,
105    P: MachineImpl,
106    <P as MachineImpl>::Adapter: MachineBuilder,
107{
108    let (mut sender, collective_adapter) = <<P as MachineImpl>::Adapter as MachineBuilder>::build_addition_unbounded(machine);
109    let adapter = Arc::new(collective_adapter);
110    sender.bind(Arc::downgrade(&adapter));
111    Server::assign_machine(adapter);
112    sender
113}
114
115/// CHANNEL_MAX is the default size for bound communication channels.
116pub const CHANNEL_MAX: usize = 250;
117
118#[allow(dead_code)]
119#[allow(non_upper_case_globals)]
120/// The default_channel_max static is the default used for creating bound channels.
121pub static default_channel_max: AtomicCell<usize> = AtomicCell::new(CHANNEL_MAX);
122/// The get_default_channel_capacity function returns the default value.
123#[allow(dead_code)]
124pub fn get_default_channel_capacity() -> usize { default_channel_max.load() }
125/// The set_default_channel_capacity function sets a new default value.
126/// setting should be performed before starting the server.
127#[allow(dead_code)]
128pub fn set_default_channel_capacity(new: usize) { default_channel_max.store(new); }
d3_core/scheduler/machine.rs

d3_core/scheduler/
machine.rs
