OBSW-Client Example #11

Merged
muellerr merged 44 commits from obsw-client-example into main 2022-09-11 16:33:17 +02:00
14 changed files with 498 additions and 275 deletions
Showing only changes of commit 9b791b1de1 - Show all commits

104
Cargo.lock generated
View File

@ -2,6 +2,17 @@
# It is not intended for manual editing. # It is not intended for manual editing.
version = 3 version = 3
[[package]]
name = "ahash"
version = "0.7.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fcb51a0695d8f838b1ee009b3fbf66bda078cd64590202a864a8f3e8c4315c47"
dependencies = [
"getrandom",
"once_cell",
"version_check",
]
[[package]] [[package]]
name = "aho-corasick" name = "aho-corasick"
version = "0.7.18" version = "0.7.18"
@ -221,8 +232,10 @@ name = "fsrc-core"
version = "0.1.0" version = "0.1.0"
dependencies = [ dependencies = [
"bus", "bus",
"delegate",
"downcast-rs", "downcast-rs",
"num", "hashbrown",
"num-traits",
"once_cell", "once_cell",
"postcard", "postcard",
"serde", "serde",
@ -239,6 +252,17 @@ dependencies = [
"spacepackets", "spacepackets",
] ]
[[package]]
name = "getrandom"
version = "0.2.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4eb1a864a501629691edf6c15a593b7a51eebaa1e8468e9ddc623de7c9b58ec6"
dependencies = [
"cfg-if 1.0.0",
"libc",
"wasi",
]
[[package]] [[package]]
name = "hash32" name = "hash32"
version = "0.2.1" version = "0.2.1"
@ -248,6 +272,15 @@ dependencies = [
"byteorder", "byteorder",
] ]
[[package]]
name = "hashbrown"
version = "0.12.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888"
dependencies = [
"ahash",
]
[[package]] [[package]]
name = "heapless" name = "heapless"
version = "0.7.14" version = "0.7.14"
@ -338,40 +371,6 @@ version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "546c37ac5d9e56f55e73b677106873d9d9f5190605e41a856503623648488cae" checksum = "546c37ac5d9e56f55e73b677106873d9d9f5190605e41a856503623648488cae"
[[package]]
name = "num"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "43db66d1170d347f9a065114077f7dccb00c1b9478c89384490a3425279a4606"
dependencies = [
"num-bigint",
"num-complex",
"num-integer",
"num-iter",
"num-rational",
"num-traits",
]
[[package]]
name = "num-bigint"
version = "0.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f93ab6289c7b344a8a9f60f88d80aa20032336fe78da341afc91c8a2341fc75f"
dependencies = [
"autocfg",
"num-integer",
"num-traits",
]
[[package]]
name = "num-complex"
version = "0.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "97fbc387afefefd5e9e39493299f3069e14a140dd34dc19b4c1c1a8fddb6a790"
dependencies = [
"num-traits",
]
[[package]] [[package]]
name = "num-integer" name = "num-integer"
version = "0.1.45" version = "0.1.45"
@ -382,29 +381,6 @@ dependencies = [
"num-traits", "num-traits",
] ]
[[package]]
name = "num-iter"
version = "0.1.43"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7d03e6c028c5dc5cac6e2dec0efda81fc887605bb3d884578bb6d6bf7514e252"
dependencies = [
"autocfg",
"num-integer",
"num-traits",
]
[[package]]
name = "num-rational"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d41702bd167c2df5520b384281bc111a4b5efcf7fbc4c9c222c815b07e0a6a6a"
dependencies = [
"autocfg",
"num-bigint",
"num-integer",
"num-traits",
]
[[package]] [[package]]
name = "num-traits" name = "num-traits"
version = "0.2.15" version = "0.2.15"
@ -677,6 +653,12 @@ version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "77439c1b53d2303b20d9459b1ade71a83c716e3f9c34f3228c00e6f185d6c002" checksum = "77439c1b53d2303b20d9459b1ade71a83c716e3f9c34f3228c00e6f185d6c002"
[[package]]
name = "version_check"
version = "0.9.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "49874b5167b65d7193b8aba1567f5c7d93d001cafc34600cee003eda787e483f"
[[package]] [[package]]
name = "void" name = "void"
version = "1.0.2" version = "1.0.2"
@ -692,6 +674,12 @@ dependencies = [
"vcell", "vcell",
] ]
[[package]]
name = "wasi"
version = "0.11.0+wasi-snapshot-preview1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423"
[[package]] [[package]]
name = "wasm-bindgen" name = "wasm-bindgen"
version = "0.2.82" version = "0.2.82"

View File

@ -7,16 +7,24 @@ edition = "2021"
[dependencies] [dependencies]
thiserror = "1.0" thiserror = "1.0"
bus = "2.2.3" delegate = "0.7.0"
num = "0.4" hashbrown = "0.12.3"
[dependencies.spacepackets] [dependencies.num-traits]
path = "../spacepackets" version = "0.2"
default-features = false
[dependencies.downcast-rs] [dependencies.downcast-rs]
version = "1.2.0" version = "1.2.0"
default-features = false default-features = false
[dependencies.bus]
version = "2.2.3"
optional = true
[dependencies.spacepackets]
path = "../spacepackets"
[dev-dependencies] [dev-dependencies]
postcard = { version = "1.0.1", features = ["use-std"] } postcard = { version = "1.0.1", features = ["use-std"] }
serde = "1.0.143" serde = "1.0.143"
@ -25,4 +33,6 @@ once_cell = "1.13.1"
[features] [features]
default = ["std"] default = ["std"]
std = ["downcast-rs/std"] std = ["downcast-rs/std", "alloc", "bus"]
alloc = []

View File

