Skip to content

R
RoboPLC
提交 95caa6dc authored 作者: Serhij S's avatar Serhij S
浏览文件
操作

mock all methods on non-linux platforms

上级 f0f993eb
显示空白字符变更
内嵌 并排
正在显示 包含 75 行增加 和 21 行删除
src/controller.rs
浏览文件 @ 95caa6dc
......@@ -7,16 +7,18 @@ use std::{
time::Duration,
};
#[cfg(target_os = "linux")]
use crate::suicide;
use crate::{
critical,
hub::Hub,
suicide,
supervisor::Supervisor,
thread_rt::{Builder, RTParams, Scheduling},
Error, Result,
};
pub use roboplc_derive::WorkerOpts;
use rtsc::data_policy::DataDeliveryPolicy;
#[cfg(target_os = "linux")]
use signal_hook::{
consts::{SIGINT, SIGTERM},
iterator::Signals,
......@@ -203,7 +205,7 @@ where
pub fn register_signals_with_shutdown_handler<H>(
&mut self,
handle_fn: H,
shutdown_timeout: Duration,
#[allow(unused_variables)] shutdown_timeout: Duration,
) -> Result<()>
where
H: Fn(&Context<D, V>) + Send + Sync + 'static,
......@@ -218,6 +220,8 @@ where
builder.park_on_errors = true;
macro_rules! sig_handler {
($handler: expr) => {{
#[cfg(target_os = "linux")]
{
let context = self.context();
let mut signals = Signals::new([SIGTERM, SIGINT])?;
move || {
......@@ -232,8 +236,14 @@ where
}
}
}
}
#[cfg(not(target_os = "linux"))]
{
move || {}
}
}};
}
#[allow(unused_variables)]
let h = handler.clone();
if let Err(e) = self.supervisor.spawn(builder.clone(), sig_handler!(h)) {
if !matches!(e, Error::RTSchedSetSchduler(_)) {
......
src/lib.rs
浏览文件 @ 95caa6dc
......@@ -6,7 +6,6 @@ use std::panic::PanicInfo;
use std::{env, sync::Arc, time::Duration};
use colored::Colorize as _;
#[cfg(target_os = "linux")]
use thread_rt::{RTParams, Scheduling};
pub use log::LevelFilter;
......@@ -90,7 +89,6 @@ pub use rtsc::data_policy::{DataDeliveryPolicy, DeliveryPolicy};
/// Reliable TCP/Serial communications
pub mod comm;
/// Controller and workers
#[cfg(target_os = "linux")]
pub mod controller;
/// In-process data communication pub/sub hub, synchronous edition
pub mod hub;
......@@ -100,10 +98,8 @@ pub mod hub_async;
/// I/O
pub mod io;
/// Task supervisor to manage real-time threads
#[cfg(target_os = "linux")]
pub mod supervisor;
/// Real-time thread functions to work with [`supervisor::Supervisor`] and standalone
#[cfg(target_os = "linux")]
/// Real-time thread functions to work with [`supervisor::Supervisor`] and standalone, Linux only
pub mod thread_rt;
/// The crate result type
......@@ -245,7 +241,6 @@ impl Error {
}
/// Immediately kills the current process and all its subprocesses with a message to stderr
#[cfg(target_os = "linux")]
pub fn critical(msg: &str) -> ! {
eprintln!("{}", msg.red().bold());
thread_rt::suicide_myself(Duration::from_secs(0), false);
......@@ -257,7 +252,6 @@ pub fn critical(msg: &str) -> ! {
/// period of time.
///
/// Prints warnings to STDOUT if warn is true
#[cfg(target_os = "linux")]
pub fn suicide(delay: Duration, warn: bool) {
let mut builder = thread_rt::Builder::new().name("suicide").rt_params(
RTParams::new()
......@@ -336,14 +330,12 @@ pub fn metrics_exporter_install(
/// Sets panic handler to immediately kill the process and its childs with SIGKILL. The process is
/// killed when panic happens in ANY thread
#[cfg(target_os = "linux")]
pub fn setup_panic() {
std::panic::set_hook(Box::new(move |info: &PanicInfo| {
panic(info);
}));
}
#[cfg(target_os = "linux")]
fn panic(info: &PanicInfo) -> ! {
eprintln!("{}", info.to_string().red().bold());
thread_rt::suicide_myself(Duration::from_secs(0), false);
......@@ -372,13 +364,10 @@ pub fn configure_logger(filter: LevelFilter) {
/// Prelude module
pub mod prelude {
#[cfg(target_os = "linux")]
pub use super::suicide;
#[cfg(target_os = "linux")]
pub use crate::controller::*;
pub use crate::hub::prelude::*;
pub use crate::io::prelude::*;
#[cfg(target_os = "linux")]
pub use crate::supervisor::prelude::*;
pub use crate::time::DurationRT;
pub use bma_ts::{Monotonic, Timestamp};
......
src/supervisor.rs
浏览文件 @ 95caa6dc
......@@ -29,8 +29,12 @@ impl<T> Default for Supervisor<T> {
macro_rules! vacant_entry {
($self:ident, $builder:ident) => {{
let Some(name) = $builder.name.clone() else { return Err(Error::SupervisorNameNotSpecified); };
let btree_map::Entry::Vacant(entry) = $self.tasks.entry(name.clone()) else { return Err(Error::SupervisorDuplicateTask(name)); };
let Some(name) = $builder.name.clone() else {
return Err(Error::SupervisorNameNotSpecified);
};
let btree_map::Entry::Vacant(entry) = $self.tasks.entry(name.clone()) else {
return Err(Error::SupervisorDuplicateTask(name));
};
entry
}};
}
......
src/thread_rt.rs
浏览文件 @ 95caa6dc
......@@ -2,19 +2,22 @@ use crate::{time::Interval, Error, Result};
use bma_ts::{Monotonic, Timestamp};
use colored::Colorize;
use core::fmt;
#[cfg(target_os = "linux")]
use libc::cpu_set_t;
#[cfg(target_os = "linux")]
use nix::{sys::signal, unistd};
use serde::{Deserialize, Serialize, Serializer};
use std::{
collections::{BTreeMap, BTreeSet},
fs,
io::BufRead,
mem,
sync::atomic::{AtomicBool, Ordering},
thread::{self, JoinHandle, Scope, ScopedJoinHandle},
time::Duration,
};
use sysinfo::{Pid, PidExt, ProcessExt, System, SystemExt};
#[cfg(target_os = "linux")]
use sysinfo::PidExt;
use sysinfo::{Pid, ProcessExt, System, SystemExt};
use tracing::warn;
static REALTIME_MODE: AtomicBool = AtomicBool::new(true);
......@@ -29,6 +32,12 @@ fn is_realtime() -> bool {
REALTIME_MODE.load(Ordering::Relaxed)
}
macro_rules! panic_os {
() => {
panic!("The function is not supported on this OS");
};
}
/// The method preallocates a heap memory region with the given size. The method is useful to
/// prevent memory fragmentation and speed up memory allocation. It is highly recommended to call
/// the method at the beginning of the program.
......@@ -38,10 +47,13 @@ fn is_realtime() -> bool {
/// # Panics
///
/// Will panic if the page size is too large (more than usize)
#[allow(unused_variables)]
pub fn prealloc_heap(size: usize) -> Result<()> {
if !is_realtime() {
return Ok(());
}
#[cfg(target_os = "linux")]
{
let page_size = unsafe {
if libc::mallopt(libc::M_MMAP_MAX, 0) != 1 {
return Err(Error::failed(
......@@ -63,6 +75,11 @@ pub fn prealloc_heap(size: usize) -> Result<()> {
}
})();
Ok(())
}
#[cfg(not(target_os = "linux"))]
{
panic_os!();
}
}
/// A thread builder object, similar to [`thread::Builder`] but with real-time capabilities
......@@ -100,6 +117,7 @@ pub enum Scheduling {
Other,
}
#[cfg(target_os = "linux")]
impl From<Scheduling> for libc::c_int {
fn from(value: Scheduling) -> Self {
match value {
......@@ -113,6 +131,7 @@ impl From<Scheduling> for libc::c_int {
}
}
#[cfg(target_os = "linux")]
impl From<libc::c_int> for Scheduling {
fn from(value: libc::c_int) -> Self {
match value {
......@@ -488,10 +507,13 @@ impl RTParams {
}
}
#[allow(unused_variables)]
fn thread_init_internal(
tx_tid: oneshot::Sender<(libc::c_int, oneshot::Sender<bool>)>,
park_on_errors: bool,
) {
#[cfg(target_os = "linux")]
{
let tid = unsafe { i32::try_from(libc::syscall(libc::SYS_gettid)).unwrap_or(-200) };
let (tx_ok, rx_ok) = oneshot::channel::<bool>();
tx_tid.send((tid, tx_ok)).unwrap();
......@@ -507,8 +529,14 @@ fn thread_init_internal(
);
}
}
}
#[cfg(not(target_os = "linux"))]
{
panic_os!();
}
}
#[allow(unused_variables)]
fn thread_init_external(
rx_tid: oneshot::Receiver<(libc::c_int, oneshot::Sender<bool>)>,
params: &RTParams,
......@@ -527,17 +555,21 @@ fn thread_init_external(
Ok(tid)
}
#[allow(unused_variables)]
fn apply_thread_params(tid: libc::c_int, params: &RTParams, quiet: bool) -> Result<()> {
if !is_realtime() {
return Ok(());
}
#[cfg(target_os = "linux")]
{
if !params.cpu_ids.is_empty() {
unsafe {
let mut cpuset: cpu_set_t = mem::zeroed();
let mut cpuset: cpu_set_t = std::mem::zeroed();
for cpu in &params.cpu_ids {
libc::CPU_SET(*cpu, &mut cpuset);
}
let res = libc::sched_setaffinity(tid, std::mem::size_of::<libc::cpu_set_t>(), &cpuset);
let res =
libc::sched_setaffinity(tid, std::mem::size_of::<libc::cpu_set_t>(), &cpuset);
if res != 0 {
if !quiet {
eprintln!(
......@@ -570,6 +602,11 @@ fn apply_thread_params(tid: libc::c_int, params: &RTParams, quiet: bool) -> Resu
}
}
Ok(())
}
#[cfg(not(target_os = "linux"))]
{
panic_os!();
}
}
macro_rules! impl_serialize_join_handle {
......@@ -597,15 +634,22 @@ pub(crate) fn suicide_myself(delay: Duration, warn: bool) {
eprintln!("{}", "KILLING THE PROCESS".red().bold());
}
kill_pstree(pid as i32, false, None);
#[cfg(target_os = "linux")]
let _ = signal::kill(unistd::Pid::from_raw(pid as i32), signal::Signal::SIGKILL);
#[cfg(not(target_os = "linux"))]
{
panic_os!();
}
}
/// Terminates a process tree with SIGTERM, waits "term_kill_interval" and repeats the opeation
/// with SIGKILL
///
/// If "term_kill_interval" is not set, SIGKILL is used immediately.
#[allow(clippy::cast_possible_wrap, clippy::cast_sign_loss)]
#[allow(clippy::cast_possible_wrap, clippy::cast_sign_loss, unused_variables)]
pub fn kill_pstree(pid: i32, kill_parent: bool, term_kill_interval: Option<Duration>) {
#[cfg(target_os = "linux")]
{
let mut sys = System::new();
sys.refresh_processes();
let mut pids = BTreeSet::new();
......@@ -627,8 +671,14 @@ pub fn kill_pstree(pid: i32, kill_parent: bool, term_kill_interval: Option<Durat
signal::Signal::SIGTERM,
kill_parent,
);
}
#[cfg(not(target_os = "linux"))]
{
panic_os!();
}
}
#[cfg(target_os = "linux")]
fn kill_process_tree(
pid: Pid,
sys: &mut sysinfo::System,
......@@ -647,6 +697,7 @@ fn kill_process_tree(
}
}
#[allow(dead_code)]
fn get_child_pids_recursive(pid: Pid, sys: &System, to: &mut BTreeSet<Pid>) {
for (i, p) in sys.processes() {
if let Some(parent) = p.parent() {
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论