Threading/Executable module #1

Merged
muellerr merged 12 commits from threading into main 2022-06-09 12:50:06 +02:00
4 changed files with 319 additions and 167 deletions
Showing only changes of commit 59103135d9 - Show all commits

View File

@ -0,0 +1,18 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="Docs" type="CargoCommandRunConfiguration" factoryName="Cargo Command">
<option name="command" value="doc" />
<option name="workingDirectory" value="file://$PROJECT_DIR$" />
<option name="channel" value="DEFAULT" />
<option name="requiredFeatures" value="true" />
<option name="allFeatures" value="false" />
<option name="emulateTerminal" value="false" />
<option name="withSudo" value="false" />
<option name="backtrace" value="SHORT" />
<envs />
<option name="isRedirectInput" value="false" />
<option name="redirectInputPath" value="" />
<method v="2">
<option name="CARGO.BUILD_TASK_PROVIDER" enabled="true" />
</method>
</configuration>
</component>

View File

@ -5,6 +5,7 @@ use std::thread;
use std::thread::JoinHandle;
use std::time::Duration;
#[derive(Debug, PartialEq)]
pub enum OpResult {
Ok,
TerminationRequested,
@ -24,24 +25,81 @@ pub trait Executable: Send {
fn periodic_op(&mut self, op_code: i32) -> Result<OpResult, Self::Error>;
}
pub fn executable_scheduler<
/// This function allows executing one task which implements the [Executable][Executable] trait
/// # Arguments
///
/// * `executable`: Executable task
/// * `task_freq`: Optional frequency of task. Required for periodic and fixed cycle tasks
/// * `op_code`: Operation code which is passed to the executable task [operation call][Executable::periodic_op]
/// * `termination`: Optional termination handler which can cancel threads with a broadcast
pub fn exec_sched_single<
T: Executable<Error = E> + Send + 'static + ?Sized,
E: Error + Send + 'static,
>(
mut executable: Box<T>,
task_freq: Option<Duration>,
op_code: i32,
mut termination: Option<BusReader<()>>,
) -> JoinHandle<Result<OpResult, E>> {
let mut cycle_count = 0;
thread::spawn(move || loop {
if let Some(ref mut terminator) = termination {
match terminator.try_recv() {
Ok(_) | Err(TryRecvError::Disconnected) => {
return Ok(OpResult::Ok);
}
Err(TryRecvError::Empty) => (),
}
}
match executable.exec_type() {
ExecutionType::OneShot => {
executable.periodic_op(op_code)?;
return Ok(OpResult::Ok);
}
ExecutionType::Infinite => {
executable.periodic_op(op_code)?;
}
ExecutionType::Cycles(cycles) => {
executable.periodic_op(op_code)?;
cycle_count += 1;
if cycle_count == cycles {
return Ok(OpResult::Ok);
}
}
}
let freq = task_freq.unwrap_or_else(|| panic!("No task frequency specified"));
thread::sleep(freq);
})
}
/// This function allows executing multiple tasks as long as the tasks implement the
/// [Executable][Executable] trait
/// # Arguments
///
/// * `executable_vec`: Vector of executable objects
/// * `task_freq`: Optional frequency of task. Required for periodic and fixed cycle tasks
/// * `op_code`: Operation code which is passed to the executable task periodic_op call
/// * `termination`: Optional termination handler which can cancel threads with a broadcast
pub fn exec_sched_multi<
T: Executable<Error = E> + Send + 'static + ?Sized,
E: Error + Send + 'static,
>(
mut executable_vec: Vec<Box<T>>,
task_freq: Option<Duration>,
op_code: i32,
mut termination: BusReader<()>,
mut termination: Option<BusReader<()>>,
) -> JoinHandle<Result<OpResult, E>> {
let mut cycle_counts = vec![0; executable_vec.len()];
let mut removal_flags = vec![false; executable_vec.len()];
thread::spawn(move || loop {
match termination.try_recv() {
if let Some(ref mut terminator) = termination {
match terminator.try_recv() {
Ok(_) | Err(TryRecvError::Disconnected) => {
removal_flags.iter_mut().for_each(|x| *x = true);
}
Err(TryRecvError::Empty) => (),
}
}
for (idx, executable) in executable_vec.iter_mut().enumerate() {
match executable.exec_type() {
ExecutionType::OneShot => {
@ -72,3 +130,226 @@ pub fn executable_scheduler<
thread::sleep(freq);
})
}
#[cfg(test)]
mod tests {
use super::{exec_sched_multi, exec_sched_single, Executable, ExecutionType, OpResult};
use std::error::Error;
use std::fmt;
use std::sync::{Arc, Mutex};
use std::time::Duration;
struct TestInfo {
exec_num: u32,
op_code: i32,
}
struct OneShotTask {
exec_num: Arc<Mutex<TestInfo>>,
}
struct FixedCyclesTask {
cycles: u32,
exec_num: Arc<Mutex<TestInfo>>,
}
struct PeriodicTask {
exec_num: Arc<Mutex<TestInfo>>,
}
#[derive(Debug, PartialEq)]
struct ExampleError {
details: String,
code: i32,
}
impl ExampleError {
fn new(msg: &str, code: i32) -> ExampleError {
ExampleError {
details: msg.to_string(),
code,
}
}
}
impl fmt::Display for ExampleError {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "{}", self.details)
}
}
impl Error for ExampleError {
fn description(&self) -> &str {
&self.details
}
}
const ONE_SHOT_TASK_NAME: &'static str = "One Shot Task";
impl Executable for OneShotTask {
type Error = ExampleError;
fn exec_type(&self) -> ExecutionType {
ExecutionType::OneShot
}
fn task_name(&self) -> &'static str {
return ONE_SHOT_TASK_NAME;
}
fn periodic_op(&mut self, op_code: i32) -> Result<OpResult, ExampleError> {
let mut data = self.exec_num.lock().expect("Locking Mutex failed");
data.exec_num += 1;
data.op_code = op_code;
std::mem::drop(data);
if op_code >= 0 {
Ok(OpResult::Ok)
} else {
Err(ExampleError::new("One Shot Task Failure", op_code))
}
}
}
const CYCLE_TASK_NAME: &'static str = "Fixed Cycles Task";
impl Executable for FixedCyclesTask {
type Error = ExampleError;
fn exec_type(&self) -> ExecutionType {
ExecutionType::Cycles(self.cycles)
}
fn task_name(&self) -> &'static str {
return CYCLE_TASK_NAME
}
fn periodic_op(&mut self, op_code: i32) -> Result<OpResult, ExampleError> {
let mut data = self.exec_num.lock().expect("Locking Mutex failed");
data.exec_num += 1;
data.op_code = op_code;
std::mem::drop(data);
if op_code >= 0 {
Ok(OpResult::Ok)
} else {
Err(ExampleError::new("Fixed Cycle Task Failure", op_code))
}
}
}
impl Executable for PeriodicTask {
type Error = ExampleError;
fn exec_type(&self) -> ExecutionType {
ExecutionType::Infinite
}
fn task_name(&self) -> &'static str {
"Periodic Task"
}
fn periodic_op(&mut self, op_code: i32) -> Result<OpResult, ExampleError> {
let mut data = self.exec_num.lock().expect("Locking Mutex failed");
data.exec_num += 1;
data.op_code = op_code;
std::mem::drop(data);
if op_code >= 0 {
Ok(OpResult::Ok)
} else {
Err(ExampleError::new("Example Task Failure", op_code))
}
}
}
#[test]
fn test_simple_one_shot() {
let expected_op_code = 42;
let shared = Arc::new(Mutex::new(TestInfo {
exec_num: 0,
op_code: 0,
}));
let exec_task = OneShotTask {
exec_num: shared.clone(),
};
let task = Box::new(exec_task);
let jhandle = exec_sched_single(
task,
Some(Duration::from_millis(100)),
expected_op_code,
None,
);
let thread_res = jhandle.join().expect("One Shot Task failed");
assert!(thread_res.is_ok());
assert_eq!(thread_res.unwrap(), OpResult::Ok);
let data = shared.lock().expect("Locking Mutex failed");
assert_eq!(data.exec_num, 1);
assert_eq!(data.op_code, expected_op_code);
}
#[test]
fn test_simple_multi_one_shot() {
let expected_op_code = 43;
let shared = Arc::new(Mutex::new(TestInfo {
exec_num: 0,
op_code: 0,
}));
let exec_task_0 = OneShotTask {
exec_num: shared.clone(),
};
let exec_task_1 = OneShotTask {
exec_num: shared.clone(),
};
let task_vec = vec![Box::new(exec_task_0), Box::new(exec_task_1)];
for task in task_vec.iter() {
assert_eq!(task.task_name(), ONE_SHOT_TASK_NAME);
}
let jhandle = exec_sched_multi(
task_vec,
Some(Duration::from_millis(100)),
expected_op_code,
None,
);
let thread_res = jhandle.join().expect("One Shot Task failed");
assert!(thread_res.is_ok());
assert_eq!(thread_res.unwrap(), OpResult::Ok);
let data = shared.lock().expect("Locking Mutex failed");
assert_eq!(data.exec_num, 2);
assert_eq!(data.op_code, expected_op_code);
}
#[test]
fn test_cycles() {
let expected_op_code = 44;
let shared = Arc::new(Mutex::new(TestInfo {
exec_num: 0,
op_code: 0,
}));
let cycled_task = Box::new(FixedCyclesTask {
exec_num: shared.clone(),
cycles: 1,
});
assert_eq!(cycled_task.task_name(), CYCLE_TASK_NAME);
let jh = exec_sched_single(cycled_task, Some(Duration::from_millis(100)), expected_op_code, None);
let thread_res = jh.join().expect("Cycles Task failed");
assert!(thread_res.is_ok());
let data = shared.lock().expect("Locking Mutex failed");
assert_eq!(thread_res.unwrap(), OpResult::Ok);
assert_eq!(data.exec_num, 1);
assert_eq!(data.op_code, expected_op_code);
}
#[test]
#[ignore]
fn test_periodic() {
let expected_op_code = 45;
let shared = Arc::new(Mutex::new(TestInfo {
exec_num: 0,
op_code: 0,
}));
let periodic_task = Box::new(PeriodicTask {
exec_num: shared.clone()
});
let jh = exec_sched_single(periodic_task, Some(Duration::from_millis(50)), expected_op_code, None);
let thread_res = jh.join().expect("Periodic Task failed");
assert!(thread_res.is_ok());
let data = shared.lock().expect("Locking Mutex failed");
assert_eq!(thread_res.unwrap(), OpResult::Ok);
assert_eq!(data.op_code, expected_op_code);
}
}

