Line data Source code
1 : //! Simple test to verify that simulator is working.
2 : #[cfg(test)]
3 : mod reliable_copy_test {
4 : use anyhow::Result;
5 : use desim::executor::{self, PollSome};
6 : use desim::options::{Delay, NetworkOptions};
7 : use desim::proto::{NetEvent, NodeEvent, ReplCell};
8 : use desim::world::{NodeId, World};
9 : use desim::{node_os::NodeOs, proto::AnyMessage};
10 : use parking_lot::Mutex;
11 : use std::sync::Arc;
12 : use tracing::info;
13 :
14 : /// Disk storage trait and implementation.
15 : pub trait Storage<T> {
16 : fn flush_pos(&self) -> u32;
17 : fn flush(&mut self) -> Result<()>;
18 : fn write(&mut self, t: T);
19 : }
20 :
21 : #[derive(Clone)]
22 : pub struct SharedStorage<T> {
23 : pub state: Arc<Mutex<InMemoryStorage<T>>>,
24 : }
25 :
26 : impl<T> SharedStorage<T> {
27 20 : pub fn new() -> Self {
28 20 : Self {
29 20 : state: Arc::new(Mutex::new(InMemoryStorage::new())),
30 20 : }
31 20 : }
32 : }
33 :
34 : impl<T> Storage<T> for SharedStorage<T> {
35 1049 : fn flush_pos(&self) -> u32 {
36 1049 : self.state.lock().flush_pos
37 1049 : }
38 :
39 100 : fn flush(&mut self) -> Result<()> {
40 100 : executor::yield_me(0);
41 100 : self.state.lock().flush()
42 100 : }
43 :
44 100 : fn write(&mut self, t: T) {
45 100 : executor::yield_me(0);
46 100 : self.state.lock().write(t);
47 100 : }
48 : }
49 :
50 : pub struct InMemoryStorage<T> {
51 : pub data: Vec<T>,
52 : pub flush_pos: u32,
53 : }
54 :
55 : impl<T> InMemoryStorage<T> {
56 20 : pub fn new() -> Self {
57 20 : Self {
58 20 : data: Vec::new(),
59 20 : flush_pos: 0,
60 20 : }
61 20 : }
62 :
63 100 : pub fn flush(&mut self) -> Result<()> {
64 100 : self.flush_pos = self.data.len() as u32;
65 100 : Ok(())
66 100 : }
67 :
68 100 : pub fn write(&mut self, t: T) {
69 100 : self.data.push(t);
70 100 : }
71 : }
72 :
73 : /// Server implementation.
74 20 : pub fn run_server(os: NodeOs, mut storage: Box<dyn Storage<u32>>) {
75 20 : info!("started server");
76 :
77 20 : let node_events = os.node_events();
78 20 : let mut epoll_vec: Vec<Box<dyn PollSome>> = vec![Box::new(node_events.clone())];
79 20 : let mut sockets = vec![];
80 :
81 : loop {
82 1537 : let index = executor::epoll_chans(&epoll_vec, -1).unwrap();
83 1537 :
84 1537 : if index == 0 {
85 568 : let node_event = node_events.must_recv();
86 568 : info!("got node event: {:?}", node_event);
87 568 : if let NodeEvent::Accept(tcp) = node_event {
88 568 : tcp.send(AnyMessage::Just32(storage.flush_pos()));
89 568 : epoll_vec.push(Box::new(tcp.recv_chan()));
90 568 : sockets.push(tcp);
91 568 : }
92 568 : continue;
93 969 : }
94 969 :
95 969 : let recv_chan = sockets[index - 1].recv_chan();
96 969 : let socket = &sockets[index - 1];
97 969 :
98 969 : let event = recv_chan.must_recv();
99 969 : info!("got event: {:?}", event);
100 381 : if let NetEvent::Message(AnyMessage::ReplCell(cell)) = event {
101 381 : if cell.seqno != storage.flush_pos() {
102 281 : info!("got out of order data: {:?}", cell);
103 281 : continue;
104 100 : }
105 100 : storage.write(cell.value);
106 100 : storage.flush().unwrap();
107 100 : socket.send(AnyMessage::Just32(storage.flush_pos()));
108 568 : }
109 : }
110 : }
111 :
112 : /// Client copies all data from array to the remote node.
113 20 : pub fn run_client(os: NodeOs, data: &[ReplCell], dst: NodeId) {
114 20 : info!("started client");
115 :
116 20 : let mut delivered = 0;
117 20 :
118 20 : let mut sock = os.open_tcp(dst);
119 20 : let mut recv_chan = sock.recv_chan();
120 :
121 1046 : while delivered < data.len() {
122 1026 : let num = &data[delivered];
123 1026 : info!("sending data: {:?}", num.clone());
124 1026 : sock.send(AnyMessage::ReplCell(num.clone()));
125 1026 :
126 1026 : // loop {
127 1026 : let event = recv_chan.recv();
128 128 : match event {
129 128 : NetEvent::Message(AnyMessage::Just32(flush_pos)) => {
130 128 : if flush_pos == 1 + delivered as u32 {
131 100 : delivered += 1;
132 100 : }
133 : }
134 : NetEvent::Closed => {
135 898 : info!("connection closed, reestablishing");
136 898 : sock = os.open_tcp(dst);
137 898 : recv_chan = sock.recv_chan();
138 : }
139 0 : _ => {}
140 : }
141 :
142 : // }
143 : }
144 :
145 20 : let sock = os.open_tcp(dst);
146 120 : for num in data {
147 100 : info!("sending data: {:?}", num.clone());
148 100 : sock.send(AnyMessage::ReplCell(num.clone()));
149 : }
150 :
151 20 : info!("sent all data and finished client");
152 20 : }
153 :
154 : /// Run test simulations.
155 : #[test]
156 1 : fn sim_example_reliable_copy() {
157 1 : utils::logging::init(
158 1 : utils::logging::LogFormat::Test,
159 1 : utils::logging::TracingErrorLayerEnablement::Disabled,
160 1 : utils::logging::Output::Stdout,
161 1 : )
162 1 : .expect("logging init failed");
163 1 :
164 1 : let delay = Delay {
165 1 : min: 1,
166 1 : max: 60,
167 1 : fail_prob: 0.4,
168 1 : };
169 1 :
170 1 : let network = NetworkOptions {
171 1 : keepalive_timeout: Some(50),
172 1 : connect_delay: delay.clone(),
173 1 : send_delay: delay.clone(),
174 1 : };
175 :
176 21 : for seed in 0..20 {
177 20 : let u32_data: [u32; 5] = [1, 2, 3, 4, 5];
178 20 : let data = u32_to_cells(&u32_data, 1);
179 20 : let world = Arc::new(World::new(seed, Arc::new(network.clone())));
180 20 :
181 20 : start_simulation(Options {
182 20 : world,
183 20 : time_limit: 1_000_000,
184 20 : client_fn: Box::new(move |os, server_id| run_client(os, &data, server_id)),
185 20 : u32_data,
186 20 : });
187 20 : }
188 1 : }
189 :
190 : pub struct Options {
191 : pub world: Arc<World>,
192 : pub time_limit: u64,
193 : pub u32_data: [u32; 5],
194 : pub client_fn: Box<dyn FnOnce(NodeOs, u32) + Send + 'static>,
195 : }
196 :
197 20 : pub fn start_simulation(options: Options) {
198 20 : let world = options.world;
199 20 :
200 20 : let client_node = world.new_node();
201 20 : let server_node = world.new_node();
202 20 : let server_id = server_node.id;
203 20 :
204 20 : // start the client thread
205 20 : client_node.launch(move |os| {
206 20 : let client_fn = options.client_fn;
207 20 : client_fn(os, server_id);
208 20 : });
209 20 :
210 20 : // start the server thread
211 20 : let shared_storage = SharedStorage::new();
212 20 : let server_storage = shared_storage.clone();
213 20 : server_node.launch(move |os| run_server(os, Box::new(server_storage)));
214 :
215 7830 : while world.step() && world.now() < options.time_limit {}
216 :
217 20 : let disk_data = shared_storage.state.lock().data.clone();
218 20 : assert!(verify_data(&disk_data, &options.u32_data[..]));
219 20 : }
220 :
221 20 : pub fn u32_to_cells(data: &[u32], client_id: u32) -> Vec<ReplCell> {
222 20 : let mut res = Vec::new();
223 100 : for (i, _) in data.iter().enumerate() {
224 100 : res.push(ReplCell {
225 100 : client_id,
226 100 : seqno: i as u32,
227 100 : value: data[i],
228 100 : });
229 100 : }
230 20 : res
231 20 : }
232 :
233 20 : fn verify_data(disk_data: &[u32], data: &[u32]) -> bool {
234 20 : if disk_data.len() != data.len() {
235 0 : return false;
236 20 : }
237 100 : for i in 0..data.len() {
238 100 : if disk_data[i] != data[i] {
239 0 : return false;
240 100 : }
241 : }
242 20 : true
243 20 : }
244 : }
|
