runtime/
thread_statuses.rs

1use std::{collections::HashMap, sync::Arc, thread::sleep, time::Duration};
2
3use parking_lot::Mutex;
4
5use crate::{RuntimeError, Status};
6
7const WAIT_TIME: Duration = Duration::from_millis(100);
8
9/// Tracker for threads current `Status`s.
10#[derive(Debug, Clone)]
11pub struct ThreadStatuses {
12	statuses: Arc<Mutex<HashMap<String, Status>>>,
13}
14
15impl ThreadStatuses {
16	/// Create a new instance.
17	#[must_use]
18	#[inline]
19	pub fn new() -> Self {
20		Self {
21			statuses: Arc::new(Mutex::new(HashMap::new())),
22		}
23	}
24
25	/// Wait for a thread with a particular name to reach an expected `Status`.
26	///
27	/// # Errors
28	/// Will error if the wait times out.
29	#[inline]
30	pub fn wait_for_status(&self, thread_name: &str, expected_status: &Status) -> Result<(), RuntimeError> {
31		let mut attempt = 0;
32
33		loop {
34			let lock = self.statuses.lock();
35			let current = lock
36				.get(thread_name)
37				.ok_or_else(|| RuntimeError::ThreadNotRegistered(String::from(thread_name)))?;
38
39			if current == expected_status {
40				return Ok(());
41			}
42			drop(lock);
43
44			sleep(WAIT_TIME);
45			attempt += 1;
46
47			if attempt > 10 {
48				return Err(RuntimeError::ThreadWaitTimeout(String::from(thread_name)));
49			}
50		}
51	}
52
53	pub(crate) fn register_thread(&self, thread_name: &str, status: Status) {
54		assert!(
55			self.statuses.lock().insert(String::from(thread_name), status).is_none(),
56			"Attempt to register more than one threads with name: {thread_name}"
57		);
58	}
59
60	pub(crate) fn update_thread(&self, thread_name: &str, status: Status) {
61		let mut lock = self.statuses.lock();
62		let current = lock.entry(String::from(thread_name)).or_insert(Status::New);
63		if !matches!(*current, Status::Error(..))
64			&& !matches!(
65				status,
66				Status::RequestPause | Status::RequestResume | Status::RequestEnd
67			) {
68			*current = status;
69		}
70	}
71
72	pub(crate) fn all_ended(&self) -> bool {
73		self.statuses
74			.lock()
75			.values()
76			.all(|status| matches!(status, &(Status::Ended | Status::Error(_))))
77	}
78}
79
80#[cfg(test)]
81mod tests {
82	use std::{ops::Mul, thread};
83
84	use claim::{assert_err, assert_ok, assert_some_eq};
85	use testutils::assert_err_eq;
86
87	use super::*;
88
89	#[test]
90	fn wait_for_status_success_immediate() {
91		let statuses = ThreadStatuses::new();
92		statuses.register_thread("name", Status::New);
93		assert_ok!(statuses.wait_for_status("name", &Status::New));
94	}
95
96	#[test]
97	fn wait_for_status_success_after_wait() {
98		let statuses = ThreadStatuses::new();
99		statuses.register_thread("name", Status::New);
100		let thread_statuses = statuses.clone();
101		_ = thread::spawn(move || {
102			sleep(WAIT_TIME.mul(4));
103			thread_statuses.update_thread("name", Status::Ended);
104		});
105
106		assert_ok!(statuses.wait_for_status("name", &Status::Ended));
107	}
108
109	#[test]
110	fn wait_for_status_not_registered_error() {
111		let statuses = ThreadStatuses::new();
112		statuses.register_thread("name", Status::New);
113		assert_err_eq!(
114			statuses.wait_for_status("not-name", &Status::Ended),
115			RuntimeError::ThreadNotRegistered(String::from("not-name"))
116		);
117	}
118
119	#[test]
120	fn wait_for_status_timeout_error() {
121		let statuses = ThreadStatuses::new();
122		statuses.register_thread("name", Status::New);
123		assert_err!(statuses.wait_for_status("name", &Status::Ended));
124	}
125
126	#[test]
127	fn register_thread() {
128		let statuses = ThreadStatuses::new();
129		statuses.register_thread("name", Status::New);
130		assert_some_eq!(statuses.statuses.lock().get("name"), &Status::New);
131	}
132
133	#[test]
134	#[should_panic]
135	fn register_thread_same_name() {
136		let statuses = ThreadStatuses::new();
137		statuses.register_thread("name", Status::New);
138		statuses.register_thread("name", Status::New);
139		assert_some_eq!(statuses.statuses.lock().get("name"), &Status::New);
140	}
141
142	#[test]
143	fn update_thread() {
144		let statuses = ThreadStatuses::new();
145		statuses.register_thread("name", Status::New);
146		statuses.update_thread("name", Status::Busy);
147		assert_some_eq!(statuses.statuses.lock().get("name"), &Status::Busy);
148	}
149
150	#[test]
151	fn all_ended_one_not_ended() {
152		let statuses = ThreadStatuses::new();
153		statuses.register_thread("name", Status::New);
154		assert!(!statuses.all_ended());
155	}
156
157	#[test]
158	fn all_ended_one_ended() {
159		let statuses = ThreadStatuses::new();
160		statuses.register_thread("name", Status::Ended);
161		assert!(statuses.all_ended());
162	}
163
164	#[test]
165	fn all_ended_multiple_with_one_ended() {
166		let statuses = ThreadStatuses::new();
167		statuses.register_thread("name0", Status::New);
168		statuses.register_thread("name1", Status::Ended);
169		assert!(!statuses.all_ended());
170	}
171
172	#[test]
173	fn all_ended_multiple_with_all_ended() {
174		let statuses = ThreadStatuses::new();
175		statuses.register_thread("name0", Status::Ended);
176		statuses.register_thread("name1", Status::Ended);
177		assert!(statuses.all_ended());
178	}
179
180	#[test]
181	fn all_ended_with_error_state() {
182		let statuses = ThreadStatuses::new();
183		statuses.register_thread("name", Status::Error(RuntimeError::ThreadError(String::from("error"))));
184		assert!(statuses.all_ended());
185	}
186}
runtime/thread_statuses.rs

runtime/
thread_statuses.rs
