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,
AtIocPause,
AfterCaServerPaused,
AfterDatabasePaused,
AfterIocPaused,
AtShutdown,
AfterStopScan,
AfterShutdown,
}
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",
InitHookState::AtIocPause => "initHookAtIocPause",
InitHookState::AfterCaServerPaused => "initHookAfterCaServerPaused",
InitHookState::AfterDatabasePaused => "initHookAfterDatabasePaused",
InitHookState::AfterIocPaused => "initHookAfterIocPaused",
InitHookState::AtShutdown => "initHookAtShutdown",
InitHookState::AfterStopScan => "initHookAfterStopScan",
InitHookState::AfterShutdown => "initHookAfterShutdown",
}
}
}
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(Debug, Clone, Copy, PartialEq, Eq)]
pub enum IocState {
Void,
Building,
Built,
Running,
Paused,
}
struct Lifecycle {
state: IocState,
scan: Option<crate::server::scan::ScanOwner>,
}
static LIFECYCLE: Mutex<Lifecycle> = Mutex::new(Lifecycle {
state: IocState::Void,
scan: None,
});
fn lifecycle() -> std::sync::MutexGuard<'static, Lifecycle> {
LIFECYCLE.lock().unwrap_or_else(|e| e.into_inner())
}
pub fn get_ioc_state() -> IocState {
lifecycle().state
}
fn set_ioc_state(state: IocState) {
lifecycle().state = state;
}
fn adopt_scan_owner(owner: crate::server::scan::ScanOwner) {
let previous = lifecycle().scan.replace(owner);
drop(previous);
}
pub fn ioc_run() -> i32 {
let from = get_ioc_state();
if from != IocState::Paused && from != IocState::Built {
crate::runtime::log::errlog_printf(&format!(
"iocRun: {} IOC not paused\n",
crate::runtime::log::erl_warning()
));
return -1;
}
init_hook_announce(InitHookState::AtIocRun);
crate::server::scan::scan_run();
init_hook_announce(InitHookState::AfterDatabaseRunning);
crate::server::db_server::db_run_servers();
init_hook_announce(InitHookState::AfterCaServerRunning);
crate::runtime::log::errlog_printf(if from == IocState::Built {
"iocRun: All initialization complete\n"
} else {
"iocRun: IOC restarted\n"
});
set_ioc_state(IocState::Running);
init_hook_announce(InitHookState::AfterIocRunning);
0
}
pub(crate) fn note_scan_owner_started() {
if get_ioc_state() == IocState::Void {
set_ioc_state(IocState::Built);
}
if get_ioc_state() == IocState::Built {
ioc_run();
} else {
crate::server::scan::scan_run();
}
}
pub fn ioc_pause() -> i32 {
if get_ioc_state() != IocState::Running {
crate::runtime::log::errlog_printf(&format!(
"iocPause: {} IOC not running\n",
crate::runtime::log::erl_warning()
));
return -1;
}
init_hook_announce(InitHookState::AtIocPause);
crate::server::db_server::db_pause_servers();
init_hook_announce(InitHookState::AfterCaServerPaused);
crate::server::scan::scan_pause();
init_hook_announce(InitHookState::AfterDatabasePaused);
set_ioc_state(IocState::Paused);
crate::runtime::log::errlog_printf("iocPause: IOC suspended\n");
init_hook_announce(InitHookState::AfterIocPaused);
0
}
pub fn ioc_shutdown() -> i32 {
if get_ioc_state() == IocState::Void {
return 0;
}
init_hook_announce(InitHookState::AtShutdown);
let owner = lifecycle().scan.take();
drop(owner);
crate::server::scan::scan_stop();
init_hook_announce(InitHookState::AfterStopScan);
crate::server::db_server::db_stop_servers();
set_ioc_state(IocState::Void);
init_hook_announce(InitHookState::AfterShutdown);
0
}
#[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,
},
ArgDesc {
name: "macros",
arg_type: ArgType::String,
},
],
"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: access_security::AcfCell,
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>>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct IocInitDecision {
run: bool,
interactive: bool,
}
impl IocInitDecision {
pub fn run(interactive: bool) -> Self {
Self {
run: true,
interactive,
}
}
pub fn skip(interactive: bool) -> Self {
Self {
run: false,
interactive,
}
}
}
#[derive(Debug)]
pub enum IocRunFailure {
StartupScript {
path: String,
reason: String,
},
StartupCommand {
line: String,
reason: String,
},
Startup(CaError),
Serving(CaError),
}
impl From<CaError> for IocRunFailure {
fn from(e: CaError) -> Self {
IocRunFailure::Startup(e)
}
}
impl From<IocRunFailure> for CaError {
fn from(failure: IocRunFailure) -> Self {
match failure {
IocRunFailure::StartupScript { reason, .. }
| IocRunFailure::StartupCommand { reason, .. } => CaError::InvalidValue(reason),
IocRunFailure::Startup(e) | IocRunFailure::Serving(e) => e,
}
}
}
impl std::fmt::Display for IocRunFailure {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
IocRunFailure::StartupScript { reason, .. }
| IocRunFailure::StartupCommand { reason, .. } => write!(f, "{reason}"),
IocRunFailure::Startup(e) | IocRunFailure::Serving(e) => write!(f, "{e}"),
}
}
}
fn as_init_failed_message(paints: bool) -> String {
let (error, magenta, reset) = if paints {
(crate::runtime::log::ERL_ERROR, "\x1b[35;1m", "\x1b[0m")
} else {
("ERROR", "", "")
};
format!("{error} iocBuild: asInit Failed.\n{magenta} The IOC has not been started.{reset}\n")
}
fn run_startup_phase(
shell: &iocsh::IocShell,
lines: &[String],
script: Option<&str>,
) -> Result<(), IocRunFailure> {
for line in lines {
shell
.execute_line_reported(line)
.map_err(|reason| IocRunFailure::StartupCommand {
line: line.clone(),
reason,
})?;
}
if let Some(script) = script {
shell
.execute_script_with_macros(script, &Default::default())
.map_err(|reason| IocRunFailure::StartupScript {
path: script.to_string(),
reason,
})?;
}
Ok(())
}
async fn run_uninitialized_tail(
db: Arc<PvDatabase>,
bridge: crate::runtime::task::BlockingBridge,
acf: access_security::AcfCell,
interactive: bool,
) -> Result<(), IocRunFailure> {
if !interactive {
std::future::pending::<()>().await;
unreachable!("a pending future never completes");
}
let (tx, rx) = crate::runtime::sync::oneshot::channel();
crate::runtime::task::MandatoryThread::new(
"iocsh",
crate::runtime::task::ThreadPriority::Iocsh,
crate::runtime::task::StackSizeClass::Big,
)
.try_spawn(move || {
crate::runtime::task::register_main_thread();
let shell = iocsh::IocShell::new_with_acf(db, bridge, acf);
let _ = tx.send(shell.run_repl());
})
.map_err(|e| CaError::InvalidValue(format!("could not start the iocsh thread: {e}")))?;
match rx.await {
Ok(Ok(())) => Ok(()),
Ok(Err(e)) => Err(IocRunFailure::Serving(CaError::InvalidValue(e))),
Err(_) => Err(IocRunFailure::Serving(CaError::InvalidValue(
"shell thread dropped".into(),
))),
}
}
type ProtocolRunner = Box<
dyn FnOnce(
IocRunConfig,
) -> std::pin::Pin<
Box<dyn std::future::Future<Output = CaResult<()>> + Send + 'static>,
> + Send
+ 'static,
>;
struct ProtocolStart {
bridge: crate::runtime::task::BlockingBridge,
port: u16,
tcp_port: Option<u16>,
acf: access_security::AcfCell,
autosave_config: Option<autosave::SaveSetConfig>,
runner: ProtocolRunner,
}
struct ProtocolServer {
handle: crate::runtime::task::TaskHandle<CaResult<()>>,
finished: Option<Result<(), IocRunFailure>>,
live: bool,
}
impl ProtocolServer {
async fn wait(&mut self) -> Result<(), IocRunFailure> {
if let Some(collected) = self.finished.take() {
return collected;
}
let joined = (&mut self.handle).await;
self.live = false;
Self::outcome(joined)
}
async fn await_serving(&mut self, generation: u64) {
tokio::select! {
biased;
joined = &mut self.handle => {
self.live = false;
self.finished = Some(Self::outcome(joined));
}
() = crate::server::db_server::serving_after(generation) => {}
}
}
fn outcome(
joined: Result<CaResult<()>, crate::runtime::task::TaskJoinError>,
) -> Result<(), IocRunFailure> {
match joined {
Ok(res) => res.map_err(IocRunFailure::Serving),
Err(e) => Err(IocRunFailure::Serving(CaError::InvalidValue(format!(
"protocol runner did not finish: {e}"
)))),
}
}
async fn shut_down(mut self) {
self.live = false;
self.handle.abort();
let _ = (&mut self.handle).await;
}
}
impl Drop for ProtocolServer {
fn drop(&mut self) {
if self.live {
self.handle.abort();
}
}
}
struct ArmedLifecycle;
impl Drop for ArmedLifecycle {
fn drop(&mut self) {
drop(take_lifecycle());
}
}
struct IocBuild {
db: Arc<PvDatabase>,
acf: access_security::AcfCell,
autosave_config: Option<autosave::SaveSetConfig>,
autosave_startup: Option<Arc<Mutex<AutosaveStartupConfig>>>,
link_set_installers: Vec<LinkSetInstaller>,
after_init_hooks: Vec<Box<dyn FnOnce() + Send>>,
protocol: ProtocolStart,
}
struct BuiltIoc {
db: Arc<PvDatabase>,
autosave_manager: Option<Arc<autosave::AutosaveManager>>,
after_init_hooks: Vec<Box<dyn FnOnce() + Send>>,
protocol: ProtocolStart,
}
struct RunningIoc {
server: ProtocolServer,
}
enum BuildOutcome {
Built(Box<BuiltIoc>),
AsInitFailed,
}
enum IocLifecycle {
Armed(Box<IocBuild>),
Built(Box<BuiltIoc>),
Running(RunningIoc),
AsInitFailed,
Failed(CaError),
}
static LIFECYCLE_OWNER: Mutex<Option<IocLifecycle>> = Mutex::new(None);
fn arm_build(build: IocBuild) {
*LIFECYCLE_OWNER.lock().unwrap() = Some(IocLifecycle::Armed(Box::new(build)));
}
fn take_lifecycle() -> Option<IocLifecycle> {
LIFECYCLE_OWNER.lock().unwrap().take()
}
fn put_lifecycle(state: IocLifecycle) {
*LIFECYCLE_OWNER.lock().unwrap() = Some(state);
}
pub(crate) enum ShellTransition {
Done,
Failed,
Refused,
NotOurs,
}
pub(crate) fn build_from_shell(bridge: &crate::runtime::task::BlockingBridge) -> ShellTransition {
match take_lifecycle() {
None => ShellTransition::NotOurs,
Some(IocLifecycle::Armed(build)) => match bridge.block_on(build.perform_build()) {
Ok(BuildOutcome::Built(built)) => {
put_lifecycle(IocLifecycle::Built(built));
ShellTransition::Done
}
Ok(BuildOutcome::AsInitFailed) => {
put_lifecycle(IocLifecycle::AsInitFailed);
ShellTransition::Failed
}
Err(e) => {
put_lifecycle(IocLifecycle::Failed(e));
ShellTransition::Failed
}
},
Some(other) => {
put_lifecycle(other);
ShellTransition::Refused
}
}
}
pub(crate) fn build_refusal() -> String {
format!(
"iocBuild: {} IOC can only be initialized from uninitialized or \
stopped state\n",
if crate::runtime::log::errlog_console_paints() {
crate::runtime::log::ERL_ERROR
} else {
"ERROR"
}
)
}
pub(crate) fn build_without_application(close_record_load: impl FnOnce()) -> bool {
if get_ioc_state() != IocState::Void {
crate::runtime::log::errlog_printf(&build_refusal());
return false;
}
crate::runtime::log::errlog_printf("Starting iocInit\n");
for line in crate::server::iocsh::misc_commands::core_release_block() {
println!("{line}");
}
set_ioc_state(IocState::Building);
close_record_load();
set_ioc_state(IocState::Built);
true
}
pub(crate) fn run_from_shell(bridge: &crate::runtime::task::BlockingBridge) -> ShellTransition {
match take_lifecycle() {
None => ShellTransition::NotOurs,
Some(IocLifecycle::Built(built)) => {
put_lifecycle(IocLifecycle::Running(bridge.block_on(built.run())));
ShellTransition::Done
}
Some(other) => {
put_lifecycle(other);
ShellTransition::Refused
}
}
}
impl IocBuild {
async fn perform_build(self) -> CaResult<BuildOutcome> {
let Self {
db,
acf,
autosave_config,
autosave_startup,
link_set_installers,
after_init_hooks,
protocol,
} = self;
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);
crate::runtime::log::errlog_printf("Starting iocInit\n");
announce!(InitHookState::AtBeginning);
for line in crate::server::iocsh::misc_commands::core_release_block() {
println!("{line}");
}
set_ioc_state(IocState::Building);
crate::server::scan::scan_pause();
crate::runtime::exit::at_exit("iocShutdown", || {
ioc_shutdown();
});
crate::runtime::task::background_init();
access_security::start_acf_watchers(&db, &acf);
announce!(InitHookState::AfterCallbackInit);
announce!(InitHookState::AfterCaLinkInit);
for installer in link_set_installers {
for cmd in installer(db.clone()).await {
iocsh::register_command(cmd);
}
}
announce!(InitHookState::AfterInitDrvSup);
announce!(InitHookState::AfterInitRecSup);
announce!(InitHookState::AfterInitDevSup);
db.drain_deferred_record_inits();
let record_count = db.records_with_device_support().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(crate::runtime::time::duration_from_secs(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);
let as_init = crate::server::iocsh::as_init(&acf);
if let Some(message) = as_init.message() {
println!("{message}");
}
if as_init.failed() {
crate::runtime::log::errlog_printf(&as_init_failed_message(
crate::runtime::log::errlog_console_paints(),
));
return Ok(BuildOutcome::AsInitFailed);
}
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 {
let mgr = builder.build().await;
eprintln!("autosave: {} save set(s) configured", mgr.set_names().len());
Some(Arc::new(mgr))
} else {
None
};
let total_records = db.all_record_names().await.len();
crate::runtime::log::errlog_printf(&format!(
"iocInit: {total_records} records, {record_count} with device support, {io_intr_count} I/O Intr\n"
));
announce!(InitHookState::AfterCaServerInit);
announce!(InitHookState::AfterIocBuilt);
set_ioc_state(IocState::Built);
Ok(BuildOutcome::Built(Box::new(BuiltIoc {
db,
autosave_manager,
after_init_hooks,
protocol,
})))
}
}
impl BuiltIoc {
async fn run(self) -> RunningIoc {
let Self {
db,
autosave_manager,
after_init_hooks,
protocol,
} = self;
db.pini_process(crate::server::record::PiniMode::Run).await;
adopt_scan_owner(crate::server::scan::ScanOwner::start(db.clone()));
db.pini_process(crate::server::record::PiniMode::Running)
.await;
for hook in after_init_hooks {
hook();
}
let ProtocolStart {
bridge,
port,
tcp_port,
acf,
autosave_config,
runner,
} = protocol;
let config = IocRunConfig {
db,
port,
tcp_port,
acf,
autosave_config,
autosave_manager,
shell_commands: Vec::new(),
after_init_hooks: Vec::new(),
};
let generation = crate::server::db_server::serving_generation();
let mut server = ProtocolServer {
handle: bridge.spawn(runner(config)),
finished: None,
live: true,
};
server.await_serving(generation).await;
RunningIoc { server }
}
}
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>>>,
commands: Vec<CommandDef>,
startup_script: Option<String>,
startup_lines: Vec<String>,
inline_pvs: Vec<(String, crate::types::EpicsValue)>,
inline_records: Vec<(String, Box<dyn Record>)>,
after_init_hooks: Vec<Box<dyn FnOnce() + Send>>,
link_set_installers: Vec<LinkSetInstaller>,
ioc_init_gate: Option<Box<dyn FnOnce() -> IocInitDecision + Send>>,
}
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,
commands: Vec::new(),
startup_script: None,
startup_lines: Vec::new(),
inline_pvs: Vec::new(),
inline_records: Vec::new(),
after_init_hooks: Vec::new(),
link_set_installers: Vec::new(),
ioc_init_gate: None,
}
}
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.commands.push(cmd);
self
}
pub fn register_shell_command(mut self, cmd: CommandDef) -> Self {
self.commands.push(cmd);
self
}
pub fn startup_commands(&self) -> &[CommandDef] {
&self.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 before_ioc_init(
mut self,
gate: impl FnOnce() -> IocInitDecision + Send + 'static,
) -> Self {
self.ioc_init_gate = Some(Box::new(gate));
self
}
pub fn startup_script(mut self, path: &str) -> Self {
self.startup_script = Some(path.to_string());
self
}
pub fn startup_line(mut self, line: &str) -> Self {
self.startup_lines.push(line.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,
{
let factory: super::RecordFactory = Box::new(factory);
super::db_loader::snapshot_declared_fields(type_name, &factory);
self.record_factories.insert(type_name.to_string(), 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 fn pv(mut self, name: &str, initial: crate::types::EpicsValue) -> Self {
self.inline_pvs.push((name.to_string(), initial));
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 + 'static,
{
self.run_phased(protocol_runner)
.await
.map_err(CaError::from)
}
pub async fn run_phased<F, Fut>(self, protocol_runner: F) -> Result<(), IocRunFailure>
where
F: FnOnce(IocRunConfig) -> Fut + Send + 'static,
Fut: std::future::Future<Output = CaResult<()>> + Send + 'static,
{
let result = self.run_to_completion(protocol_runner).await;
crate::runtime::exit::call_at_exits();
result
}
async fn run_to_completion<F, Fut>(self, protocol_runner: F) -> Result<(), IocRunFailure>
where
F: FnOnce(IocRunConfig) -> Fut + Send + 'static,
Fut: std::future::Future<Output = CaResult<()>> + Send + 'static,
{
let db = Arc::new(PvDatabase::new());
db.begin_load()
.expect("a database created a line ago has not run iocInit");
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 commands,
startup_script,
startup_lines,
inline_pvs,
inline_records,
after_init_hooks,
link_set_installers,
ioc_init_gate,
} = self;
let acf = access_security::new_acf_cell(acf);
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());
commands.extend(cmds);
}
commands.push(db_load_group_startup_command());
for cmd in commands {
iocsh::register_command(cmd);
}
db.install_device_support_resolver(device_support_resolver(
device_factories,
dynamic_device_factory,
));
db.install_subroutine_registry(subroutine_registry).await;
for (name, value) in inline_pvs {
db.add_pv(&name, value).await?;
}
for (name, record) in inline_records {
db.add_record(&name, record).await?;
}
arm_build(IocBuild {
db: db.clone(),
acf: acf.clone(),
autosave_config: autosave_config.clone(),
autosave_startup,
link_set_installers,
after_init_hooks,
protocol: ProtocolStart {
bridge: bridge.clone(),
port,
tcp_port,
acf: acf.clone(),
autosave_config,
runner: Box::new(move |config| Box::pin(protocol_runner(config))),
},
});
let _armed = ArmedLifecycle;
if startup_script.is_some() || !startup_lines.is_empty() {
let _script_phase = iocsh::startup_script_phase();
let script = startup_script;
let db1 = db.clone();
let b1 = bridge.clone();
let acf1 = acf.clone();
let (tx, rx) = crate::runtime::sync::oneshot::channel();
crate::runtime::task::MandatoryThread::new(
"iocsh-startup",
crate::runtime::task::ThreadPriority::Iocsh,
crate::runtime::task::StackSizeClass::Big,
)
.try_spawn(move || {
let shell = iocsh::IocShell::new_with_acf(db1, b1, acf1);
let _ = tx.send(run_startup_phase(&shell, &startup_lines, script.as_deref()));
})
.map_err(|e| {
CaError::InvalidValue(format!("could not start the iocsh-startup thread: {e}"))
})?;
rx.await
.map_err(|_| CaError::InvalidValue("startup thread dropped".into()))??;
}
let decision = ioc_init_gate.map(|gate| gate());
let outcome = match take_lifecycle() {
Some(IocLifecycle::Armed(build)) => {
if let Some(decision) = decision
&& !decision.run
{
return run_uninitialized_tail(db, bridge, acf, decision.interactive).await;
}
match build.perform_build().await? {
BuildOutcome::Built(built) => Ok(built.run().await),
BuildOutcome::AsInitFailed => Err(()),
}
}
Some(IocLifecycle::Built(built)) => {
crate::runtime::log::errlog_printf(
"iocInit: startup script built the IOC without running it; running it now\n",
);
Ok(built.run().await)
}
Some(IocLifecycle::Running(running)) => Ok(running),
Some(IocLifecycle::AsInitFailed) => Err(()),
Some(IocLifecycle::Failed(e)) => return Err(e.into()),
None => unreachable!(
"the lifecycle owner is armed before the startup script runs and \
every transition puts a state back"
),
};
let running = match outcome {
Ok(running) => running,
Err(()) => {
return match decision {
Some(decision) => {
eprintln!("{} during iocInit()", crate::runtime::log::ERL_ERROR);
run_uninitialized_tail(db, bridge, acf, decision.interactive).await
}
None => Err(IocRunFailure::Startup(CaError::InvalidValue(
"iocBuild: asInit Failed.".into(),
))),
};
}
};
let mut server = running.server;
let pending = db.take_after_ioc_running();
if !pending.is_empty() {
let db1 = db.clone();
let b1 = bridge.clone();
let acf1 = acf.clone();
let (tx, rx) = crate::runtime::sync::oneshot::channel();
crate::runtime::task::MandatoryThread::new(
"iocsh-after-ioc-running",
crate::runtime::task::ThreadPriority::Iocsh,
crate::runtime::task::StackSizeClass::Big,
)
.try_spawn(move || {
let shell = iocsh::IocShell::new_with_acf(db1, b1, acf1);
let mut errs: Vec<String> = Vec::new();
for line in pending {
match shell.execute_line(&line) {
Err(e) => errs.push(format!("{line}: {e}")),
Ok(iocsh::registry::CommandOutcome::Failed) => {
errs.push(format!("{line}: failed"));
}
Ok(
iocsh::registry::CommandOutcome::Continue
| iocsh::registry::CommandOutcome::Exit,
) => {}
}
}
let _ = tx.send(errs);
})
.map_err(|e| {
CaError::InvalidValue(format!(
"could not start the iocsh-after-ioc-running thread: {e}"
))
})?;
if let Ok(errs) = rx.await {
for e in errs {
eprintln!("afterIocRunning: {e}");
}
}
}
#[cfg(not(epics_embedded_target))]
let ctrl_c = async {
let _ = tokio::signal::ctrl_c().await;
};
#[cfg(epics_embedded_target)]
let ctrl_c = std::future::pending::<()>();
#[cfg(all(unix, not(epics_embedded_target)))]
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(epics_embedded_target))))]
let sigterm = std::future::pending::<()>();
let outcome = tokio::select! {
biased;
res = server.wait() => Some(res),
_ = ctrl_c => {
tracing::info!(target: "epics_base_rs::ioc_app", "SIGINT received, shutting down IOC");
None
}
_ = sigterm => {
tracing::info!(target: "epics_base_rs::ioc_app", "SIGTERM received, shutting down IOC");
None
}
};
match outcome {
Some(res) => res,
None => {
server.shut_down().await;
Ok(())
}
}
}
}
pub type DeviceSupportResolver =
Arc<dyn Fn(&DeviceSupportContext) -> Option<Box<dyn DeviceSupport>> + Send + Sync>;
pub(crate) fn device_support_resolver(
factories: HashMap<String, DeviceSupportFactory>,
dynamic_factory: Option<DynamicDeviceSupportFactory>,
) -> DeviceSupportResolver {
Arc::new(move |ctx: &DeviceSupportContext| {
if let Some(factory) = factories.get(ctx.dtyp) {
Some(factory())
} else if let Some(dyn_factory) = dynamic_factory.as_ref() {
dyn_factory(ctx)
} else {
None
}
})
}
pub(crate) fn attach_device_support(
instance: &mut record::RecordInstance,
name: &str,
resolve: Option<&DeviceSupportResolver>,
) -> bool {
let dtyp = instance.common.dtyp.as_str().to_string();
if instance.common.dtyp.is_soft() {
return false;
}
let ctx = DeviceSupportContext {
dtyp: &dtyp,
inp: &instance.common.inp,
out: &instance.common.out,
};
let dev_opt = resolve.and_then(|resolve| resolve(&ctx));
let Some(dev) = dev_opt else {
eprintln!("warning: no device support registered for DTYP '{dtyp}' (record: {name})");
crate::server::recgbl::rec_gbl_no_device_support(instance.record.record_type(), name);
return false;
};
crate::server::device_support::attach_device_to_record(instance, dev);
true
}
pub(crate) fn wire_subroutine(
instance: &mut record::RecordInstance,
name: &str,
registry: &HashMap<String, Arc<SubroutineFn>>,
) -> bool {
let rt = instance.record.record_type();
if rt != "sub" && rt != "aSub" {
return true;
}
let erl = crate::runtime::log::ERL_ERROR;
if let Some(crate::types::EpicsValue::String(inam_field)) = instance.record.get_field("INAM") {
let inam = inam_field.as_str_lossy();
if !inam.is_empty() {
let Some(init_fn) = registry.get(inam.as_ref()) else {
eprintln!("{name}.INAM {erl} function '{inam}' not found");
return false;
};
let init_fn = init_fn.clone();
if let Err(e) = init_fn(&mut *instance.record) {
eprintln!("iocInit: {name}.INAM '{inam}' init routine failed: {e}");
}
}
}
if instance.record.is_subroutine_name_field("SNAM")
&& let Some(crate::types::EpicsValue::String(snam_field)) =
instance.record.get_field("SNAM")
{
let snam = snam_field.as_str_lossy();
if snam.is_empty() {
instance.subroutine = None;
if rt == "sub" {
crate::runtime::log::errlog_printf(&format!("{name}.SNAM is empty\n"));
instance.enter_pact();
return false;
}
} else {
instance.subroutine = registry.get(snam.as_ref()).cloned();
if instance.subroutine.is_none() {
eprintln!("{name}.SNAM {erl} function '{snam}' not found");
return false;
}
}
}
match rt {
"sub" => {
if let Some(val) = instance.record.get_field("VAL") {
for field in ["MLST", "ALST", "LALM"] {
let _ = instance.record.put_field(field, val.clone());
}
}
}
_ => {
if let Some(snam_field) = instance.record.get_field("SNAM") {
let _ = instance.record.put_field("ONAM", snam_field);
}
}
}
true
}
async fn demote_io_intr_to_passive(db: &PvDatabase, name: &str, message: &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);
}
crate::server::recgbl::rec_gbl_record_error("", name, message);
}
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<record::RecordCell>)> = {
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, "scanAdd: I/O Intr not valid (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();
let prio = inst.common.callback_priority();
crate::runtime::task::spawn_background(prio, async move {
while intr_rx.recv().await.is_some() {
if !crate::server::scan::scan_is_running() {
continue;
}
let process = independent || {
let inst = rec_arc_clone.read();
inst.common.scan == record::ScanType::IoIntr
};
if !process {
continue;
}
let mut visited = crate::server::database::ProcStack::new();
let _ = db_clone
.process_record_readback(&rec_name, &mut visited)
.await;
}
});
count += 1;
} else if on_io_intr {
demote.push((name, "scanAdd: I/O Intr not valid (no get_ioint_info)"));
}
inst.device = Some(dev);
}
for (name, message) in demote {
demote_io_intr_to_passive(&db, &name, message).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();
let prio = inst.common.callback_priority();
crate::runtime::task::spawn_background(prio, async move {
while let Some(post) = rx.recv().await {
let _ = db_clone.post_property(&rec_name, post);
}
});
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 source_guard::{Comments, production};
use std::sync::Mutex as StdMutex;
use std::sync::atomic::{AtomicUsize, Ordering};
static INIT_HOOK_TEST_LOCK: StdMutex<()> = StdMutex::new(());
#[test]
fn the_as_init_failure_lines_are_c_s() {
assert_eq!(
as_init_failed_message(false),
"ERROR iocBuild: asInit Failed.\n The IOC has not been started.\n"
);
assert_eq!(
as_init_failed_message(true),
format!(
"{} iocBuild: asInit Failed.\n\u{1b}[35;1m The IOC has not been \
started.\u{1b}[0m\n",
crate::runtime::log::ERL_ERROR
)
);
}
#[test]
fn iocsh_threads_take_the_iocsh_band() {
let prod = production(include_str!("ioc_app.rs"), Comments::Strip);
assert_eq!(
prod.matches("MandatoryThread::new(").count(),
3,
"the startup-script, afterIocRunning and uninitialised-tail threads"
);
assert_eq!(
prod.matches("name_current_thread(").count(),
0,
"naming without banding leaves the thread one level above idle on \
the target; the `MandatoryThread` prologue is the whole of it"
);
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", "iocsh"] {
let at = prod
.find(&format!("\"{name}\","))
.unwrap_or_else(|| panic!("the {name} thread moved; update this guard"));
let head = &prod[at..(at + 700).min(prod.len())];
assert!(
head.contains("ThreadPriority::Iocsh"),
"{name} must be declared at `ThreadPriority::Iocsh` \
(posix/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 tmpdir = tempfile::tempdir().expect("fixture root");
let a = tmpdir.path().join("qsrv_q_a.json");
let b = tmpdir.path().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());
assert_eq!(db.records_with_device_support().await, 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.install_device_support_resolver(device_support_resolver(HashMap::new(), None));
db.add_record("TEST", Box::new(AiRecord::new(0.0)))
.await
.unwrap();
assert_eq!(db.records_with_device_support().await, 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.install_device_support_resolver(device_support_resolver(factories, None));
db.add_loaded_record(
"AI:WITH:INFO",
Box::new(AiRecord::new(0.0)),
crate::server::database::RecordLoad {
common_fields: vec![(
"DTYP".to_string(),
crate::types::EpicsValue::String("TestRecording".into()),
)],
info_tags: vec![
("asyn:READBACK".to_string(), "1".to_string()),
("Q:group".to_string(), "demo".to_string()),
],
},
)
.await
.unwrap();
assert_eq!(
db.records_with_device_support().await,
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 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 db = Arc::new(PvDatabase::new());
db.install_device_support_resolver(device_support_resolver(HashMap::new(), dynamic));
for name in &names {
db.add_loaded_record(
name,
Box::new(AiRecord::new(0.0)),
crate::server::database::RecordLoad::from_common_fields(vec![
(
"DTYP".to_string(),
crate::types::EpicsValue::String("SeqDev".into()),
),
(
"INP".to_string(),
crate::types::EpicsValue::String(format!("@{name}").into()),
),
]),
)
.await
.unwrap();
}
assert_eq!(db.records_with_device_support().await, 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(
crate::server::device_support::DeviceUdf::Defined,
))
}
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".into();
inst.device = Some(Box::new(ReadbackDev {
writes: writes.clone(),
readback_val: 0,
}));
}
{
let mut visited = crate::server::database::ProcStack::new();
db.process_record_readback("BO:RBK", &mut visited)
.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 = crate::server::database::ProcStack::new();
db.process_record_with_links("BO:RBK", &mut visited)
.await
.unwrap();
}
assert_eq!(
writes.load(Ordering::SeqCst),
1,
"a put/scan cycle must write the setpoint to the driver exactly once"
);
}
}
#[cfg(test)]
mod lifecycle_tests {
use super::*;
use crate::server::scan::{ScanCtl, scan_ctl};
#[test]
fn a_void_ioc_refuses_run_and_pause() {
assert_eq!(get_ioc_state(), IocState::Void);
assert_eq!(ioc_run(), -1, "iocRun from iocVoid is C's -1");
assert_eq!(ioc_pause(), -1, "iocPause from iocVoid is C's -1");
assert_eq!(get_ioc_state(), IocState::Void, "a refusal changes nothing");
}
#[test]
fn shutting_down_a_void_ioc_is_a_success() {
assert_eq!(ioc_shutdown(), 0);
assert_eq!(get_ioc_state(), IocState::Void);
}
#[test]
fn the_lifecycle_walks_void_running_paused_running_void() {
note_scan_owner_started();
assert_eq!(get_ioc_state(), IocState::Running);
assert_eq!(scan_ctl(), ScanCtl::Run);
assert_eq!(ioc_pause(), 0);
assert_eq!(get_ioc_state(), IocState::Paused);
assert_eq!(
scan_ctl(),
ScanCtl::Pause,
"iocPause must close the gate every asynchronous scan source reads"
);
assert_eq!(ioc_run(), 0);
assert_eq!(get_ioc_state(), IocState::Running);
assert_eq!(scan_ctl(), ScanCtl::Run);
assert_eq!(ioc_shutdown(), 0);
assert_eq!(get_ioc_state(), IocState::Void);
assert_eq!(scan_ctl(), ScanCtl::Exit);
}
#[test]
fn a_repeated_transition_is_refused_from_its_own_end_state() {
note_scan_owner_started();
assert_eq!(ioc_run(), -1, "already running");
assert_eq!(get_ioc_state(), IocState::Running);
assert_eq!(ioc_pause(), 0);
assert_eq!(ioc_pause(), -1, "already paused");
assert_eq!(get_ioc_state(), IocState::Paused);
}
#[test]
fn a_paused_ioc_can_be_shut_down() {
note_scan_owner_started();
assert_eq!(ioc_pause(), 0);
assert_eq!(ioc_shutdown(), 0);
assert_eq!(get_ioc_state(), IocState::Void);
}
#[epics_macros_rs::epics_test]
async fn a_paused_ioc_does_not_process_its_event_records() {
use crate::error::CaResult;
use crate::server::record::{FieldDesc, ProcessOutcome, Record, ScanType};
use crate::types::EpicsValue;
use std::sync::atomic::{AtomicUsize, Ordering};
struct CountProbe(Arc<AtomicUsize>);
impl Record for CountProbe {
fn record_type(&self) -> &'static str {
"ioc_pause_probe"
}
fn process(&mut self) -> CaResult<ProcessOutcome> {
self.0.fetch_add(1, Ordering::SeqCst);
Ok(ProcessOutcome::complete())
}
fn get_field(&self, _name: &str) -> Option<EpicsValue> {
None
}
fn put_field(&mut self, _name: &str, _value: EpicsValue) -> CaResult<()> {
Ok(())
}
fn declared_fields(&self) -> &'static [FieldDesc] {
&[]
}
}
let runs = Arc::new(AtomicUsize::new(0));
let db = Arc::new(PvDatabase::new());
db.add_record("EV", Box::new(CountProbe(Arc::clone(&runs))))
.await
.unwrap();
{
let rec = db.get_record("EV").unwrap();
rec.write().common.scan = ScanType::Event;
}
db.update_scan_index("EV", ScanType::Passive, ScanType::Event, 0, 0);
note_scan_owner_started();
db.post_event().await;
let while_running = runs.load(Ordering::SeqCst);
assert_eq!(while_running, 1, "a running IOC processes its Event list");
assert_eq!(ioc_pause(), 0);
db.post_event().await;
assert_eq!(
runs.load(Ordering::SeqCst),
while_running,
"a posted event must not process anything while the IOC is paused"
);
assert_eq!(ioc_run(), 0);
db.post_event().await;
assert_eq!(
runs.load(Ordering::SeqCst),
while_running + 1,
"iocRun must reopen the gate"
);
}
}