use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use crate::error::{CaError, CaResult};
use crate::runtime::net::cas_server_port;
use crate::server::record::{self, Record, SubroutineFn};
use crate::server::database::PvDatabase;
use crate::server::device_support::DeviceSupport;
use crate::server::iocsh::{self, registry::CommandDef};
use crate::server::{DeviceSupportFactory, access_security, autosave};
use autosave::startup::AutosaveStartupConfig;
pub mod init_hooks {
use std::sync::{Arc, Mutex};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum InitHookState {
AtIocBuild,
AtBeginning,
AfterCallbackInit,
AfterCaLinkInit,
AfterInitDrvSup,
AfterInitRecSup,
AfterInitDevSup,
AfterInitDatabase,
AfterFinishDevSup,
AfterScanInit,
AfterInitialProcess,
AfterCaServerInit,
AfterIocBuilt,
AtIocRun,
AfterDatabaseRunning,
AfterCaServerRunning,
AfterIocRunning,
}
impl InitHookState {
pub fn name(&self) -> &'static str {
match self {
InitHookState::AtIocBuild => "initHookAtIocBuild",
InitHookState::AtBeginning => "initHookAtBeginning",
InitHookState::AfterCallbackInit => "initHookAfterCallbackInit",
InitHookState::AfterCaLinkInit => "initHookAfterCaLinkInit",
InitHookState::AfterInitDrvSup => "initHookAfterInitDrvSup",
InitHookState::AfterInitRecSup => "initHookAfterInitRecSup",
InitHookState::AfterInitDevSup => "initHookAfterInitDevSup",
InitHookState::AfterInitDatabase => "initHookAfterInitDatabase",
InitHookState::AfterFinishDevSup => "initHookAfterFinishDevSup",
InitHookState::AfterScanInit => "initHookAfterScanInit",
InitHookState::AfterInitialProcess => "initHookAfterInitialProcess",
InitHookState::AfterCaServerInit => "initHookAfterCaServerInit",
InitHookState::AfterIocBuilt => "initHookAfterIocBuilt",
InitHookState::AtIocRun => "initHookAtIocRun",
InitHookState::AfterDatabaseRunning => "initHookAfterDatabaseRunning",
InitHookState::AfterCaServerRunning => "initHookAfterCaServerRunning",
InitHookState::AfterIocRunning => "initHookAfterIocRunning",
}
}
}
pub type InitHookFunction = Arc<dyn Fn(InitHookState) + Send + Sync>;
static HOOKS: Mutex<Vec<InitHookFunction>> = Mutex::new(Vec::new());
pub fn init_hook_register(func: InitHookFunction) {
HOOKS.lock().unwrap().push(func);
}
pub fn init_hook_announce(state: InitHookState) {
let snapshot: Vec<InitHookFunction> = HOOKS.lock().unwrap().clone();
for cb in snapshot {
cb(state);
}
}
#[cfg(test)]
pub fn init_hook_free() {
HOOKS.lock().unwrap().clear();
}
}
pub use init_hooks::{InitHookFunction, InitHookState, init_hook_announce, init_hook_register};
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct GroupLoadRequest {
pub filename: String,
pub macros: String,
}
static GROUP_LOAD_REQUESTS: std::sync::LazyLock<Mutex<Vec<GroupLoadRequest>>> =
std::sync::LazyLock::new(|| Mutex::new(Vec::new()));
pub fn take_group_load_requests() -> Vec<GroupLoadRequest> {
std::mem::take(&mut *GROUP_LOAD_REQUESTS.lock().unwrap())
}
pub fn db_load_group_startup_command() -> CommandDef {
use crate::server::iocsh::registry::{
ArgDesc, ArgType, ArgValue, CommandContext, CommandOutcome,
};
CommandDef::new(
"dbLoadGroup",
vec![
ArgDesc {
name: "filename",
arg_type: ArgType::String,
optional: false,
},
ArgDesc {
name: "macros",
arg_type: ArgType::String,
optional: true,
},
],
"dbLoadGroup <jsonFilename> [<macros>]",
move |args: &[ArgValue], ctx: &CommandContext| {
let filename = match args.first() {
Some(ArgValue::String(s)) => s.clone(),
_ => return Err("dbLoadGroup: missing filename".into()),
};
let macros = match args.get(1) {
Some(ArgValue::String(s)) => s.clone(),
_ => String::new(),
};
let mut queue = GROUP_LOAD_REQUESTS.lock().unwrap();
if let Some(rest) = filename.strip_prefix('-') {
if rest == "*" {
let n = queue.len();
queue.clear();
ctx.println(&format!(
"dbLoadGroup: cleared all queued group files ({n} removed)"
));
} else {
let before = queue.len();
queue.retain(|r| !(r.filename == rest && r.macros == macros));
let dropped = before - queue.len();
ctx.println(&format!(
"dbLoadGroup: removed '{rest}' ({dropped} queued entr{} dropped)",
if dropped == 1 { "y" } else { "ies" }
));
}
return Ok(CommandOutcome::Continue);
}
if let Err(e) = std::fs::metadata(&filename) {
return Err(format!("dbLoadGroup: error opening \"{filename}\": {e}"));
}
queue.retain(|r| !(r.filename == filename && r.macros == macros));
queue.push(GroupLoadRequest {
filename: filename.clone(),
macros,
});
ctx.println(&format!(
"dbLoadGroup: queued '{filename}' ({} group file(s) queued)",
queue.len()
));
Ok(CommandOutcome::Continue)
},
)
}
pub struct DeviceSupportContext<'a> {
pub dtyp: &'a str,
pub inp: &'a str,
pub out: &'a str,
}
pub type DynamicDeviceSupportFactory =
Box<dyn Fn(&DeviceSupportContext) -> Option<Box<dyn DeviceSupport>> + Send + Sync>;
pub type LinkSetInstaller = Box<
dyn FnOnce(
Arc<PvDatabase>,
) -> std::pin::Pin<
Box<dyn std::future::Future<Output = Vec<CommandDef>> + Send + 'static>,
> + Send
+ 'static,
>;
pub struct IocRunConfig {
pub db: Arc<PvDatabase>,
pub port: u16,
pub tcp_port: Option<u16>,
pub acf: Option<access_security::AccessSecurityConfig>,
pub autosave_config: Option<autosave::SaveSetConfig>,
pub autosave_manager: Option<Arc<autosave::AutosaveManager>>,
pub shell_commands: Vec<CommandDef>,
pub after_init_hooks: Vec<Box<dyn FnOnce() + Send>>,
}
pub struct IocApplication {
port: u16,
tcp_port: Option<u16>,
device_factories: HashMap<String, DeviceSupportFactory>,
dynamic_device_factory: Option<DynamicDeviceSupportFactory>,
record_factories: HashMap<String, super::RecordFactory>,
subroutine_registry: HashMap<String, Arc<SubroutineFn>>,
acf: Option<access_security::AccessSecurityConfig>,
autosave_config: Option<autosave::SaveSetConfig>,
autosave_startup: Option<Arc<Mutex<AutosaveStartupConfig>>>,
startup_commands: Vec<CommandDef>,
shell_commands: Vec<CommandDef>,
startup_script: Option<String>,
inline_records: Vec<(String, Box<dyn Record>)>,
after_init_hooks: Vec<Box<dyn FnOnce() + Send>>,
link_set_installers: Vec<LinkSetInstaller>,
}
impl IocApplication {
pub fn new() -> Self {
let device_factories: HashMap<String, DeviceSupportFactory> = HashMap::new();
Self {
port: cas_server_port(),
tcp_port: None,
device_factories,
dynamic_device_factory: Some(Box::new(
crate::server::builtin_devices::builtin_dynamic_factory,
)),
record_factories: HashMap::new(),
subroutine_registry: HashMap::new(),
acf: None,
autosave_config: None,
autosave_startup: None,
startup_commands: Vec::new(),
shell_commands: Vec::new(),
startup_script: None,
inline_records: Vec::new(),
after_init_hooks: Vec::new(),
link_set_installers: Vec::new(),
}
}
pub fn port(mut self, port: u16) -> Self {
self.port = port;
self
}
pub fn tcp_port(mut self, port: u16) -> Self {
self.tcp_port = Some(port);
self
}
pub fn register_device_support<F>(mut self, dtyp: &str, factory: F) -> Self
where
F: Fn() -> Box<dyn DeviceSupport> + Send + Sync + 'static,
{
self.device_factories
.insert(dtyp.to_string(), Box::new(factory));
self
}
pub fn register_dynamic_device_support<F>(mut self, factory: F) -> Self
where
F: Fn(&DeviceSupportContext) -> Option<Box<dyn DeviceSupport>> + Send + Sync + 'static,
{
if let Some(existing) = self.dynamic_device_factory.take() {
self.dynamic_device_factory = Some(Box::new(move |ctx: &DeviceSupportContext| {
factory(ctx).or_else(|| existing(ctx))
}));
} else {
self.dynamic_device_factory = Some(Box::new(factory));
}
self
}
pub fn register_startup_command(mut self, cmd: CommandDef) -> Self {
self.startup_commands.push(cmd);
self
}
pub fn register_shell_command(mut self, cmd: CommandDef) -> Self {
self.shell_commands.push(cmd);
self
}
pub fn startup_commands(&self) -> &[CommandDef] {
&self.startup_commands
}
pub fn register_after_init(mut self, hook: impl FnOnce() + Send + 'static) -> Self {
self.after_init_hooks.push(Box::new(hook));
self
}
pub fn register_link_set_installer<F, Fut>(mut self, installer: F) -> Self
where
F: FnOnce(Arc<PvDatabase>) -> Fut + Send + 'static,
Fut: std::future::Future<Output = Vec<CommandDef>> + Send + 'static,
{
self.link_set_installers
.push(Box::new(move |db| Box::pin(installer(db))));
self
}
pub fn startup_script(mut self, path: &str) -> Self {
self.startup_script = Some(path.to_string());
self
}
pub fn register_record_type<F>(mut self, type_name: &str, factory: F) -> Self
where
F: Fn() -> Box<dyn Record> + Send + Sync + 'static,
{
self.record_factories
.insert(type_name.to_string(), Box::new(factory));
self
}
pub fn register_subroutine<F>(mut self, name: &str, func: F) -> Self
where
F: Fn(&mut dyn Record) -> CaResult<i64> + Send + Sync + 'static,
{
self.subroutine_registry
.insert(name.to_string(), Arc::new(Box::new(func)));
self
}
pub fn autosave(mut self, config: autosave::SaveSetConfig) -> Self {
self.autosave_config = Some(config);
self
}
pub fn autosave_startup(mut self, config: Arc<Mutex<AutosaveStartupConfig>>) -> Self {
self.autosave_startup = Some(config);
self
}
pub fn acf(mut self, config: access_security::AccessSecurityConfig) -> Self {
self.acf = Some(config);
self
}
pub fn record(mut self, name: &str, record: impl Record) -> Self {
self.inline_records
.push((name.to_string(), Box::new(record)));
self
}
pub fn record_boxed(mut self, name: &str, record: Box<dyn Record>) -> Self {
self.inline_records.push((name.to_string(), record));
self
}
pub async fn run<F, Fut>(self, protocol_runner: F) -> CaResult<()>
where
F: FnOnce(IocRunConfig) -> Fut + Send + 'static,
Fut: std::future::Future<Output = CaResult<()>> + Send,
{
let db = Arc::new(PvDatabase::new());
db.begin_load()
.expect("a database created a line ago has not run iocInit");
#[cfg(target_os = "rtems")]
crate::runtime::task::background_init();
let bridge = crate::runtime::task::BlockingBridge::capture();
let Self {
port,
tcp_port,
device_factories,
dynamic_device_factory,
record_factories,
subroutine_registry,
acf,
autosave_config,
autosave_startup,
mut startup_commands,
mut shell_commands,
startup_script,
inline_records,
after_init_hooks,
link_set_installers,
} = self;
for (name, factory) in record_factories {
super::db_loader::register_record_type(&name, factory);
}
if let Some(ref config) = autosave_startup {
let cmds = AutosaveStartupConfig::register_startup_commands(config.clone());
startup_commands.extend(cmds);
}
startup_commands.push(db_load_group_startup_command());
for (name, record) in inline_records {
db.add_record(&name, record).await?;
}
if let Some(script) = startup_script {
let db1 = db.clone();
let b1 = bridge.clone();
let (tx, rx) = crate::runtime::sync::oneshot::channel();
std::thread::Builder::new()
.name("iocsh-startup".into())
.stack_size(crate::runtime::task::StackSizeClass::Big.bytes())
.spawn(move || {
let _ = crate::runtime::task::enter_ioc_thread(
crate::runtime::task::ThreadPriority::Iocsh,
);
let shell = iocsh::IocShell::new(db1, b1);
for cmd in startup_commands {
shell.register(cmd);
}
let result = shell.execute_script(&script);
let _ = tx.send(result);
})
.expect("failed to spawn startup thread");
let result = rx
.await
.map_err(|_| CaError::InvalidValue("startup thread dropped".into()))?;
result.map_err(|e| CaError::InvalidValue(e))?;
}
let (pass0_files, pass1_files, builder_opt) = if let Some(ref config) = autosave_startup {
let cfg = config.lock().unwrap();
let pass0: Vec<std::path::PathBuf> = cfg
.pass0_restores
.iter()
.map(|r| cfg.resolve_save_file(&r.filename))
.collect();
let pass1: Vec<std::path::PathBuf> = cfg
.pass1_restores
.iter()
.map(|r| cfg.resolve_save_file(&r.filename))
.collect();
let builder = if !cfg.monitor_sets.is_empty() || !cfg.triggered_sets.is_empty() {
Some(cfg.into_builder())
} else {
None
};
(pass0, pass1, builder)
} else {
(Vec::new(), Vec::new(), None)
};
type AsyncHook = Box<
dyn FnOnce()
-> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send + 'static>>
+ Send
+ 'static,
>;
let mut lifecycle_hooks: Vec<(InitHookState, AsyncHook)> = Vec::new();
{
let db_p0 = db.clone();
let files = pass0_files.clone();
lifecycle_hooks.push((
InitHookState::AfterInitDevSup,
Box::new(move || {
Box::pin(async move {
for sav_path in &files {
match autosave::restore_from_file(&db_p0, sav_path).await {
Ok(count) if count > 0 => {
eprintln!(
"pass0 restore: {count} PVs from {}",
sav_path.display()
);
}
Err(e) => {
eprintln!(
"pass0 restore warning: {} - {e}",
sav_path.display()
);
}
_ => {}
}
}
})
}),
));
}
{
let db_p1 = db.clone();
let files = pass1_files.clone();
let cfg_path = autosave_config.as_ref().map(|c| c.save_path.clone());
lifecycle_hooks.push((
InitHookState::AfterInitDatabase,
Box::new(move || {
Box::pin(async move {
for sav_path in &files {
match autosave::restore_from_file(&db_p1, sav_path).await {
Ok(count) if count > 0 => {
eprintln!(
"pass1 restore: {count} PVs from {}",
sav_path.display()
);
}
Err(e) => {
eprintln!(
"pass1 restore warning: {} - {e}",
sav_path.display()
);
}
_ => {}
}
}
if let Some(path) = cfg_path {
match autosave::restore_from_file(&db_p1, &path).await {
Ok(count) if count > 0 => {
eprintln!("autosave: restored {count} PVs");
}
Err(e) => {
eprintln!("autosave restore warning: {} - {e}", path.display());
}
_ => {}
}
}
})
}),
));
}
macro_rules! announce {
($state:expr) => {{
let state = $state;
init_hook_announce(state);
let mut i = 0;
while i < lifecycle_hooks.len() {
if lifecycle_hooks[i].0 == state {
let (_, hook) = lifecycle_hooks.remove(i);
hook().await;
} else {
i += 1;
}
}
}};
}
announce!(InitHookState::AtIocBuild);
announce!(InitHookState::AtBeginning);
announce!(InitHookState::AfterCallbackInit);
announce!(InitHookState::AfterCaLinkInit);
for installer in link_set_installers {
shell_commands.extend(installer(db.clone()).await);
}
announce!(InitHookState::AfterInitDrvSup);
announce!(InitHookState::AfterInitRecSup);
announce!(InitHookState::AfterInitDevSup);
let record_count =
wire_device_support(&db, &device_factories, &dynamic_device_factory).await?;
db.install_subroutine_registry(subroutine_registry.clone())
.await;
wire_subroutines(&db, &subroutine_registry).await;
let io_intr_count = setup_io_intr(db.clone()).await;
setup_property_posts(db.clone()).await;
db.initialize_link_locality().await;
db.setup_cp_links().await;
db.setup_external_link_opens().await;
let link_wait_secs = crate::runtime::env::get("EPICS_RS_INIT_LINK_TIMEOUT")
.and_then(|s| s.parse::<f64>().ok())
.unwrap_or(10.0)
.max(0.0);
if link_wait_secs > 0.0 {
let (connected, total) = db
.wait_for_external_links(std::time::Duration::from_secs_f64(link_wait_secs))
.await;
if total > 0 {
if connected == total {
eprintln!("iocInit: {connected}/{total} external links connected");
} else {
let unconnected = db.unconnected_external_links().await;
eprintln!(
"iocInit: {connected}/{total} external links connected after \
{link_wait_secs}s — proceeding without: {}",
unconnected.join(", ")
);
}
}
}
db.ioc_init().await;
announce!(InitHookState::AfterInitDatabase);
announce!(InitHookState::AfterFinishDevSup);
announce!(InitHookState::AfterScanInit);
{
db.pini_process(crate::server::record::PiniMode::Yes).await;
db.mark_pini_done();
}
announce!(InitHookState::AfterInitialProcess);
let autosave_manager = if let Some(builder) = builder_opt {
match builder.build().await {
Ok(mgr) => {
eprintln!("autosave: {} save set(s) configured", mgr.set_names().len());
Some(Arc::new(mgr))
}
Err(e) => {
eprintln!("autosave: failed to build manager: {e}");
None
}
}
} else {
None
};
let total_records = db.all_record_names().await.len();
eprintln!(
"iocInit: {total_records} records, {record_count} with device support, {io_intr_count} I/O Intr"
);
announce!(InitHookState::AfterCaServerInit);
announce!(InitHookState::AfterIocBuilt);
announce!(InitHookState::AtIocRun);
db.pini_process(crate::server::record::PiniMode::Run).await;
let _scan_owner = crate::server::scan::ScanOwner::start(db.clone());
announce!(InitHookState::AfterDatabaseRunning);
announce!(InitHookState::AfterCaServerRunning);
for hook in after_init_hooks {
hook();
}
announce!(InitHookState::AfterIocRunning);
db.pini_process(crate::server::record::PiniMode::Running)
.await;
let pending = db.take_after_ioc_running();
if !pending.is_empty() {
let db1 = db.clone();
let b1 = bridge.clone();
let shell_cmds_clone = shell_commands.clone();
let (tx, rx) = crate::runtime::sync::oneshot::channel();
std::thread::Builder::new()
.name("iocsh-after-ioc-running".into())
.stack_size(crate::runtime::task::StackSizeClass::Big.bytes())
.spawn(move || {
let _ = crate::runtime::task::enter_ioc_thread(
crate::runtime::task::ThreadPriority::Iocsh,
);
let shell = iocsh::IocShell::new(db1, b1);
for cmd in shell_cmds_clone {
shell.register(cmd);
}
let mut errs: Vec<String> = Vec::new();
for line in pending {
if let Err(e) = shell.execute_line(&line) {
errs.push(format!("{line}: {e}"));
}
}
let _ = tx.send(errs);
})
.expect("failed to spawn afterIocRunning thread");
if let Ok(errs) = rx.await {
for e in errs {
eprintln!("afterIocRunning: {e}");
}
}
}
let config = IocRunConfig {
db,
port,
tcp_port,
acf,
autosave_config,
autosave_manager,
shell_commands,
after_init_hooks: Vec::new(),
};
let runner_fut = protocol_runner(config);
tokio::pin!(runner_fut);
#[cfg(not(target_os = "rtems"))]
let ctrl_c = async {
let _ = tokio::signal::ctrl_c().await;
};
#[cfg(target_os = "rtems")]
let ctrl_c = std::future::pending::<()>();
#[cfg(all(unix, not(target_os = "rtems")))]
let sigterm = async {
if let Ok(mut sig) =
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
{
let _ = sig.recv().await;
} else {
std::future::pending::<()>().await;
}
};
#[cfg(not(all(unix, not(target_os = "rtems"))))]
let sigterm = std::future::pending::<()>();
tokio::select! {
biased;
res = &mut runner_fut => res,
_ = ctrl_c => {
tracing::info!(target: "epics_base_rs::ioc_app", "SIGINT received, shutting down IOC");
Ok(())
}
_ = sigterm => {
tracing::info!(target: "epics_base_rs::ioc_app", "SIGTERM received, shutting down IOC");
Ok(())
}
}
}
}
pub(crate) async fn wire_device_support(
db: &PvDatabase,
factories: &HashMap<String, DeviceSupportFactory>,
dynamic_factory: &Option<DynamicDeviceSupportFactory>,
) -> CaResult<usize> {
let names = db.all_record_names().await;
let mut count = 0;
for name in names {
if let Some(rec_arc) = db.get_record(&name) {
let mut instance = rec_arc.write();
let dtyp = instance.common.dtyp.clone();
if !crate::server::device_support::is_soft_dtyp(&dtyp) {
let ctx = DeviceSupportContext {
dtyp: &dtyp,
inp: &instance.common.inp,
out: &instance.common.out,
};
let dev_opt = if let Some(factory) = factories.get(&dtyp) {
Some(factory())
} else if let Some(dyn_factory) = dynamic_factory {
dyn_factory(&ctx)
} else {
None
};
if let Some(dev) = dev_opt {
crate::server::device_support::wire_device_to_record(&mut instance, dev);
count += 1;
} else {
eprintln!(
"warning: no device support registered for DTYP '{dtyp}' (record: {name})"
);
}
}
}
}
Ok(count)
}
async fn wire_subroutines(db: &PvDatabase, registry: &HashMap<String, Arc<SubroutineFn>>) {
if registry.is_empty() {
return;
}
let names = db.all_record_names().await;
for name in names {
if let Some(rec_arc) = db.get_record(&name) {
let mut instance = rec_arc.write();
let rt = instance.record.record_type();
if rt == "sub" || rt == "aSub" {
if let Some(crate::types::EpicsValue::String(inam)) =
instance.record.get_field("INAM")
{
let inam = inam.as_str_lossy();
if !inam.is_empty() {
match registry.get(inam.as_ref()) {
Some(init_fn) => {
let init_fn = init_fn.clone();
if let Err(e) = init_fn(&mut *instance.record) {
eprintln!(
"iocInit: {name}.INAM '{inam}' init routine failed: {e}"
);
}
}
None => eprintln!("iocInit: {name}.INAM function '{inam}' not found"),
}
}
}
if let Some(crate::types::EpicsValue::String(snam)) =
instance.record.get_field("SNAM")
{
if let Some(sub_fn) = registry.get(snam.as_str_lossy().as_ref()) {
instance.subroutine = Some(sub_fn.clone());
}
}
}
}
}
}
async fn demote_io_intr_to_passive(db: &PvDatabase, name: &str, reason: &str) {
let Some(rec_arc) = db.get_record(name) else {
return;
};
let result = {
let mut inst = rec_arc.write();
if inst.common.scan != record::ScanType::IoIntr {
return;
}
inst.set_scan(record::ScanType::Passive)
};
if let record::CommonFieldPutResult::ScanChanged {
old_scan,
new_scan,
phas,
} = result
{
db.update_scan_index(name, old_scan, new_scan, phas, phas);
}
eprintln!("scanAdd: I/O Intr not valid ({reason}), {name} set to Passive");
}
pub(crate) async fn setup_io_intr(db: Arc<PvDatabase>) -> usize {
let all_names = db.all_record_names().await;
let io_intr_recs: Vec<(String, Arc<parking_lot::RwLock<record::RecordInstance>>)> = {
let mut recs = Vec::new();
for name in &all_names {
if let Some(arc) = db.get_record(name) {
recs.push((name.clone(), arc));
}
}
recs
};
let mut count = 0;
let mut demote: Vec<(String, &'static str)> = Vec::new();
for (name, rec_arc) in io_intr_recs {
let mut inst = rec_arc.write();
let independent = inst
.device
.as_ref()
.is_some_and(|d| d.io_intr_scan_independent());
let on_io_intr = inst.common.scan == record::ScanType::IoIntr;
if !on_io_intr && !independent {
continue;
}
let Some(mut dev) = inst.device.take() else {
if on_io_intr {
demote.push((name, "no DSET"));
}
continue;
};
if let Some(mut intr_rx) = dev.io_intr_receiver() {
let db_clone = db.clone();
let rec_name = name.clone();
let rec_arc_clone = rec_arc.clone();
crate::runtime::task::spawn(async move {
while intr_rx.recv().await.is_some() {
let process = independent || {
let inst = rec_arc_clone.read();
inst.common.scan == record::ScanType::IoIntr
};
if !process {
continue;
}
let mut visited = std::collections::HashSet::new();
let _ = db_clone
.process_record_readback(&rec_name, &mut visited, 0)
.await;
}
});
count += 1;
} else if on_io_intr {
demote.push((name, "no interrupt source from device support"));
}
inst.device = Some(dev);
}
for (name, reason) in demote {
demote_io_intr_to_passive(&db, &name, reason).await;
}
count
}
pub(crate) async fn setup_property_posts(db: Arc<PvDatabase>) -> usize {
let names = db.all_record_names().await;
let mut count = 0;
for name in names {
if let Some(rec_arc) = db.get_record(&name) {
let mut inst = rec_arc.write();
if let Some(mut dev) = inst.device.take() {
if let Some(mut rx) = dev.property_post_receiver() {
let db_clone = db.clone();
let rec_name = name.clone();
crate::runtime::task::spawn(async move {
while let Some(fields) = rx.recv().await {
let _ = db_clone.post_property_fields(&rec_name, fields);
}
});
count += 1;
}
inst.device = Some(dev);
}
}
}
count
}
#[cfg(test)]
mod io_intr_scan_add_tests {
use super::setup_io_intr;
use crate::server::database::PvDatabase;
use crate::server::record::ScanType;
use crate::server::records::ai::AiRecord;
use std::sync::Arc;
#[epics_macros_rs::epics_test]
async fn io_intr_without_device_support_is_demoted_to_passive() {
let db = Arc::new(PvDatabase::new());
db.add_record("NODEV", Box::new(AiRecord::new(0.0)))
.await
.unwrap();
{
let rec = db.get_record("NODEV").unwrap();
let mut inst = rec.write();
inst.common.scan = ScanType::IoIntr;
}
db.update_scan_index("NODEV", ScanType::Passive, ScanType::IoIntr, 0, 0);
assert_eq!(
db.records_for_scan(ScanType::IoIntr).await,
vec!["NODEV".to_string()],
"precondition: the record starts in the I/O Intr bucket"
);
let wired = setup_io_intr(db.clone()).await;
assert_eq!(wired, 0, "no device support ⇒ nothing to wire");
let rec = db.get_record("NODEV").unwrap();
assert_eq!(
rec.read().common.scan,
ScanType::Passive,
"an unusable I/O Intr record must be demoted to Passive"
);
assert!(
db.records_for_scan(ScanType::IoIntr).await.is_empty(),
"and must leave the I/O Intr scan list"
);
}
#[epics_macros_rs::epics_test]
async fn io_intr_with_device_but_no_interrupt_source_is_demoted_to_passive() {
use crate::error::CaResult;
use crate::server::device_support::DeviceSupport;
use crate::server::record::Record;
struct NoIntrDevice;
impl DeviceSupport for NoIntrDevice {
fn write(&mut self, _record: &mut dyn Record) -> CaResult<()> {
Ok(())
}
fn dtyp(&self) -> &str {
"NoIntr"
}
}
let db = Arc::new(PvDatabase::new());
db.add_record("NOINTR", Box::new(AiRecord::new(0.0)))
.await
.unwrap();
{
let rec = db.get_record("NOINTR").unwrap();
let mut inst = rec.write();
inst.common.scan = ScanType::IoIntr;
inst.device = Some(Box::new(NoIntrDevice));
}
db.update_scan_index("NOINTR", ScanType::Passive, ScanType::IoIntr, 0, 0);
let wired = setup_io_intr(db.clone()).await;
assert_eq!(wired, 0, "no interrupt source ⇒ nothing to wire");
let rec = db.get_record("NOINTR").unwrap();
{
let inst = rec.read();
assert_eq!(
inst.common.scan,
ScanType::Passive,
"device support with no interrupt source must demote SCAN to Passive"
);
assert!(
inst.device.is_some(),
"the demotion must not drop the record's device support"
);
}
assert!(db.records_for_scan(ScanType::IoIntr).await.is_empty());
}
#[epics_macros_rs::epics_test]
async fn a_passive_record_is_not_touched_by_the_io_intr_pass() {
let db = Arc::new(PvDatabase::new());
db.add_record("PASV", Box::new(AiRecord::new(0.0)))
.await
.unwrap();
let wired = setup_io_intr(db.clone()).await;
assert_eq!(wired, 0);
let rec = db.get_record("PASV").unwrap();
assert_eq!(rec.read().common.scan, ScanType::Passive);
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Mutex as StdMutex;
use std::sync::atomic::{AtomicUsize, Ordering};
static INIT_HOOK_TEST_LOCK: StdMutex<()> = StdMutex::new(());
fn production_scope(src: &str) -> &str {
match src.find("\n#[cfg(test)]") {
Some(i) => &src[..i],
None => src,
}
}
#[test]
fn iocsh_threads_take_the_iocsh_band() {
let prod = production_scope(include_str!("ioc_app.rs"));
assert_eq!(
prod.matches("enter_ioc_thread(").count(),
2,
"the startup-script thread and the afterIocRunning thread"
);
assert_eq!(
prod.matches("name_current_thread(").count(),
0,
"naming without banding leaves the thread one level above idle on \
the target; `enter_ioc_thread` is the whole prologue"
);
assert_eq!(
prod.matches("apply_to_current_thread(").count(),
0,
"banding without naming leaves an RTEMS-anonymous thread"
);
for name in ["iocsh-startup", "iocsh-after-ioc-running"] {
let at = prod
.find(&format!(".name(\"{name}\""))
.unwrap_or_else(|| panic!("the {name} thread moved; update this guard"));
let head = &prod[at..(at + 700).min(prod.len())];
assert!(
head.contains("enter_ioc_thread(") && head.contains("ThreadPriority::Iocsh"),
"{name} must enter its IOC thread role at `ThreadPriority::Iocsh` \
(rtems_init.c:1002)"
);
}
}
#[test]
fn the_iocsh_band_is_epics_thread_priority_iocsh() {
assert_eq!(crate::runtime::task::ThreadPriority::Iocsh.value(), 91);
}
#[test]
fn dbloadgroup_startup_command_queues_and_removes() {
let _ = take_group_load_requests();
let rt = tokio::runtime::Runtime::new().unwrap();
let db = Arc::new(PvDatabase::new());
let bridge = {
let _guard = rt.enter();
crate::runtime::task::BlockingBridge::capture()
};
let shell = iocsh::IocShell::new(db, bridge);
shell.register(db_load_group_startup_command());
let a = std::env::temp_dir().join("qsrv_q_a.json");
let b = std::env::temp_dir().join("qsrv_q_b.json");
std::fs::write(&a, "{}").unwrap();
std::fs::write(&b, "{}").unwrap();
shell
.execute_line(&format!("dbLoadGroup(\"{}\")", a.display()))
.unwrap();
shell
.execute_line(&format!("dbLoadGroup(\"{}\",\"M=1\")", b.display()))
.unwrap();
shell
.execute_line(&format!("dbLoadGroup(\"{}\")", a.display()))
.unwrap();
assert!(
shell
.execute_line("dbLoadGroup(\"/no/such/group.json\")")
.is_err(),
"a missing group file must error at command time"
);
shell
.execute_line(&format!("dbLoadGroup(\"-{}\")", a.display()))
.unwrap();
let reqs = take_group_load_requests();
assert_eq!(reqs.len(), 1, "only the (b, M=1) entry must remain");
assert_eq!(reqs[0].filename, b.to_string_lossy());
assert_eq!(reqs[0].macros, "M=1");
shell
.execute_line(&format!("dbLoadGroup(\"{}\")", b.display()))
.unwrap();
shell.execute_line("dbLoadGroup(\"-*\")").unwrap();
assert!(
take_group_load_requests().is_empty(),
"dbLoadGroup(\"-*\") must clear the queue"
);
let _ = std::fs::remove_file(&a);
let _ = std::fs::remove_file(&b);
}
#[test]
fn init_hook_register_and_announce_in_order() {
let _guard = INIT_HOOK_TEST_LOCK.lock().unwrap();
init_hooks::init_hook_free();
let seen: Arc<StdMutex<Vec<InitHookState>>> = Arc::new(StdMutex::new(Vec::new()));
let seen_cb = seen.clone();
init_hook_register(Arc::new(move |state| {
seen_cb.lock().unwrap().push(state);
}));
let order = [
InitHookState::AtIocBuild,
InitHookState::AfterInitDevSup,
InitHookState::AfterInitDatabase,
InitHookState::AfterInitialProcess,
InitHookState::AfterIocRunning,
];
for &s in &order {
init_hook_announce(s);
}
let got = seen.lock().unwrap().clone();
assert_eq!(got, order, "hooks must fire in announce order");
init_hooks::init_hook_free();
}
#[test]
fn init_hook_reentrant_register_does_not_deadlock() {
let _guard = INIT_HOOK_TEST_LOCK.lock().unwrap();
init_hooks::init_hook_free();
let inner_calls = Arc::new(AtomicUsize::new(0));
let inner_for_outer = inner_calls.clone();
init_hook_register(Arc::new(move |_state| {
let inner = inner_for_outer.clone();
init_hook_register(Arc::new(move |_s| {
inner.fetch_add(1, Ordering::SeqCst);
}));
}));
init_hook_announce(InitHookState::AtIocBuild);
assert_eq!(inner_calls.load(Ordering::SeqCst), 0);
init_hook_announce(InitHookState::AfterIocRunning);
assert!(inner_calls.load(Ordering::SeqCst) >= 1);
init_hooks::init_hook_free();
}
#[test]
fn init_hook_state_names_match_c() {
assert_eq!(InitHookState::AtIocBuild.name(), "initHookAtIocBuild");
assert_eq!(
InitHookState::AfterInitDevSup.name(),
"initHookAfterInitDevSup"
);
assert_eq!(
InitHookState::AfterInitDatabase.name(),
"initHookAfterInitDatabase"
);
assert_eq!(
InitHookState::AfterIocRunning.name(),
"initHookAfterIocRunning"
);
}
#[epics_macros_rs::epics_test]
async fn test_ioc_application_empty() {
let db = Arc::new(PvDatabase::new());
let factories = HashMap::new();
let count = wire_device_support(&db, &factories, &None).await.unwrap();
assert_eq!(count, 0);
}
#[epics_macros_rs::epics_test]
async fn test_wire_device_support_no_dtyp() {
use crate::server::records::ai::AiRecord;
let db = Arc::new(PvDatabase::new());
db.add_record("TEST", Box::new(AiRecord::new(0.0)))
.await
.unwrap();
let factories = HashMap::new();
let count = wire_device_support(&db, &factories, &None).await.unwrap();
assert_eq!(count, 0); }
#[epics_macros_rs::epics_test]
async fn wire_device_support_forwards_info_tags_to_driver() {
use crate::server::device_support::{DeviceReadOutcome, DeviceSupport};
use crate::server::record::ScanType;
use crate::server::records::ai::AiRecord;
use std::sync::{Arc as StdArc, Mutex as StdMutex};
struct RecordingDev {
seen: StdArc<StdMutex<HashMap<String, String>>>,
}
impl DeviceSupport for RecordingDev {
fn write(&mut self, _record: &mut dyn crate::server::record::Record) -> CaResult<()> {
Ok(())
}
fn dtyp(&self) -> &str {
"TestRecording"
}
fn read(
&mut self,
_record: &mut dyn crate::server::record::Record,
) -> CaResult<DeviceReadOutcome> {
Ok(DeviceReadOutcome::ok())
}
fn apply_record_info(&mut self, info: &HashMap<String, String>) {
let mut g = self.seen.lock().unwrap();
*g = info.clone();
}
fn set_record_info(&mut self, _name: &str, _scan: ScanType) {}
}
let seen = StdArc::new(StdMutex::new(HashMap::<String, String>::new()));
let seen_factory = seen.clone();
let mut factories: HashMap<String, DeviceSupportFactory> = HashMap::new();
factories.insert(
"TestRecording".to_string(),
Box::new(move || {
Box::new(RecordingDev {
seen: seen_factory.clone(),
})
}),
);
let db = Arc::new(PvDatabase::new());
db.add_record("AI:WITH:INFO", Box::new(AiRecord::new(0.0)))
.await
.unwrap();
let rec = db.get_record("AI:WITH:INFO").unwrap();
{
let mut inst = rec.write();
inst.common.dtyp = "TestRecording".to_string();
inst.set_info("asyn:READBACK", "1");
inst.set_info("Q:group", "demo");
}
let count = wire_device_support(&db, &factories, &None).await.unwrap();
assert_eq!(count, 1, "device support must have attached");
let observed = seen.lock().unwrap().clone();
assert_eq!(observed.get("asyn:READBACK").map(String::as_str), Some("1"));
assert_eq!(observed.get("Q:group").map(String::as_str), Some("demo"));
}
#[epics_macros_rs::epics_test]
async fn wire_device_support_binds_in_database_load_order() {
use crate::server::device_support::{DeviceReadOutcome, DeviceSupport};
use crate::server::records::ai::AiRecord;
use std::sync::{Arc as StdArc, Mutex as StdMutex};
struct NoopDev;
impl DeviceSupport for NoopDev {
fn write(&mut self, _record: &mut dyn crate::server::record::Record) -> CaResult<()> {
Ok(())
}
fn dtyp(&self) -> &str {
"SeqDev"
}
fn read(
&mut self,
_record: &mut dyn crate::server::record::Record,
) -> CaResult<DeviceReadOutcome> {
Ok(DeviceReadOutcome::ok())
}
}
let names: Vec<String> = (0..24)
.map(|i: usize| format!("LOAD:{:02}", (i * 7 + 3) % 24))
.collect();
let db = Arc::new(PvDatabase::new());
for name in &names {
db.add_record(name, Box::new(AiRecord::new(0.0)))
.await
.unwrap();
let rec = db.get_record(name).unwrap();
let mut inst = rec.write();
inst.common.dtyp = "SeqDev".to_string();
inst.common.inp = format!("@{name}");
}
let wired: StdArc<StdMutex<Vec<String>>> = StdArc::new(StdMutex::new(Vec::new()));
let captured = wired.clone();
let dynamic: Option<DynamicDeviceSupportFactory> =
Some(Box::new(move |ctx: &DeviceSupportContext| {
captured
.lock()
.unwrap()
.push(ctx.inp.trim_start_matches('@').to_string());
Some(Box::new(NoopDev) as Box<dyn DeviceSupport>)
}));
let factories: HashMap<String, DeviceSupportFactory> = HashMap::new();
let count = wire_device_support(&db, &factories, &dynamic)
.await
.unwrap();
assert_eq!(count, names.len());
let wired = std::mem::take(&mut *wired.lock().unwrap());
assert_eq!(
wired, names,
"device support must bind in database load order (C initDevSup), \
not HashMap hash order"
);
}
#[epics_macros_rs::epics_test]
async fn readback_output_cycle_reads_back_and_skips_device_write() {
use crate::server::device_support::{DeviceReadOutcome, DeviceSupport};
use crate::server::record::ScanType;
use crate::server::records::bo::BoRecord;
use crate::types::EpicsValue;
use std::sync::Arc as StdArc;
use std::sync::atomic::{AtomicUsize, Ordering};
struct ReadbackDev {
writes: StdArc<AtomicUsize>,
readback_val: u16,
}
impl DeviceSupport for ReadbackDev {
fn dtyp(&self) -> &str {
"TestReadback"
}
fn io_intr_scan_independent(&self) -> bool {
true
}
fn output_callback_readback(&self) -> bool {
true
}
fn read(
&mut self,
record: &mut dyn crate::server::record::Record,
) -> CaResult<DeviceReadOutcome> {
record.set_val(EpicsValue::Enum(self.readback_val))?;
Ok(DeviceReadOutcome::computed())
}
fn write(&mut self, _record: &mut dyn crate::server::record::Record) -> CaResult<()> {
self.writes.fetch_add(1, Ordering::SeqCst);
Ok(())
}
fn set_record_info(&mut self, _name: &str, _scan: ScanType) {}
}
let writes = StdArc::new(AtomicUsize::new(0));
let db = Arc::new(PvDatabase::new());
db.add_record("BO:RBK", Box::new(BoRecord::new(1)))
.await
.unwrap();
{
let rec = db.get_record("BO:RBK").unwrap();
let mut inst = rec.write();
inst.common.dtyp = "TestReadback".to_string();
inst.device = Some(Box::new(ReadbackDev {
writes: writes.clone(),
readback_val: 0,
}));
}
{
let mut visited = std::collections::HashSet::new();
db.process_record_readback("BO:RBK", &mut visited, 0)
.await
.unwrap();
}
{
let rec = db.get_record("BO:RBK").unwrap();
let inst = rec.read();
assert_eq!(
inst.record.get_field("VAL"),
Some(EpicsValue::Enum(0)),
"readback cycle must pull the driver callback value (0) into VAL"
);
}
assert_eq!(
writes.load(Ordering::SeqCst),
0,
"readback cycle must NOT write VAL back to the driver (no re-trigger)"
);
{
let rec = db.get_record("BO:RBK").unwrap();
let mut inst = rec.write();
inst.record.put_field("VAL", EpicsValue::Enum(1)).unwrap();
}
{
let mut visited = std::collections::HashSet::new();
db.process_record_with_links("BO:RBK", &mut visited, 0)
.await
.unwrap();
}
assert_eq!(
writes.load(Ordering::SeqCst),
1,
"a put/scan cycle must write the setpoint to the driver exactly once"
);
}
}