rust/ops-sat-rs
6
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 4 Wiki Activity

Networking update

Browse Source
This commit is contained in:
Robin Müller
2024-04-19 17:40:38 +02:00
parent 0da70ab5ac
commit ddac5ceab3
16 changed files with 790 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
src/config.rs
View File

@ -8,6 +8,7 @@ use std::net::Ipv4Addr;
use std::path::{Path, PathBuf};
pub const STOP_FILE_NAME: &str = "stop-experiment";
pub const CONFIG_FILE_NAME: &str = "exp278.toml";
pub const HOME_FOLER_EXPERIMENT: &str = "/home/exp278";
pub const LOG_FOLDER: &str = "logs";
@ -21,7 +22,8 @@ pub const VALID_PACKET_ID_LIST: &[PacketId] = &[PacketId::new_for_tc(true, EXPER
// TODO: Would be nice if this can be commanded as well..
/// Can be enabled to print all SPP packets received from the SPP server on port 4096.
pub const SPP_CLIENT_WIRETAPPING_RX: bool = false;
pub const SPP_CLIENT_WIRETAPPING_RX: bool = true;
pub const SPP_CLIENT_WIRETAPPING_TX: bool = true;
#[derive(Copy, Clone, PartialEq, Eq, Debug, TryFromPrimitive, IntoPrimitive)]
#[repr(u8)]
@ -51,6 +53,80 @@ lazy_static! {
};
}
pub mod cfg_file {
use std::{
fs::File,
io::Read,
path::{Path, PathBuf},
};
use super::{CONFIG_FILE_NAME, HOME_PATH, TCP_SPP_SERVER_PORT};
pub const SPP_CLIENT_PORT_CFG_KEY: &str = "tcp_spp_server_port";
#[derive(Debug)]
pub struct AppCfg {
pub tcp_spp_server_port: u16,
}
impl Default for AppCfg {
fn default() -> Self {
Self {
tcp_spp_server_port: TCP_SPP_SERVER_PORT,
}
}
}
pub fn create_app_config() -> AppCfg {
let mut cfg_path = HOME_PATH.clone();
cfg_path.push(CONFIG_FILE_NAME);
let cfg_path_home = cfg_path.as_path();
let relevant_path = if Path::new(CONFIG_FILE_NAME).exists() {
Some(PathBuf::from(Path::new(CONFIG_FILE_NAME)))
} else if cfg_path_home.exists() {
Some(PathBuf::from(cfg_path_home))
} else {
None
};
let mut app_cfg = AppCfg::default();
if relevant_path.is_none() {
log::warn!("No config file found, using default values");
return app_cfg;
}
let relevant_path = relevant_path.unwrap();
match File::open(relevant_path.as_path()) {
Ok(mut file) => {
let mut toml_str = String::new();
match file.read_to_string(&mut toml_str) {
Ok(_size) => match toml_str.parse::<toml::Table>() {
Ok(table) => {
handle_config_file_table(table, &mut app_cfg);
}
Err(e) => log::error!("error parsing TOML config file: {e}"),
},
Err(e) => log::error!("error reading TOML config file: {e}"),
}
}
Err(e) => log::error!("error opening TOML config file: {e}"),
}
app_cfg
}
#[allow(clippy::collapsible_match)]
pub fn handle_config_file_table(table: toml::Table, app_cfg: &mut AppCfg) {
if let Some(value) = table.get(SPP_CLIENT_PORT_CFG_KEY) {
if let toml::Value::Integer(port) = value {
if *port < 0 {
log::warn!("invalid port value, is negative");
} else {
app_cfg.tcp_spp_server_port = *port as u16
}
}
}
}
}
pub mod tmtc_err {
use super::*;
use satrs::res_code::ResultU16;
@ -140,11 +216,14 @@ pub mod components {
}
pub mod tasks {
use std::time::Duration;
pub const FREQ_MS_UDP_TMTC: u64 = 200;
pub const FREQ_MS_EVENT_HANDLING: u64 = 400;
pub const FREQ_MS_AOCS: u64 = 500;
pub const FREQ_MS_PUS_STACK: u64 = 200;
pub const FREQ_MS_CTRL: u64 = 400;
pub const STOP_CHECK_FREQUENCY: u64 = 400;
pub const STOP_CHECK_FREQUENCY_MS: u64 = 400;
pub const STOP_CHECK_FREQUENCY: Duration = Duration::from_millis(STOP_CHECK_FREQUENCY_MS);
}

18
src/handlers/camera.rs
View File

@ -24,7 +24,6 @@
/// v Y
///
/// see also https://opssat1.esoc.esa.int/dmsf/files/6/view
use crate::requests::CompositeRequest;
use derive_new::new;
use log::debug;
@ -154,10 +153,20 @@ impl IMS100BatchHandler {
self.handle_hk_request(&msg.requestor_info, hk_request);
}
CompositeRequest::Action(action_request) => {
self.handle_action_request(&msg.requestor_info, action_request);
if let Err(e) =
self.handle_action_request(&msg.requestor_info, action_request)
{
log::warn!("camera action request IO error: {e}");
}
}
},
Err(e) => match e {
mpsc::TryRecvError::Empty => break,
mpsc::TryRecvError::Disconnected => {
log::warn!("composite request receiver disconnected");
break;
}
},
Err(_) => {}
}
}
}
@ -236,6 +245,7 @@ impl IMS100BatchHandler {
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub fn take_picture_from_str(
&mut self,
R: &str,
@ -281,4 +291,4 @@ mod tests {
fn test_crc() {
// TODO
}
}
}

5
src/interface/tcp_server.rs
View File

@ -1,7 +1,6 @@
use std::{
collections::VecDeque,
sync::{atomic::AtomicBool, mpsc, Arc, Mutex},
time::Duration,
};
use log::{info, warn};
@ -113,9 +112,7 @@ impl TcpTask {
}
pub fn periodic_operation(&mut self) {
let result = self
.0
.handle_all_connections(Some(Duration::from_millis(STOP_CHECK_FREQUENCY)));
let result = self.0.handle_all_connections(Some(STOP_CHECK_FREQUENCY));
match result {
Ok(_conn_result) => (),
Err(e) => {

394
src/interface/tcp_spp_client.rs
View File

@ -1,12 +1,12 @@
use std::io::{self, Read, Write};
use std::io::{self, Read};
use std::net::TcpStream as StdTcpStream;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::mpsc;
use std::time::Duration;
use mio::net::TcpStream;
use mio::net::TcpStream as MioTcpStream;
use mio::{Events, Interest, Poll, Token};
use ops_sat_rs::config::tasks::STOP_CHECK_FREQUENCY;
use ops_sat_rs::config::{SPP_CLIENT_WIRETAPPING_RX, TCP_SPP_SERVER_PORT};
use ops_sat_rs::config::{SPP_CLIENT_WIRETAPPING_RX, SPP_CLIENT_WIRETAPPING_TX};
use satrs::encoding::ccsds::parse_buffer_for_ccsds_space_packets;
use satrs::queue::GenericSendError;
use satrs::spacepackets::PacketId;
@ -17,19 +17,16 @@ use thiserror::Error;
use super::{SimpleSpValidator, TcpComponent};
#[derive(Debug, Error)]
pub enum PacketForwardingError {
pub enum ClientError {
#[error("send error: {0}")]
Send(#[from] GenericSendError),
#[error("io error: {0}")]
Io(#[from] io::Error),
}
pub struct TcpSppClient {
#[allow(dead_code)]
pub struct TcpSppClientCommon {
id: ComponentId,
poll: Poll,
events: Events,
// Optional to allow periodic reconnection attempts on the TCP server.
client: Option<TcpStream>,
read_buf: [u8; 4096],
tm_tcp_client_rx: mpsc::Receiver<PacketAsVec>,
server_addr: SocketAddr,
@ -37,143 +34,13 @@ pub struct TcpSppClient {
validator: SimpleSpValidator,
}
impl TcpSppClient {
pub fn new(
id: ComponentId,
tc_source_tx: mpsc::Sender<PacketAsVec>,
tm_tcp_client_rx: mpsc::Receiver<PacketAsVec>,
valid_ids: &'static [PacketId],
) -> io::Result<Self> {
let mut poll = Poll::new()?;
let events = Events::with_capacity(128);
let server_addr =
SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 1, 1)), TCP_SPP_SERVER_PORT);
let client = Self::attempt_connection(&mut poll, &server_addr);
if client.is_err() {
log::warn!(
"connection to TCP server {} failed: {}",
server_addr,
client.unwrap_err()
);
return Ok(Self {
id,
poll,
events,
client: None,
read_buf: [0; 4096],
server_addr,
tm_tcp_client_rx,
tc_source_tx,
validator: SimpleSpValidator::new(TcpComponent::Client, valid_ids.to_vec()),
});
}
Ok(Self {
id,
poll,
events,
client: Some(client.unwrap()),
read_buf: [0; 4096],
server_addr,
tm_tcp_client_rx,
tc_source_tx,
validator: SimpleSpValidator::new(TcpComponent::Client, valid_ids.to_vec()),
})
}
pub fn attempt_connection(poll: &mut Poll, server_addr: &SocketAddr) -> io::Result<TcpStream> {
let mut client = TcpStream::connect(*server_addr)?;
poll.registry().register(
&mut client,
Token(0),
Interest::READABLE | Interest::WRITABLE,
)?;
Ok(client)
}
pub fn periodic_operation(&mut self) -> Result<(), PacketForwardingError> {
if self.client.is_some() {
return self.perform_regular_operation();
} else {
let client_result = Self::attempt_connection(&mut self.poll, &self.server_addr);
match client_result {
Ok(client) => {
self.client = Some(client);
self.perform_regular_operation()?;
}
Err(ref e) => {
log::warn!(
"connection to TCP server {} failed: {}",
self.server_addr,
e
);
}
}
}
Ok(())
}
pub fn perform_regular_operation(&mut self) -> Result<(), PacketForwardingError> {
self.poll.poll(
&mut self.events,
Some(Duration::from_millis(STOP_CHECK_FREQUENCY)),
)?;
let events: Vec<mio::event::Event> = self.events.iter().cloned().collect();
for event in events {
if event.token() == Token(0) {
if event.is_readable() {
self.read_from_server()?;
}
if event.is_writable() {
self.write_to_server()?;
}
}
}
Ok(())
}
pub fn read_from_server(&mut self) -> Result<(), PacketForwardingError> {
let client = self
.client
.as_mut()
.expect("TCP stream invalid when it should not be");
match client.read(&mut self.read_buf) {
Ok(0) => return Err(io::Error::from(io::ErrorKind::BrokenPipe).into()),
Ok(read_bytes) => self.handle_read_bytstream(read_bytes)?,
Err(e) => return Err(e.into()),
}
Ok(())
}
pub fn write_to_server(&mut self) -> io::Result<()> {
let client = self
.client
.as_mut()
.expect("TCP stream invalid when it should not be");
loop {
match self.tm_tcp_client_rx.try_recv() {
Ok(tm) => {
client.write_all(&tm.packet)?;
}
Err(e) => match e {
mpsc::TryRecvError::Empty => break,
mpsc::TryRecvError::Disconnected => {
log::error!("TM sender to TCP client has disconnected");
break;
}
},
}
}
Ok(())
}
pub fn handle_read_bytstream(
&mut self,
read_bytes: usize,
) -> Result<(), PacketForwardingError> {
#[allow(dead_code)]
impl TcpSppClientCommon {
pub fn handle_read_bytstream(&mut self, read_bytes: usize) -> Result<(), ClientError> {
let mut dummy = 0;
if SPP_CLIENT_WIRETAPPING_RX {
log::debug!(
"received {} bytes on TCP client: {:x?}",
"SPP TCP RX {} bytes: {:x?}",
read_bytes,
&self.read_buf[..read_bytes]
);
@ -188,4 +55,243 @@ impl TcpSppClient {
)?;
Ok(())
}
pub fn write_to_server(&mut self, client: &mut impl io::Write) -> io::Result<()> {
loop {
match self.tm_tcp_client_rx.try_recv() {
Ok(tm) => {
if SPP_CLIENT_WIRETAPPING_TX {
log::debug!(
"SPP TCP TX {}: {:x?}",
tm.packet.len(),
tm.packet.as_slice()
);
}
client.write_all(&tm.packet)?;
}
Err(e) => match e {
mpsc::TryRecvError::Empty => break,
mpsc::TryRecvError::Disconnected => {
log::error!("TM sender to TCP client has disconnected");
break;
}
},
}
}
Ok(())
}
}
#[derive(Debug, PartialEq, Eq)]
pub enum ConnectionStatus {
Unknown,
Connected,
LostConnection,
TryingReconnect,
}
#[allow(dead_code)]
pub struct TcpSppClientMio {
common: TcpSppClientCommon,
poll: Poll,
events: Events,
// Optional to allow periodic reconnection attempts on the TCP server.
client: Option<MioTcpStream>,
connection: ConnectionStatus,
}
#[allow(dead_code)]
impl TcpSppClientMio {
pub fn new(
id: ComponentId,
tc_source_tx: mpsc::Sender<PacketAsVec>,
tm_tcp_client_rx: mpsc::Receiver<PacketAsVec>,
valid_ids: &'static [PacketId],
port: u16,
) -> io::Result<Self> {
let poll = Poll::new()?;
let events = Events::with_capacity(128);
let mut client = Self {
common: TcpSppClientCommon {
id,
read_buf: [0; 4096],
tm_tcp_client_rx,
server_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port),
tc_source_tx,
validator: SimpleSpValidator::new(TcpComponent::Client, valid_ids.to_vec()),
},
poll,
events,
client: None,
connection: ConnectionStatus::Unknown,
};
client.connect()?;
Ok(client)
}
pub fn connect(&mut self) -> io::Result<()> {
let mut client = MioTcpStream::connect(self.common.server_addr)?;
self.poll.registry().register(
&mut client,
Token(0),
Interest::READABLE | Interest::WRITABLE,
)?;
self.client = Some(client);
self.connection = ConnectionStatus::TryingReconnect;
Ok(())
}
pub fn operation(&mut self) -> Result<(), ClientError> {
match self.connection {
ConnectionStatus::TryingReconnect | ConnectionStatus::Unknown => {
self.check_conn_status()?
}
ConnectionStatus::Connected => {
self.check_conn_status()?;
self.poll
.poll(&mut self.events, Some(STOP_CHECK_FREQUENCY))?;
let events: Vec<mio::event::Event> = self.events.iter().cloned().collect();
for event in events {
if event.token() == Token(0) {
if event.is_readable() {
log::warn!("TCP client is readable");
self.read_from_server()?;
}
if event.is_writable() {
log::warn!("TCP client is writable");
self.common.write_to_server(self.client.as_mut().unwrap())?;
}
}
}
return Ok(());
}
ConnectionStatus::LostConnection => self.connect()?,
};
std::thread::sleep(STOP_CHECK_FREQUENCY);
Ok(())
}
pub fn read_from_server(&mut self) -> Result<(), ClientError> {
match self
.client
.as_mut()
.unwrap()
.read(&mut self.common.read_buf)
{
Ok(0) => (),
Ok(read_bytes) => self.common.handle_read_bytstream(read_bytes)?,
Err(e) => return Err(e.into()),
}
Ok(())
}
pub fn check_conn_status(&mut self) -> io::Result<()> {
match self.client.as_mut().unwrap().peer_addr() {
Ok(_) => {
if self.connection == ConnectionStatus::Unknown
|| self.connection == ConnectionStatus::TryingReconnect
{
self.connection = ConnectionStatus::Connected;
}
Ok(())
}
Err(e) => {
if e.kind() == io::ErrorKind::NotConnected {
log::warn!("lost connection, or do not have one");
self.connection = ConnectionStatus::LostConnection;
return Ok(());
}
Err(e)
}
}
}
}
pub struct TcpSppClientStd {
common: TcpSppClientCommon,
// Optional to allow periodic reconnection attempts on the TCP server.
stream: Option<StdTcpStream>,
}
impl TcpSppClientStd {
pub fn new(
id: ComponentId,
tc_source_tx: mpsc::Sender<PacketAsVec>,
tm_tcp_client_rx: mpsc::Receiver<PacketAsVec>,
valid_ids: &'static [PacketId],
port: u16,
) -> io::Result<Self> {
let server_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port);
let client = match StdTcpStream::connect(server_addr) {
Ok(stream) => {
stream.set_read_timeout(Some(STOP_CHECK_FREQUENCY))?;
Some(stream)
}
Err(e) => {
log::warn!("error connecting to server: {}", e);
None
}
};
Ok(Self {
common: TcpSppClientCommon {
id,
read_buf: [0; 4096],
tm_tcp_client_rx,
server_addr,
tc_source_tx,
validator: SimpleSpValidator::new(TcpComponent::Client, valid_ids.to_vec()),
},
stream: client,
})
}
pub fn attempt_connect(&mut self, log_error: bool) -> io::Result<bool> {
Ok(match StdTcpStream::connect(self.common.server_addr) {
Ok(stream) => {
stream.set_read_timeout(Some(STOP_CHECK_FREQUENCY))?;
self.stream = Some(stream);
true
}
Err(e) => {
if log_error {
log::warn!("error connecting to server: {}", e);
}
false
}
})
}
pub fn operation(&mut self) -> Result<(), ClientError> {
if let Some(client) = &mut self.stream {
match client.read(&mut self.common.read_buf) {
// Not sure if this can happen or this is actually an error condition..
Ok(0) => {
log::info!("server closed connection");
self.stream = None;
}
Ok(read_bytes) => self.common.handle_read_bytstream(read_bytes)?,
Err(e) => {
if e.kind() == io::ErrorKind::WouldBlock || e.kind() == io::ErrorKind::TimedOut
{
self.common.write_to_server(client)?;
return Ok(());
}
log::warn!("server error: {e:?}");
if e.kind() == io::ErrorKind::ConnectionReset {
self.stream = None;
}
return Err(e.into());
}
}
} else {
if self.attempt_connect(false)? {
log::info!("reconnected to server succesfully");
return self.operation();
}
std::thread::sleep(STOP_CHECK_FREQUENCY);
}
Ok(())
}
}

19
src/main.rs
View File

@ -7,6 +7,7 @@ use std::{
use log::info;
use ops_sat_rs::config::{
cfg_file::create_app_config,
components::{CONTROLLER_ID, TCP_SERVER, TCP_SPP_CLIENT, UDP_SERVER},
tasks::{FREQ_MS_CTRL, FREQ_MS_PUS_STACK},
VALID_PACKET_ID_LIST,
@ -14,7 +15,7 @@ use ops_sat_rs::config::{
use ops_sat_rs::config::{tasks::FREQ_MS_UDP_TMTC, OBSW_SERVER_ADDR, SERVER_PORT};
use satrs::hal::std::{tcp_server::ServerConfig, udp_server::UdpTcServer};
use crate::tmtc::tc_source::TcSourceTaskDynamic;
use crate::pus::{PusTcDistributor, PusTcMpscRouter};
use crate::tmtc::tm_sink::TmFunnelDynamic;
use crate::{controller::ExperimentController, pus::test::create_test_service};
use crate::{
@ -22,10 +23,7 @@ use crate::{
interface::udp_server::{DynamicUdpTmHandler, UdpTmtcServer},
logger::setup_logger,
};
use crate::{
interface::tcp_spp_client::TcpSppClient,
pus::{PusTcDistributor, PusTcMpscRouter},
};
use crate::{interface::tcp_spp_client::TcpSppClientStd, tmtc::tc_source::TcSourceTaskDynamic};
use crate::{
pus::{action::create_action_service, stack::PusStack},
requests::GenericRequestRouter,
@ -43,6 +41,9 @@ fn main() {
setup_logger().expect("setting up logging with fern failed");
println!("OPS-SAT Rust Experiment OBSW");
let app_cfg = create_app_config();
println!("App Configuration: {:?}", app_cfg);
let stop_signal = Arc::new(AtomicBool::new(false));
let (tc_source_tx, tc_source_rx) = mpsc::channel();
@ -165,11 +166,12 @@ fn main() {
stop_signal.clone(),
);
let mut tcp_spp_client = TcpSppClient::new(
let mut tcp_spp_client = TcpSppClientStd::new(
TCP_SPP_CLIENT.id(),
tc_source_tx,
tm_tcp_client_rx,
VALID_PACKET_ID_LIST,
app_cfg.tcp_spp_server_port,
)
.expect("creating TCP SPP client failed");
@ -228,7 +230,10 @@ fn main() {
.spawn(move || {
info!("Running TCP SPP client");
loop {
let _result = tcp_spp_client.periodic_operation();
let result = tcp_spp_client.operation();
if let Err(e) = result {
log::error!("TCP SPP client error: {}", e);
}
if tcp_client_stop_signal.load(std::sync::atomic::Ordering::Relaxed) {
break;
}

1
src/pus/test.rs
View File

@ -42,7 +42,6 @@ pub struct TestCustomServiceWrapper {
EcssTcInVecConverter,
VerificationReporter,
>,
// pub test_srv_event_sender: mpsc::Sender<EventMessageU32>,
}
impl TestCustomServiceWrapper {

7
src/tmtc/tm_sink.rs
View File

@ -1,6 +1,6 @@
use std::sync::atomic::AtomicBool;
use std::sync::Arc;
use std::{collections::HashMap, sync::mpsc, time::Duration};
use std::{collections::HashMap, sync::mpsc};
use log::info;
use ops_sat_rs::config::tasks::STOP_CHECK_FREQUENCY;
@ -101,10 +101,7 @@ impl TmFunnelDynamic {
pub fn operation(&mut self) {
loop {
match self
.tm_funnel_rx
.recv_timeout(Duration::from_millis(STOP_CHECK_FREQUENCY))
{
match self.tm_funnel_rx.recv_timeout(STOP_CHECK_FREQUENCY) {
Ok(mut tm) => {
// Read the TM, set sequence counter and message counter, and finally update
// the CRC.