@ -1,6 +1,9 @@
//! [Event][crate::events::Event] management and forwarding //! [Event][crate::events::Event] management and forwarding
use crate::events::{Event, EventRaw, GroupId}; use crate::events::{Event, EventRaw, GroupId};
use std::collections::HashMap; use alloc::boxed::Box;
use alloc::vec;
use alloc::vec::Vec;
use hashbrown::HashMap;
#[derive(PartialEq, Eq, Hash, Copy, Clone)] #[derive(PartialEq, Eq, Hash, Copy, Clone)]
enum ListenerType { enum ListenerType {
@ -62,11 +65,14 @@ impl<E> EventManager<E> {
key: ListenerType, key: ListenerType,
dest: impl EventListener<Error = E> + 'static, dest: impl EventListener<Error = E> + 'static,
) { ) {
if let std::collections::hash_map::Entry::Vacant(e) = self.listeners.entry(key) { if !self.listeners.contains_key(&key) {
e.insert(vec![Listener { self.listeners.insert(
ltype: key, key,
dest: Box::new(dest), vec![Listener {
}]); ltype: key,
dest: Box::new(dest),
}],
);
} else { } else {
let vec = self.listeners.get_mut(&key).unwrap(); let vec = self.listeners.get_mut(&key).unwrap();
// To prevent double insertions // To prevent double insertions
@ -117,6 +123,7 @@ mod tests {
use super::{EventListener, HandlerResult, ReceivesAllEvent}; use super::{EventListener, HandlerResult, ReceivesAllEvent};
use crate::event_man::EventManager; use crate::event_man::EventManager;
use crate::events::{Event, Severity}; use crate::events::{Event, Severity};
use alloc::boxed::Box;
use std::sync::mpsc::{channel, Receiver, SendError, Sender}; use std::sync::mpsc::{channel, Receiver, SendError, Sender};
use std::thread; use std::thread;
use std::time::Duration; use std::time::Duration;

View File

@ -1,5 +1,4 @@
//! Event support module //! Event support module
use num::pow;
pub type GroupId = u16; pub type GroupId = u16;
pub type UniqueId = u16; pub type UniqueId = u16;
@ -47,7 +46,7 @@ impl Event {
/// * `unique_id`: Each event has a unique 16 bit ID occupying the last 16 bits of the /// * `unique_id`: Each event has a unique 16 bit ID occupying the last 16 bits of the
/// raw event ID /// raw event ID
pub fn new(severity: Severity, group_id: GroupId, unique_id: UniqueId) -> Option<Event> { pub fn new(severity: Severity, group_id: GroupId, unique_id: UniqueId) -> Option<Event> {
if group_id > (pow::pow(2u8 as u16, 13) - 1) { if group_id > (2u16.pow(13) - 1) {
return None; return None;
} }
Some(Event { Some(Event {

View File

@ -1,10 +1,13 @@
//! Task scheduling module //! Task scheduling module
use bus::BusReader; use bus::BusReader;
use std::boxed::Box;
use std::error::Error; use std::error::Error;
use std::sync::mpsc::TryRecvError; use std::sync::mpsc::TryRecvError;
use std::thread; use std::thread;
use std::thread::JoinHandle; use std::thread::JoinHandle;
use std::time::Duration; use std::time::Duration;
use std::vec;
use std::vec::Vec;
#[derive(Debug, PartialEq, Eq)] #[derive(Debug, PartialEq, Eq)]
pub enum OpResult { pub enum OpResult {
@ -138,10 +141,13 @@ pub fn exec_sched_multi<
mod tests { mod tests {
use super::{exec_sched_multi, exec_sched_single, Executable, ExecutionType, OpResult}; use super::{exec_sched_multi, exec_sched_single, Executable, ExecutionType, OpResult};
use bus::Bus; use bus::Bus;
use std::boxed::Box;
use std::error::Error; use std::error::Error;
use std::string::{String, ToString};
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
use std::time::Duration; use std::time::Duration;
use std::{fmt, thread}; use std::vec::Vec;
use std::{fmt, thread, vec};
struct TestInfo { struct TestInfo {
exec_num: u32, exec_num: u32,

View File

@ -39,7 +39,6 @@ impl<E> UdpTcServer<E> {
} }
pub fn try_recv_tc(&mut self) -> Result<(usize, SocketAddr), ReceiveResult<E>> { pub fn try_recv_tc(&mut self) -> Result<(usize, SocketAddr), ReceiveResult<E>> {
// .map_err(|e| IoError(e))?;
let res = match self.socket.recv_from(&mut self.recv_buf) { let res = match self.socket.recv_from(&mut self.recv_buf) {
Ok(res) => res, Ok(res) => res,
Err(e) => { Err(e) => {
@ -57,4 +56,8 @@ impl<E> UdpTcServer<E> {
.map_err(|e| ReceiveResult::ReceiverError(e))?; .map_err(|e| ReceiveResult::ReceiverError(e))?;
Ok(res) Ok(res)
} }
pub fn last_sender(&self) -> Option<SocketAddr> {
self.sender_addr
}
} }

View File

@ -7,12 +7,21 @@
//! The crates can generally be used in a `no_std` environment, but some crates expect that the //! The crates can generally be used in a `no_std` environment, but some crates expect that the
//! [alloc](https://doc.rust-lang.org/alloc) crate is available to allow boxed trait objects. //! [alloc](https://doc.rust-lang.org/alloc) crate is available to allow boxed trait objects.
//! These are used to supply user code to the crates. //! These are used to supply user code to the crates.
#![no_std]
#[cfg(feature = "alloc")]
extern crate alloc;
#[cfg(any(feature = "std", test))]
extern crate std;
pub mod error; pub mod error;
#[cfg(feature = "alloc")]
pub mod event_man; pub mod event_man;
pub mod events; pub mod events;
#[cfg(feature = "std")]
pub mod executable; pub mod executable;
pub mod hal; pub mod hal;
pub mod objects; pub mod objects;
#[cfg(feature = "alloc")]
pub mod pool; pub mod pool;
pub mod tmtc; pub mod tmtc;

View File

@ -29,12 +29,12 @@
//! } //! }
//! //!
//! impl SystemObject for ExampleSysObj { //! impl SystemObject for ExampleSysObj {
//! //! type Error = ();
//! fn get_object_id(&self) -> &ObjectId { //! fn get_object_id(&self) -> &ObjectId {
//! &self.id //! &self.id
//! } //! }
//! //!
//! fn initialize(&mut self) -> Result<(), Box<dyn Error>> { //! fn initialize(&mut self) -> Result<(), Self::Error> {
//! self.was_initialized = true; //! self.was_initialized = true;
//! Ok(()) //! Ok(())
//! } //! }
@ -51,9 +51,12 @@
//! assert_eq!(example_obj.id, obj_id); //! assert_eq!(example_obj.id, obj_id);
//! assert_eq!(example_obj.dummy, 42); //! assert_eq!(example_obj.dummy, 42);
//! ``` //! ```
use alloc::boxed::Box;
#[cfg(not(feature = "std"))]
use alloc::vec::Vec;
use downcast_rs::Downcast; use downcast_rs::Downcast;
use std::collections::HashMap; use hashbrown::HashMap;
#[cfg(feature = "std")]
use std::error::Error; use std::error::Error;
#[derive(PartialEq, Eq, Hash, Copy, Clone, Debug)] #[derive(PartialEq, Eq, Hash, Copy, Clone, Debug)]
@ -65,33 +68,37 @@ pub struct ObjectId {
/// Each object which is stored inside the [object manager][ObjectManager] needs to implemented /// Each object which is stored inside the [object manager][ObjectManager] needs to implemented
/// this trait /// this trait
pub trait SystemObject: Downcast { pub trait SystemObject: Downcast {
type Error;
fn get_object_id(&self) -> &ObjectId; fn get_object_id(&self) -> &ObjectId;
fn initialize(&mut self) -> Result<(), Box<dyn Error>>; fn initialize(&mut self) -> Result<(), Self::Error>;
} }
downcast_rs::impl_downcast!(SystemObject); downcast_rs::impl_downcast!(SystemObject assoc Error);
pub trait ManagedSystemObject: SystemObject + Send {} pub trait ManagedSystemObject: SystemObject + Send {}
downcast_rs::impl_downcast!(ManagedSystemObject); downcast_rs::impl_downcast!(ManagedSystemObject assoc Error);
/// Helper module to manage multiple [ManagedSystemObjects][ManagedSystemObject] by mapping them /// Helper module to manage multiple [ManagedSystemObjects][ManagedSystemObject] by mapping them
/// using an [object ID][ObjectId] /// using an [object ID][ObjectId]
pub struct ObjectManager { #[cfg(feature = "alloc")]
obj_map: HashMap<ObjectId, Box<dyn ManagedSystemObject>>, pub struct ObjectManager<E> {
obj_map: HashMap<ObjectId, Box<dyn ManagedSystemObject<Error = E>>>,
} }
impl Default for ObjectManager { #[cfg(feature = "alloc")]
impl<E: 'static> Default for ObjectManager<E> {
fn default() -> Self { fn default() -> Self {
Self::new() Self::new()
} }
} }
impl ObjectManager { #[cfg(feature = "alloc")]
pub fn new() -> ObjectManager { impl<E: 'static> ObjectManager<E> {
pub fn new() -> Self {
ObjectManager { ObjectManager {
obj_map: HashMap::new(), obj_map: HashMap::new(),
} }
} }
pub fn insert(&mut self, sys_obj: Box<dyn ManagedSystemObject>) -> bool { pub fn insert(&mut self, sys_obj: Box<dyn ManagedSystemObject<Error = E>>) -> bool {
let obj_id = sys_obj.get_object_id(); let obj_id = sys_obj.get_object_id();
if self.obj_map.contains_key(obj_id) { if self.obj_map.contains_key(obj_id) {
return false; return false;
@ -114,14 +121,14 @@ impl ObjectManager {
/// Retrieve a reference to an object stored inside the manager. The type to retrieve needs to /// Retrieve a reference to an object stored inside the manager. The type to retrieve needs to
/// be explicitly passed as a generic parameter or specified on the left hand side of the /// be explicitly passed as a generic parameter or specified on the left hand side of the
/// expression. /// expression.
pub fn get_ref<T: ManagedSystemObject>(&self, key: &ObjectId) -> Option<&T> { pub fn get_ref<T: ManagedSystemObject<Error = E>>(&self, key: &ObjectId) -> Option<&T> {
self.obj_map.get(key).and_then(|o| o.downcast_ref::<T>()) self.obj_map.get(key).and_then(|o| o.downcast_ref::<T>())
} }
/// Retrieve a mutable reference to an object stored inside the manager. The type to retrieve /// Retrieve a mutable reference to an object stored inside the manager. The type to retrieve
/// needs to be explicitly passed as a generic parameter or specified on the left hand side /// needs to be explicitly passed as a generic parameter or specified on the left hand side
/// of the expression. /// of the expression.
pub fn get_mut<T: ManagedSystemObject>(&mut self, key: &ObjectId) -> Option<&mut T> { pub fn get_mut<T: ManagedSystemObject<Error = E>>(&mut self, key: &ObjectId) -> Option<&mut T> {
self.obj_map self.obj_map
.get_mut(key) .get_mut(key)
.and_then(|o| o.downcast_mut::<T>()) .and_then(|o| o.downcast_mut::<T>())
@ -131,7 +138,8 @@ impl ObjectManager {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use crate::objects::{ManagedSystemObject, ObjectId, ObjectManager, SystemObject}; use crate::objects::{ManagedSystemObject, ObjectId, ObjectManager, SystemObject};
use std::error::Error; use std::boxed::Box;
use std::string::String;
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
use std::thread; use std::thread;
@ -152,11 +160,12 @@ mod tests {
} }
impl SystemObject for ExampleSysObj { impl SystemObject for ExampleSysObj {
type Error = ();
fn get_object_id(&self) -> &ObjectId { fn get_object_id(&self) -> &ObjectId {
&self.id &self.id
} }
fn initialize(&mut self) -> Result<(), Box<dyn Error>> { fn initialize(&mut self) -> Result<(), Self::Error> {
self.was_initialized = true; self.was_initialized = true;
Ok(()) Ok(())
} }
@ -171,11 +180,12 @@ mod tests {
} }
impl SystemObject for OtherExampleObject { impl SystemObject for OtherExampleObject {
type Error = ();
fn get_object_id(&self) -> &ObjectId { fn get_object_id(&self) -> &ObjectId {
&self.id &self.id
} }
fn initialize(&mut self) -> Result<(), Box<dyn Error>> { fn initialize(&mut self) -> Result<(), Self::Error> {
self.was_initialized = true; self.was_initialized = true;
Ok(()) Ok(())
} }

View File

@ -73,6 +73,12 @@
//! assert_eq!(buf_read_back[0], 7); //! assert_eq!(buf_read_back[0], 7);
//! } //! }
//! ``` //! ```
use alloc::format;
use alloc::string::String;
use alloc::vec;
use alloc::vec::Vec;
use delegate::delegate;
type NumBlocks = u16; type NumBlocks = u16;
/// Configuration structure of the [local pool][LocalPool] /// Configuration structure of the [local pool][LocalPool]
@ -172,12 +178,12 @@ impl LocalPool {
/// size and then copy the given data to the block. Yields a [StoreAddr] which can be used /// size and then copy the given data to the block. Yields a [StoreAddr] which can be used
/// to access the data stored in the pool /// to access the data stored in the pool
pub fn add(&mut self, data: &[u8]) -> Result<StoreAddr, StoreError> { pub fn add(&mut self, data: &[u8]) -> Result<StoreAddr, StoreError> {
let data_len = data.as_ref().len(); let data_len = data.len();
if data_len > Self::MAX_SIZE { if data_len > Self::MAX_SIZE {
return Err(StoreError::DataTooLarge(data_len)); return Err(StoreError::DataTooLarge(data_len));
} }
let addr = self.reserve(data_len)?; let addr = self.reserve(data_len)?;
self.write(&addr, data.as_ref())?; self.write(&addr, data)?;
Ok(addr) Ok(addr)
} }
@ -202,27 +208,74 @@ impl LocalPool {
Ok(block) Ok(block)
} }
/// This function behaves like [Self::modify], but consumes the provided address and returns a
/// RAII conformant guard object.
///
/// Unless the guard [PoolRwGuard::release] method is called, the data for the
/// given address will be deleted automatically when the guard is dropped.
/// This can prevent memory leaks. Users can read (and modify) the data and release the guard
/// if the data in the store is valid for further processing. If the data is faulty, no
/// manual deletion is necessary when returning from a processing function prematurely.
pub fn modify_with_guard(&mut self, addr: StoreAddr) -> PoolRwGuard {
PoolRwGuard::new(self, addr)
}
/// Read data by yielding a read-only reference given a [StoreAddr] /// Read data by yielding a read-only reference given a [StoreAddr]
pub fn read(&self, addr: &StoreAddr) -> Result<&[u8], StoreError> { pub fn read(&self, addr: &StoreAddr) -> Result<&[u8], StoreError> {
let curr_size = self.addr_check(addr)?; let curr_size = self.addr_check(addr)?;
let raw_pos = self.raw_pos(addr).unwrap(); let raw_pos = self.raw_pos(addr).unwrap();
let block = &self.pool.get(addr.pool_idx as usize).unwrap()[raw_pos..curr_size]; let block = &self.pool.get(addr.pool_idx as usize).unwrap()[raw_pos..raw_pos + curr_size];
Ok(block) Ok(block)
} }
/// This function behaves like [Self::read], but consumes the provided address and returns a
/// RAII conformant guard object.
///
/// Unless the guard [PoolRwGuard::release] method is called, the data for the
/// given address will be deleted automatically when the guard is dropped.
/// This can prevent memory leaks. Users can read the data and release the guard
/// if the data in the store is valid for further processing. If the data is faulty, no
/// manual deletion is necessary when returning from a processing function prematurely.
pub fn read_with_guard(&mut self, addr: StoreAddr) -> PoolGuard {
PoolGuard::new(self, addr)
}
/// Delete data inside the pool given a [StoreAddr] /// Delete data inside the pool given a [StoreAddr]
pub fn delete(&mut self, addr: StoreAddr) -> Result<(), StoreError> { pub fn delete(&mut self, addr: StoreAddr) -> Result<(), StoreError> {
self.addr_check(&addr)?; self.addr_check(&addr)?;
let block_size = self.pool_cfg.cfg.get(addr.pool_idx as usize).unwrap().1; let block_size = self.pool_cfg.cfg.get(addr.pool_idx as usize).unwrap().1;
let raw_pos = self.raw_pos(&addr).unwrap(); let raw_pos = self.raw_pos(&addr).unwrap();
let block = &mut self.pool.get_mut(addr.pool_idx as usize).unwrap()[raw_pos..block_size]; let block =
&mut self.pool.get_mut(addr.pool_idx as usize).unwrap()[raw_pos..raw_pos + block_size];
let size_list = self.sizes_lists.get_mut(addr.pool_idx as usize).unwrap(); let size_list = self.sizes_lists.get_mut(addr.pool_idx as usize).unwrap();
size_list[addr.packet_idx as usize] = Self::STORE_FREE; size_list[addr.packet_idx as usize] = Self::STORE_FREE;
block.fill(0); block.fill(0);
Ok(()) Ok(())
} }
pub fn has_element_at(&self, addr: &StoreAddr) -> Result<bool, StoreError> {
self.validate_addr(addr)?;
let pool_idx = addr.pool_idx as usize;
let size_list = self.sizes_lists.get(pool_idx).unwrap();
let curr_size = size_list[addr.packet_idx as usize];
if curr_size == Self::STORE_FREE {
return Ok(false);
}
Ok(true)
}
fn addr_check(&self, addr: &StoreAddr) -> Result<usize, StoreError> { fn addr_check(&self, addr: &StoreAddr) -> Result<usize, StoreError> {
self.validate_addr(addr)?;
let pool_idx = addr.pool_idx as usize;
let size_list = self.sizes_lists.get(pool_idx).unwrap();
let curr_size = size_list[addr.packet_idx as usize];
if curr_size == Self::STORE_FREE {
return Err(StoreError::DataDoesNotExist(*addr));
}
Ok(curr_size)
}
fn validate_addr(&self, addr: &StoreAddr) -> Result<(), StoreError> {
let pool_idx = addr.pool_idx as usize; let pool_idx = addr.pool_idx as usize;
if pool_idx as usize >= self.pool_cfg.cfg.len() { if pool_idx as usize >= self.pool_cfg.cfg.len() {
return Err(StoreError::InvalidStoreId( return Err(StoreError::InvalidStoreId(
@ -236,12 +289,7 @@ impl LocalPool {
Some(*addr), Some(*addr),
)); ));
} }
let size_list = self.sizes_lists.get(pool_idx).unwrap(); Ok(())
let curr_size = size_list[addr.packet_idx as usize];
if curr_size == Self::STORE_FREE {
return Err(StoreError::DataDoesNotExist(*addr));
}
Ok(curr_size)
} }
fn reserve(&mut self, data_len: usize) -> Result<StoreAddr, StoreError> { fn reserve(&mut self, data_len: usize) -> Result<StoreAddr, StoreError> {
@ -279,7 +327,7 @@ impl LocalPool {
addr addr
)) ))
})?; })?;
let pool_slice = &mut subpool[packet_pos..self.pool_cfg.cfg[addr.pool_idx as usize].1]; let pool_slice = &mut subpool[packet_pos..packet_pos + data.len()];
pool_slice.copy_from_slice(data); pool_slice.copy_from_slice(data);
Ok(()) Ok(())
} }
@ -306,9 +354,78 @@ impl LocalPool {
} }
} }
pub struct PoolGuard<'a> {
pool: &'a mut LocalPool,
pub addr: StoreAddr,
no_deletion: bool,
deletion_failed_error: Option<StoreError>,
}
/// This helper object
impl<'a> PoolGuard<'a> {
pub fn new(pool: &'a mut LocalPool, addr: StoreAddr) -> Self {
Self {
pool,
addr,
no_deletion: false,
deletion_failed_error: None,
}
}
pub fn read(&self) -> Result<&[u8], StoreError> {
self.pool.read(&self.addr)
}
pub fn release(&mut self) {
self.no_deletion = true;
}
}
impl Drop for PoolGuard<'_> {
fn drop(&mut self) {
if !self.no_deletion {
if let Err(e) = self.pool.delete(self.addr) {
self.deletion_failed_error = Some(e);
}
}
}
}
pub struct PoolRwGuard<'a> {
guard: PoolGuard<'a>,
}
impl<'a> PoolRwGuard<'a> {
pub fn new(pool: &'a mut LocalPool, addr: StoreAddr) -> Self {
Self {
guard: PoolGuard::new(pool, addr),
}
}
pub fn modify(&mut self) -> Result<&mut [u8], StoreError> {
self.guard.pool.modify(&self.guard.addr)
}
delegate!(
to self.guard {
pub fn read(&self) -> Result<&[u8], StoreError>;
pub fn release(&mut self);
}
);
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use crate::pool::{LocalPool, PoolCfg, StoreAddr, StoreError, StoreIdError}; use crate::pool::{
LocalPool, PoolCfg, PoolGuard, PoolRwGuard, StoreAddr, StoreError, StoreIdError,
};
use std::vec;
fn basic_small_pool() -> LocalPool {
// 4 buckets of 4 bytes, 2 of 8 bytes and 1 of 16 bytes
let pool_cfg = PoolCfg::new(vec![(4, 4), (2, 8), (1, 16)]);
LocalPool::new(pool_cfg)
}
#[test] #[test]
fn test_cfg() { fn test_cfg() {
@ -330,10 +447,100 @@ mod tests {
} }
#[test] #[test]
fn test_basic() { fn test_add_and_read() {
// 4 buckets of 4 bytes, 2 of 8 bytes and 1 of 16 bytes let mut local_pool = basic_small_pool();
let pool_cfg = PoolCfg::new(vec![(4, 4), (2, 8), (1, 16)]); let mut test_buf: [u8; 16] = [0; 16];
let mut local_pool = LocalPool::new(pool_cfg); for (i, val) in test_buf.iter_mut().enumerate() {
*val = i as u8;
}
let addr = local_pool.add(&test_buf).expect("Adding data failed");
// Read back data and verify correctness
let res = local_pool.read(&addr);
assert!(res.is_ok());
let buf_read_back = res.unwrap();
assert_eq!(buf_read_back.len(), 16);
for (i, &val) in buf_read_back.iter().enumerate() {
assert_eq!(val, i as u8);
}
}
#[test]
fn test_add_smaller_than_full_slot() {
let mut local_pool = basic_small_pool();
let test_buf: [u8; 12] = [0; 12];
let addr = local_pool.add(&test_buf).expect("Adding data failed");
let res = local_pool.read(&addr).expect("Read back failed");
assert_eq!(res.len(), 12);
}
#[test]
fn test_delete() {
let mut local_pool = basic_small_pool();
let test_buf: [u8; 16] = [0; 16];
let addr = local_pool.add(&test_buf).expect("Adding data failed");
// Delete the data
let res = local_pool.delete(addr);
assert!(res.is_ok());
// Verify that the slot is free by trying to get a reference to it
let res = local_pool.free_element(12);
assert!(res.is_ok());
let (addr, buf_ref) = res.unwrap();
assert_eq!(
addr,
StoreAddr {
pool_idx: 2,
packet_idx: 0
}
);
assert_eq!(buf_ref.len(), 12);
}
#[test]
fn test_modify() {
let mut local_pool = basic_small_pool();
let mut test_buf: [u8; 16] = [0; 16];
for (i, val) in test_buf.iter_mut().enumerate() {
*val = i as u8;
}
let addr = local_pool.add(&test_buf).expect("Adding data failed");
{
// Verify that the slot is free by trying to get a reference to it
let res = local_pool.modify(&addr).expect("Modifying data failed");
res[0] = 0;
res[1] = 0x42;
}
let res = local_pool.read(&addr).expect("Reading back data failed");
assert_eq!(res[0], 0);
assert_eq!(res[1], 0x42);
assert_eq!(res[2], 2);
assert_eq!(res[3], 3);
}
#[test]
fn test_consecutive_reservation() {
let mut local_pool = basic_small_pool();
// Reserve two smaller blocks consecutively and verify that the third reservation fails
let res = local_pool.free_element(8);
assert!(res.is_ok());
let (addr0, _) = res.unwrap();
let res = local_pool.free_element(8);
assert!(res.is_ok());
let (addr1, _) = res.unwrap();
let res = local_pool.free_element(8);
assert!(res.is_err());
let err = res.unwrap_err();
assert_eq!(err, StoreError::StoreFull(1));
// Verify that the two deletions are successful
assert!(local_pool.delete(addr0).is_ok());
assert!(local_pool.delete(addr1).is_ok());
}
#[test]
fn test_read_does_not_exist() {
let local_pool = basic_small_pool();
// Try to access data which does not exist // Try to access data which does not exist
let res = local_pool.read(&StoreAddr { let res = local_pool.read(&StoreAddr {
packet_idx: 0, packet_idx: 0,
@ -344,22 +551,13 @@ mod tests {
res.unwrap_err(), res.unwrap_err(),
StoreError::DataDoesNotExist { .. } StoreError::DataDoesNotExist { .. }
)); ));
let mut test_buf: [u8; 16] = [0; 16]; }
for (i, val) in test_buf.iter_mut().enumerate() {
*val = i as u8;
}
let res = local_pool.add(&test_buf);
assert!(res.is_ok());
let addr = res.unwrap();
// Only the second subpool has enough storage and only one bucket
assert_eq!(
addr,
StoreAddr {
pool_idx: 2,
packet_idx: 0
}
);
#[test]
fn test_store_full() {
let mut local_pool = basic_small_pool();
let test_buf: [u8; 16] = [0; 16];
assert!(local_pool.add(&test_buf).is_ok());
// The subpool is now full and the call should fail accordingly // The subpool is now full and the call should fail accordingly
let res = local_pool.add(&test_buf); let res = local_pool.add(&test_buf);
assert!(res.is_err()); assert!(res.is_err());
@ -368,112 +566,121 @@ mod tests {
if let StoreError::StoreFull(subpool) = err { if let StoreError::StoreFull(subpool) = err {
assert_eq!(subpool, 2); assert_eq!(subpool, 2);
} }
}
// Read back data and verify correctness #[test]
fn test_invalid_pool_idx() {
let local_pool = basic_small_pool();
let addr = StoreAddr {
pool_idx: 3,
packet_idx: 0,
};
let res = local_pool.read(&addr); let res = local_pool.read(&addr);
assert!(res.is_ok()); assert!(res.is_err());
let buf_read_back = res.unwrap(); let err = res.unwrap_err();
assert_eq!(buf_read_back.len(), 16); assert!(matches!(
for (i, &val) in buf_read_back.iter().enumerate() { err,
assert_eq!(val, i as u8); StoreError::InvalidStoreId(StoreIdError::InvalidSubpool(3), Some(_))
} ));
}
// Delete the data #[test]
let res = local_pool.delete(addr); fn test_invalid_packet_idx() {
assert!(res.is_ok()); let local_pool = basic_small_pool();
let addr = StoreAddr {
pool_idx: 2,
packet_idx: 1,
};
assert_eq!(addr.raw(), 0x00020001);
let res = local_pool.read(&addr);
assert!(res.is_err());
let err = res.unwrap_err();
assert!(matches!(
err,
StoreError::InvalidStoreId(StoreIdError::InvalidPacketIdx(1), Some(_))
));
}
{ #[test]
// Verify that the slot is free by trying to get a reference to it fn test_add_too_large() {
let res = local_pool.free_element(12); let mut local_pool = basic_small_pool();
assert!(res.is_ok()); let data_too_large = [0; 20];
let (addr, buf_ref) = res.unwrap(); let res = local_pool.add(&data_too_large);
assert_eq!( assert!(res.is_err());
addr, let err = res.unwrap_err();
StoreAddr { assert_eq!(err, StoreError::DataTooLarge(20));
pool_idx: 2, }
packet_idx: 0
}
);
assert_eq!(buf_ref.len(), 12);
assert_eq!(buf_ref, [0; 12]);
buf_ref[0] = 5;
buf_ref[11] = 12;
}
{ #[test]
// Try to request a slot which is too large fn test_data_too_large_1() {
let res = local_pool.free_element(20); let mut local_pool = basic_small_pool();
assert!(res.is_err()); let res = local_pool.free_element(LocalPool::MAX_SIZE + 1);
assert_eq!(res.unwrap_err(), StoreError::DataTooLarge(20)); assert!(res.is_err());
assert_eq!(
res.unwrap_err(),
StoreError::DataTooLarge(LocalPool::MAX_SIZE + 1)
);
}
// Try to modify the 12 bytes requested previously #[test]
let res = local_pool.modify(&addr); fn test_free_element_too_large() {
assert!(res.is_ok()); let mut local_pool = basic_small_pool();
let buf_ref = res.unwrap(); // Try to request a slot which is too large
assert_eq!(buf_ref[0], 5); let res = local_pool.free_element(20);
assert_eq!(buf_ref[11], 12); assert!(res.is_err());
buf_ref[0] = 0; assert_eq!(res.unwrap_err(), StoreError::DataTooLarge(20));
buf_ref[11] = 0; }
}
{ #[test]
let addr = StoreAddr { fn test_pool_guard_deletion_man_creation() {
pool_idx: 3, let mut local_pool = basic_small_pool();
packet_idx: 0, let test_buf: [u8; 16] = [0; 16];
}; let addr = local_pool.add(&test_buf).expect("Adding data failed");
let res = local_pool.read(&addr); let read_guard = PoolGuard::new(&mut local_pool, addr);
assert!(res.is_err()); drop(read_guard);
let err = res.unwrap_err(); assert!(!local_pool.has_element_at(&addr).expect("Invalid address"));
assert!(matches!( }
err,
StoreError::InvalidStoreId(StoreIdError::InvalidSubpool(3), Some(_))
));
}
{ #[test]
let addr = StoreAddr { fn test_pool_guard_deletion() {
pool_idx: 2, let mut local_pool = basic_small_pool();
packet_idx: 1, let test_buf: [u8; 16] = [0; 16];
}; let addr = local_pool.add(&test_buf).expect("Adding data failed");
assert_eq!(addr.raw(), 0x00020001); let read_guard = local_pool.read_with_guard(addr);
let res = local_pool.read(&addr); drop(read_guard);
assert!(res.is_err()); assert!(!local_pool.has_element_at(&addr).expect("Invalid address"));
let err = res.unwrap_err(); }
assert!(matches!(
err,
StoreError::InvalidStoreId(StoreIdError::InvalidPacketIdx(1), Some(_))
));
let data_too_large = [0; 20]; #[test]
let res = local_pool.add(&data_too_large); fn test_pool_guard_with_release() {
assert!(res.is_err()); let mut local_pool = basic_small_pool();
let err = res.unwrap_err(); let test_buf: [u8; 16] = [0; 16];
assert_eq!(err, StoreError::DataTooLarge(20)); let addr = local_pool.add(&test_buf).expect("Adding data failed");
let mut read_guard = PoolGuard::new(&mut local_pool, addr);
read_guard.release();
drop(read_guard);
assert!(local_pool.has_element_at(&addr).expect("Invalid address"));
}
let res = local_pool.free_element(LocalPool::MAX_SIZE + 1); #[test]
assert!(res.is_err()); fn test_pool_modify_guard_man_creation() {
assert_eq!( let mut local_pool = basic_small_pool();
res.unwrap_err(), let test_buf: [u8; 16] = [0; 16];
StoreError::DataTooLarge(LocalPool::MAX_SIZE + 1) let addr = local_pool.add(&test_buf).expect("Adding data failed");
); let mut rw_guard = PoolRwGuard::new(&mut local_pool, addr);
} let _ = rw_guard.modify().expect("modify failed");
drop(rw_guard);
assert!(!local_pool.has_element_at(&addr).expect("Invalid address"));
}
{ #[test]
// Reserve two smaller blocks consecutively and verify that the third reservation fails fn test_pool_modify_guard() {
let res = local_pool.free_element(8); let mut local_pool = basic_small_pool();
assert!(res.is_ok()); let test_buf: [u8; 16] = [0; 16];
let (addr0, _) = res.unwrap(); let addr = local_pool.add(&test_buf).expect("Adding data failed");
let res = local_pool.free_element(8); let mut rw_guard = local_pool.modify_with_guard(addr);
assert!(res.is_ok()); let _ = rw_guard.modify().expect("modify failed");
let (addr1, _) = res.unwrap(); drop(rw_guard);
let res = local_pool.free_element(8); assert!(!local_pool.has_element_at(&addr).expect("Invalid address"));
assert!(res.is_err());
let err = res.unwrap_err();
assert_eq!(err, StoreError::StoreFull(1));
// Verify that the two deletions are successful
assert!(local_pool.delete(addr0).is_ok());
assert!(local_pool.delete(addr1).is_ok());
}
} }
} }

View File

@ -85,6 +85,7 @@
//! mutable_ref.mutable_foo(); //! mutable_ref.mutable_foo();
//! ``` //! ```
use crate::tmtc::{ReceivesCcsdsTc, ReceivesTc}; use crate::tmtc::{ReceivesCcsdsTc, ReceivesTc};
use alloc::boxed::Box;
use downcast_rs::Downcast; use downcast_rs::Downcast;
use spacepackets::{CcsdsPacket, PacketError, SizeMissmatch, SpHeader}; use spacepackets::{CcsdsPacket, PacketError, SizeMissmatch, SpHeader};
@ -195,6 +196,7 @@ pub(crate) mod tests {
use spacepackets::CcsdsPacket; use spacepackets::CcsdsPacket;
use std::collections::VecDeque; use std::collections::VecDeque;
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
use std::vec::Vec;
pub fn generate_ping_tc(buf: &mut [u8]) -> &[u8] { pub fn generate_ping_tc(buf: &mut [u8]) -> &[u8] {
let mut sph = SpHeader::tc(0x002, 0x34, 0).unwrap(); let mut sph = SpHeader::tc(0x002, 0x34, 0).unwrap();

View File

@ -9,7 +9,9 @@ use crate::error::{FsrcErrorRaw, FsrcGroupIds};
use spacepackets::tc::PusTc; use spacepackets::tc::PusTc;
use spacepackets::SpHeader; use spacepackets::SpHeader;
#[cfg(feature = "alloc")]
pub mod ccsds_distrib; pub mod ccsds_distrib;
#[cfg(feature = "alloc")]
pub mod pus_distrib; pub mod pus_distrib;
pub use ccsds_distrib::{CcsdsDistributor, CcsdsError, CcsdsPacketHandler}; pub use ccsds_distrib::{CcsdsDistributor, CcsdsError, CcsdsPacketHandler};

View File

@ -61,6 +61,7 @@
//! assert_eq!(concrete_handler_ref.handler_call_count, 1); //! assert_eq!(concrete_handler_ref.handler_call_count, 1);
//! ``` //! ```
use crate::tmtc::{ReceivesCcsdsTc, ReceivesEcssPusTc, ReceivesTc}; use crate::tmtc::{ReceivesCcsdsTc, ReceivesEcssPusTc, ReceivesTc};
use alloc::boxed::Box;
use downcast_rs::Downcast; use downcast_rs::Downcast;
use spacepackets::ecss::{PusError, PusPacket}; use spacepackets::ecss::{PusError, PusPacket};
use spacepackets::tc::PusTc; use spacepackets::tc::PusTc;
@ -142,7 +143,9 @@ mod tests {
use spacepackets::ecss::PusError; use spacepackets::ecss::PusError;
use spacepackets::tc::PusTc; use spacepackets::tc::PusTc;
use spacepackets::CcsdsPacket; use spacepackets::CcsdsPacket;
#[cfg(feature = "std")]
use std::collections::VecDeque; use std::collections::VecDeque;
#[cfg(feature = "std")]
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
struct PusHandlerSharedQueue { struct PusHandlerSharedQueue {
@ -245,6 +248,7 @@ mod tests {
} }
#[test] #[test]
#[cfg(feature = "std")]
fn test_pus_distribution() { fn test_pus_distribution() {
let known_packet_queue = Arc::new(Mutex::default()); let known_packet_queue = Arc::new(Mutex::default());
let unknown_packet_queue = Arc::new(Mutex::default()); let unknown_packet_queue = Arc::new(Mutex::default());

View File

@ -1,69 +1,34 @@
use fsrc_core::pool::{LocalPool, PoolCfg, StoreAddr, StoreError}; use fsrc_core::pool::{LocalPool, PoolCfg, PoolGuard, StoreAddr};
use std::ops::DerefMut;
use std::sync::mpsc;
use std::sync::mpsc::{Receiver, Sender};
use std::sync::{Arc, RwLock}; use std::sync::{Arc, RwLock};
use std::thread; use std::thread;
struct PoolAccessDummy<'a> { const DUMMY_DATA: [u8; 4] = [0, 1, 2, 3];
pool: &'a mut LocalPool,
addr: Option<StoreAddr>,
no_deletion: bool,
}
impl PoolAccessDummy<'_> { #[test]
#[allow(dead_code)] fn threaded_usage() {
fn modify(&mut self, addr: &StoreAddr) -> Result<&mut [u8], StoreError> {
self.pool.modify(addr)
}
#[allow(dead_code)]
fn release(&mut self) {
self.no_deletion = true;
}
}
impl Drop for PoolAccessDummy<'_> {
fn drop(&mut self) {
if self.no_deletion {
println!("Pool access: Drop with no deletion")
} else {
if let Some(addr) = self.addr {
let res = self.pool.delete(addr);
if res.is_err() {
println!("Pool access: Deletion failed");
}
}
println!("Pool access: Drop with deletion");
}
}
}
fn main() {
println!("Hello World");
let pool_cfg = PoolCfg::new(vec![(16, 6), (32, 3), (8, 12)]); let pool_cfg = PoolCfg::new(vec![(16, 6), (32, 3), (8, 12)]);
let shared_dummy = Arc::new(RwLock::new(LocalPool::new(pool_cfg))); let shared_dummy = Arc::new(RwLock::new(LocalPool::new(pool_cfg)));
let _shared_clone = shared_dummy.clone(); let shared_clone = shared_dummy.clone();
let jh0 = thread::spawn(move || loop { let (tx, rx): (Sender<StoreAddr>, Receiver<StoreAddr>) = mpsc::channel();
{ let jh0 = thread::spawn(move || {
let mut dummy = shared_dummy.write().unwrap(); let mut dummy = shared_dummy.write().unwrap();
let dummy_data = [0, 1, 2, 3]; let addr = dummy.add(&DUMMY_DATA).expect("Writing data failed");
let _addr = dummy.add(&dummy_data).expect("Writing data failed"); tx.send(addr).expect("Sending store address failed");
// let mut accessor = dummy.modify_with_accessor();
// let buf = accessor.modify();
}
}); });
let jh1 = thread::spawn(move || loop { let jh1 = thread::spawn(move || {
let mut pool_access = shared_clone.write().unwrap();
let addr;
{ {
// let dummy = shared_clone.read().unwrap(); addr = rx.recv().expect("Receiving store address failed");
// let buf = dummy.read(); let pg = PoolGuard::new(pool_access.deref_mut(), addr);
// println!("Buffer 0: {:?}", buf[0]); let read_res = pg.read().expect("Reading failed");
assert_eq!(read_res, DUMMY_DATA);
} }
assert!(!pool_access.has_element_at(&addr).expect("Invalid address"));
//let mut dummy = shared_clone.write().unwrap();
//let mut accessor = dummy.modify_with_accessor();
//let buf = accessor.modify();
//buf[0] = 3;
//accessor.release();
}); });
jh0.join().unwrap(); jh0.join().unwrap();
jh1.join().unwrap(); jh1.join().unwrap();

View File

@ -11,6 +11,7 @@ use spacepackets::tm::{PusTm, PusTmSecondaryHeader};
use spacepackets::{CcsdsPacket, SpHeader}; use spacepackets::{CcsdsPacket, SpHeader};
use std::net::{IpAddr, SocketAddr}; use std::net::{IpAddr, SocketAddr};
use std::sync::{mpsc, Arc, Mutex}; use std::sync::{mpsc, Arc, Mutex};
use std::sync::mpsc::TryRecvError;
use std::thread; use std::thread;
const PUS_APID: u16 = 0x02; const PUS_APID: u16 = 0x02;
@ -162,7 +163,17 @@ fn main() {
println!("IO error {e}"); println!("IO error {e}");
} }
ReceiveResult::WouldBlock => { ReceiveResult::WouldBlock => {
// TODO: Send TM Here if let Some(recv_addr) = udp_tmtc_server.udp_tc_server.last_sender() {
// TODO: Send TM Here
match udp_tmtc_server.tm_rx.try_recv() {
Ok(addr) => {
udp_tmtc_server.tm_store.lock().expect("Locking TM store failed").pool.read()
udp_tmtc_server.udp_tc_server.socket.send_to()
}
Err(_) => {}
}
}
} }
}, },
} }