View File

@ -124,7 +124,7 @@ mod tests {
#[test]
fn test_obj_manager_simple() {
let mut obj_manager = ObjectManager::new();
let mut obj_manager = ObjectManager::default();
let expl_obj_id = ObjectId {
id: 0,
name: "Example 0",
@ -159,5 +159,17 @@ mod tests {
let expl_obj_back_casted = obj_back_casted.unwrap();
assert_eq!(expl_obj_back_casted.string, String::from("Hello Test"));
assert!(expl_obj_back_casted.was_initialized);
let existing_obj_id = ObjectId {
id: 12,
name: "Example 1",
};
let invalid_obj = OtherExampleObject {
id: existing_obj_id,
string: String::from("Hello Test"),
was_initialized: false,
};
assert_eq!(obj_manager.insert(Box::new(invalid_obj)), false);
}
}

View File

@ -1,162 +1,3 @@
use bus::{Bus};
use launchpad::core::executable::{executable_scheduler, Executable, ExecutionType, OpResult};
use std::error::Error;
use std::fmt;
use std::thread;
use std::time::Duration;
struct OneShotTask {}
struct FixedCyclesTask {}
struct PeriodicTask {}
#[derive(Debug)]
struct ExampleError {
details: String,
}
impl ExampleError {
fn new(msg: &str) -> ExampleError {
ExampleError {
details: msg.to_string(),
}
}
}
impl fmt::Display for ExampleError {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "{}", self.details)
}
}
impl Error for ExampleError {
fn description(&self) -> &str {
&self.details
}
}
impl Executable for OneShotTask {
type Error = ExampleError;
fn exec_type(&self) -> ExecutionType {
ExecutionType::OneShot
}
fn task_name(&self) -> &'static str {
"One Shot Task"
}
fn periodic_op(&mut self, op_code: i32) -> Result<OpResult, ExampleError> {
if op_code >= 0 {
println!("One-shot operation with operation code {op_code} OK!");
Ok(OpResult::Ok)
} else {
println!("One-shot operation failure by passing op code {op_code}!");
Err(ExampleError::new("Example Task Failure"))
}
}
}
impl Executable for FixedCyclesTask {
type Error = ExampleError;
fn exec_type(&self) -> ExecutionType {
ExecutionType::Cycles(3)
}
fn task_name(&self) -> &'static str {
"Fixed Cycles Task"
}
fn periodic_op(&mut self, op_code: i32) -> Result<OpResult, ExampleError> {
if op_code >= 0 {
println!("Fixed-cycle operation with operation code {op_code} OK!");
Ok(OpResult::Ok)
} else {
println!("Fixed-cycle operation failure by passing op code {op_code}!");
Err(ExampleError::new("Example Task Failure"))
}
}
}
impl Executable for PeriodicTask {
type Error = ExampleError;
fn exec_type(&self) -> ExecutionType {
ExecutionType::Infinite
}
fn task_name(&self) -> &'static str {
"Periodic Task"
}
fn periodic_op(&mut self, op_code: i32) -> Result<OpResult, ExampleError> {
if op_code >= 0 {
println!("Periodic operation with operation code {op_code} OK!");
Ok(OpResult::Ok)
} else {
println!("Periodic operation failure by passing op code {op_code}!");
Err(ExampleError::new("Example Task Failure"))
}
}
}
fn test0(term_bus: &mut Bus<()>) {
let exec_task = OneShotTask {};
let task_vec = vec![Box::new(exec_task)];
let jhandle = executable_scheduler(
task_vec,
Some(Duration::from_millis(100)),
0,
term_bus.add_rx(),
);
let exec_task2 = FixedCyclesTask {};
let task_vec2: Vec<Box<dyn Executable<Error = ExampleError> + Send>> =
vec![Box::new(exec_task2)];
let jhandle2 = executable_scheduler(
task_vec2,
Some(Duration::from_millis(100)),
1,
term_bus.add_rx(),
);
jhandle
.join()
.expect("Joining thread failed")
.expect("Task failed");
jhandle2
.join()
.expect("Joining thread 2 failed")
.expect("Task 2 failed");
}
fn test1(term_bus: &mut Bus<()>) {
let one_shot_in_vec = OneShotTask {};
let cycles_in_vec = FixedCyclesTask {};
let periodic_in_vec = PeriodicTask {};
let test_vec: Vec<Box<dyn Executable<Error = ExampleError>>> = vec![
Box::new(one_shot_in_vec),
Box::new(cycles_in_vec),
Box::new(periodic_in_vec),
];
let jhandle3 = executable_scheduler(
test_vec,
Some(Duration::from_millis(100)),
3,
term_bus.add_rx(),
);
thread::sleep(Duration::from_millis(5000));
println!("Broadcasting cancel");
term_bus.broadcast(());
jhandle3
.join()
.expect("Joining thread 3 failed")
.expect("Task 3 failed");
}
fn main() {
let mut tx = Bus::new(5);
test0(&mut tx);
test1(&mut tx);
println!("hello");
}