rust/sat-rs
rust/sat-rs
5
1
Fork 0
Code Issues 9 Pull Requests Projects Releases 22 Wiki Activity

Implementation of scheduler in pus and tmtc handler #29

Merged
muellerr merged 13 commits from pus_schedule_implementation into main 2023-02-01 13:40:49 +01:00
Conversation 1 Commits 13 Files Changed 5 +73 -67
3 changed files with 73 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit a7aad003ab - Show all commits

28
satrs-core/src/pus/scheduling.rs
View File

@ -709,7 +709,6 @@ mod tests {
} }
#[test] #[test]
#[should_panic]
fn insert_wrong_service() { fn insert_wrong_service() {
let mut scheduler = let mut scheduler =
PusScheduler::new(UnixTimestamp::new_only_seconds(0), Duration::from_secs(5)); PusScheduler::new(UnixTimestamp::new_only_seconds(0), Duration::from_secs(5));
@ -719,13 +718,19 @@ mod tests {
let mut buf: [u8; 32] = [0; 32]; let mut buf: [u8; 32] = [0; 32];
let tc = wrong_tc_service(UnixTimestamp::new_only_seconds(100), &mut buf); let tc = wrong_tc_service(UnixTimestamp::new_only_seconds(100), &mut buf);
let addr = scheduler match scheduler.insert_wrapped_tc::<spacepackets::time::cds::TimeProvider>(&tc, &mut pool) {
.insert_wrapped_tc::<spacepackets::time::cds::TimeProvider>(&tc, &mut pool) Ok(_) => {
.unwrap(); panic!();
}
Err(e) => {
if e != ScheduleError::WrongService {
panic!();
}
}
}
} }
#[test] #[test]
#[should_panic]
fn insert_wrong_subservice() { fn insert_wrong_subservice() {
let mut scheduler = let mut scheduler =
PusScheduler::new(UnixTimestamp::new_only_seconds(0), Duration::from_secs(5)); PusScheduler::new(UnixTimestamp::new_only_seconds(0), Duration::from_secs(5));
@ -735,8 +740,15 @@ mod tests {
let mut buf: [u8; 32] = [0; 32]; let mut buf: [u8; 32] = [0; 32];
let tc = wrong_tc_subservice(UnixTimestamp::new_only_seconds(100), &mut buf); let tc = wrong_tc_subservice(UnixTimestamp::new_only_seconds(100), &mut buf);
let addr = scheduler match scheduler.insert_wrapped_tc::<spacepackets::time::cds::TimeProvider>(&tc, &mut pool) {
.insert_wrapped_tc::<spacepackets::time::cds::TimeProvider>(&tc, &mut pool) Ok(_) => {
.unwrap(); panic!();
}
Err(e) => {
if e != ScheduleError::WrongSubservice {
panic!();
}
}
}
} }
} }

67
satrs-example/src/pus.rs
View File

