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