@ -19,6 +19,7 @@ use satrs_core::{
spacepackets::time::cds::TimeProvider, spacepackets::time::TimeWriter, spacepackets::SpHeader, spacepackets::time::cds::TimeProvider, spacepackets::time::TimeWriter, spacepackets::SpHeader,
}; };
use satrs_example::{hk_err, tmtc_err}; use satrs_example::{hk_err, tmtc_err};
use std::cell::RefCell;
use std::collections::HashMap; use std::collections::HashMap;
use std::rc::Rc; use std::rc::Rc;
use std::sync::mpsc::Sender; use std::sync::mpsc::Sender;
@ -36,7 +37,7 @@ pub struct PusReceiver {
request_map: HashMap<u32, Sender<RequestWithToken>>, request_map: HashMap<u32, Sender<RequestWithToken>>,
stamper: TimeProvider, stamper: TimeProvider,
time_stamp: [u8; 7], time_stamp: [u8; 7],
scheduler: Arc<Mutex<PusScheduler>>, scheduler: Rc<RefCell<PusScheduler>>,
} }
impl PusReceiver { impl PusReceiver {
@ -48,7 +49,7 @@ impl PusReceiver {
tc_source: PusTcSource, tc_source: PusTcSource,
event_request_tx: Sender<EventRequestWithToken>, event_request_tx: Sender<EventRequestWithToken>,
request_map: HashMap<u32, Sender<RequestWithToken>>, request_map: HashMap<u32, Sender<RequestWithToken>>,
scheduler: Arc<Mutex<PusScheduler>>, scheduler: Rc<RefCell<PusScheduler>>,
) -> Self { ) -> Self {
Self { Self {
tm_helper: PusTmWithCdsShortHelper::new(apid), tm_helper: PusTmWithCdsShortHelper::new(apid),
@ -286,28 +287,16 @@ impl PusReceiver {
} }
fn handle_scheduled_tc(&mut self, pus_tc: &PusTc, token: VerificationToken<TcStateAccepted>) { fn handle_scheduled_tc(&mut self, pus_tc: &PusTc, token: VerificationToken<TcStateAccepted>) {
if pus_tc.user_data().is_none() {
self.update_time_stamp();
self.verif_reporter
.start_failure(
token,
FailParams::new(Some(&self.time_stamp), &tmtc_err::NOT_ENOUGH_APP_DATA, None),
)
.expect("Sending start failure TM failed");
return;
}
self.update_time_stamp(); self.update_time_stamp();
match pus_tc.subservice() { match pus_tc.subservice() {
1 => { 1 => {
let mut scheduler = self.scheduler.lock().expect("Lock of scheduler failed");
let start_token = self let start_token = self
.verif_reporter .verif_reporter
.start_success(token, Some(&self.time_stamp)) .start_success(token, Some(&self.time_stamp))
.expect("Error sending start success"); .expect("Error sending start success");
let mut scheduler = self.scheduler.borrow_mut();
scheduler.enable(); scheduler.enable();
if scheduler.is_enabled() { if scheduler.is_enabled() {
self.verif_reporter self.verif_reporter
@ -318,11 +307,12 @@ impl PusReceiver {
} }
} }
2 => { 2 => {
let mut scheduler = self.scheduler.lock().expect("Lock of scheduler failed");
let start_token = self let start_token = self
.verif_reporter .verif_reporter
.start_success(token, Some(&self.time_stamp)) .start_success(token, Some(&self.time_stamp))
.expect("Error sending start success"); .expect("Error sending start success");
let mut scheduler = self.scheduler.borrow_mut();
scheduler.disable(); scheduler.disable();
if !scheduler.is_enabled() { if !scheduler.is_enabled() {
self.verif_reporter self.verif_reporter
@ -333,45 +323,50 @@ impl PusReceiver {
} }
} }
3 => { 3 => {
let mut scheduler = self.scheduler.lock().expect("Lock of scheduler failed");
let start_token = self let start_token = self
.verif_reporter .verif_reporter
.start_success(token, Some(&self.time_stamp)) .start_success(token, Some(&self.time_stamp))
.expect("Error sending start success"); .expect("Error sending start success");
match self.tc_source.tc_store.pool.write() {
Ok(mut pool) => { let mut scheduler = self.scheduler.borrow_mut();
match scheduler.reset(pool.as_mut()) { let mut pool = self
Ok(_) => { .tc_source
.tc_store
.pool
.write()
.expect("Locking pool failed");
scheduler
.reset(pool.as_mut())
.expect("Error resetting TC Pool");
self.verif_reporter self.verif_reporter
.completion_success(start_token, Some(&self.time_stamp)) .completion_success(start_token, Some(&self.time_stamp))
.expect("Error sending completion success"); .expect("Error sending completion success");
} }
Err(_) => {
// TODO
}
}
}
Err(_) => {}
}
}
4 => { 4 => {
let mut scheduler = self.scheduler.lock().expect("Lock of scheduler failed");
let start_token = self let start_token = self
.verif_reporter .verif_reporter
.start_success(token, Some(&self.time_stamp)) .start_success(token, Some(&self.time_stamp))
.expect("Error sending start success"); .expect("Error sending start success");
match self.tc_source.tc_store.pool.write() {
Ok(mut pool) => { let mut scheduler = self.scheduler.borrow_mut();
let mut pool = self
.tc_source
.tc_store
.pool
.write()
.expect("Locking pool failed");
scheduler scheduler
.insert_wrapped_tc::<spacepackets::time::cds::TimeProvider>( .insert_wrapped_tc::<spacepackets::time::cds::TimeProvider>(
pus_tc, pus_tc,
pool.as_mut(), pool.as_mut(),
) )
.expect("TODO: panic message"); .expect("TODO: panic message");
}
Err(_) => {} self.verif_reporter
} .completion_success(start_token, Some(&self.time_stamp))
.expect("Error sending completion success");
//let addr = self.tc_source.tc_store.add_pus_tc().unwrap(); //let addr = self.tc_source.tc_store.add_pus_tc().unwrap();
//let unix_time = UnixTimestamp::new_only_seconds(self.stamper.unix_seconds()); //let unix_time = UnixTimestamp::new_only_seconds(self.stamper.unix_seconds());

27
satrs-example/src/tmtc.rs
View File

@ -1,6 +1,7 @@
use satrs_core::events::EventU32; use satrs_core::events::EventU32;
use satrs_core::hal::host::udp_server::{ReceiveResult, UdpTcServer}; use satrs_core::hal::host::udp_server::{ReceiveResult, UdpTcServer};
use satrs_core::params::Params; use satrs_core::params::Params;
use std::cell::RefCell;
use std::collections::HashMap; use std::collections::HashMap;
use std::error::Error; use std::error::Error;
use std::fmt::{Display, Formatter}; use std::fmt::{Display, Formatter};
@ -161,7 +162,7 @@ impl ReceivesCcsdsTc for PusTcSource {
} }
} }
pub fn core_tmtc_task(args: OtherArgs, mut tc_args: TcArgs, tm_args: TmArgs) { pub fn core_tmtc_task(args: OtherArgs, mut tc_args: TcArgs, tm_args: TmArgs) {
let mut scheduler = Arc::new(Mutex::new( let mut scheduler = Rc::new(RefCell::new(
PusScheduler::new_with_current_init_time(Duration::from_secs(5)).unwrap(), PusScheduler::new_with_current_init_time(Duration::from_secs(5)).unwrap(),
)); ));
@ -192,7 +193,6 @@ pub fn core_tmtc_task(args: OtherArgs, mut tc_args: TcArgs, tm_args: TmArgs) {
tm_store: tm_args.tm_store.pool.clone(), tm_store: tm_args.tm_store.pool.clone(),
}; };
let mut test_closure = |boolvar: bool, store_addr: &StoreAddr| true;
let (mut tc_source, mut tc_receiver) = tc_args.split(); let (mut tc_source, mut tc_receiver) = tc_args.split();
loop { loop {
@ -213,21 +213,20 @@ fn core_tmtc_loop(
tc_source: &mut PusTcSource, tc_source: &mut PusTcSource,
tc_receiver: &mut Receiver<StoreAddr>, tc_receiver: &mut Receiver<StoreAddr>,
pus_receiver: &mut PusReceiver, pus_receiver: &mut PusReceiver,
scheduler: Arc<Mutex<PusScheduler>>, scheduler: Rc<RefCell<PusScheduler>>,
) { ) {
let releaser = |enabled: bool, addr: &StoreAddr| { let releaser = |enabled: bool, addr: &StoreAddr| -> bool {
tc_source.tc_source.send(*addr); match tc_source.tc_source.send(*addr) {
true Ok(_) => true,
Err(_) => false,
}
}; };
let mut scheduler = scheduler.lock().expect("Lock of scheduler failed"); let mut scheduler = scheduler.borrow_mut();
match tc_source.tc_store.pool.write() { let mut pool = tc_source.tc_store.pool.write().expect("error locking pool");
Ok(mut pool) => match scheduler.release_telecommands(releaser, pool.as_mut()) { scheduler
Ok(_) => {} .release_telecommands(releaser, pool.as_mut())
Err(_) => {} .expect("error releasing tc");
},
Err(_) => {}
}
while poll_tc_server(udp_tmtc_server) {} while poll_tc_server(udp_tmtc_server) {}
match tc_receiver.try_recv() { match tc_receiver.try_recv() {