epics_libcom_rs/runtime/
log_client.rs

1//! The IOC log client — C `modules/libcom/src/log/{logClient,iocLog}.c`
2//! @`R7.0.10`.
3//!
4//! A site runs one `iocLogServer` and every IOC forwards its `errlog` stream
5//! to it over TCP. The client is nothing more than an errlog listener with a
6//! socket: [`ioc_log_init`] registers one with
7//! [`crate::runtime::log::errlog_add_listener`], and from then on every
8//! message the errlog worker drains is appended to a 16 KiB buffer and pushed
9//! to the server by a reconnecting background thread.
10//!
11//! Everything a site can observe is reproduced: the buffer size and its
12//! overflow message, the 5-second reconnect period, the `iocLogPrefix`
13//! write-once rule and its warning, and the exact `iocLog:` diagnostics for a
14//! missing or out-of-range environment variable.
15
16use std::io::Write;
17use std::net::{IpAddr, Ipv4Addr, SocketAddr, TcpStream};
18use std::sync::atomic::{AtomicBool, Ordering};
19use std::sync::{Arc, Condvar, Mutex, OnceLock};
20use std::time::Duration;
21
22use crate::runtime::env_table::{EPICS_IOC_LOG_INET, EPICS_IOC_LOG_PORT};
23
24/// Why [`ioc_log_init`] declined. C returns a bare `iocLogError` (-1) for
25/// both; naming them keeps the caller from having to re-read stderr to find
26/// out which happened.
27#[derive(Debug, Clone, Copy, PartialEq, Eq)]
28pub enum IocLogError {
29    /// `EPICS_IOC_LOG_INET` / `EPICS_IOC_LOG_PORT` do not name a log server.
30    NoServerConfigured,
31    /// The reconnection thread could not be created.
32    NoRestartThread,
33}
34
35impl std::fmt::Display for IocLogError {
36    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
37        match self {
38            Self::NoServerConfigured => {
39                f.write_str("no log server configured (EPICS_IOC_LOG_INET / EPICS_IOC_LOG_PORT)")
40            }
41            Self::NoRestartThread => f.write_str("could not start the log client thread"),
42        }
43    }
44}
45
46impl std::error::Error for IocLogError {}
47
48/// C `logClient::msgBuf[0x4000]` (`logClient.c:42`).
49const MSG_BUF_SIZE: usize = 0x4000;
50/// C `LOG_RESTART_DELAY` (`logClient.c:60`).
51const RESTART_DELAY: Duration = Duration::from_secs(5);
52
53/// C `iocLogDisable` (`iocLog.c:26`) — an exported iocsh variable, so it can
54/// be flipped before *or* after `iocLogInit`.
55static IOC_LOG_DISABLE: AtomicBool = AtomicBool::new(false);
56
57/// C `logClientPrefix` (`logClient.c:66`): file-scope, shared by every client,
58/// and prepended to every message.
59static PREFIX: Mutex<Option<String>> = Mutex::new(None);
60
61/// C's single `iocLogClient` (`iocLog.c:31`) — `iocLogInit` is a no-op once
62/// this is set, however many times it is called.
63static CLIENT: OnceLock<Arc<LogClient>> = OnceLock::new();
64
65struct LogClientState {
66    sock: Option<TcpStream>,
67    /// C `nextMsgIndex` bytes of `msgBuf`.
68    msg_buf: Vec<u8>,
69    connect_count: u32,
70    shutdown: bool,
71}
72
73struct LogClient {
74    addr: SocketAddr,
75    /// C `pClient->name`, the dotted address the diagnostics quote.
76    name: String,
77    state: Mutex<LogClientState>,
78    /// C `shutdownNotify` — what cuts the restart thread's wait short.
79    wake: Condvar,
80}
81
82impl LogClient {
83    /// C `sendMessageChunk` (`logClient.c:171-196`): fill the buffer, flushing
84    /// when it is full, and report the overflow exactly once per chunk.
85    fn send_chunk(&self, state: &mut LogClientState, text: &[u8]) {
86        let mut rest = text;
87        while !rest.is_empty() {
88            let mut left = MSG_BUF_SIZE - state.msg_buf.len();
89            if left < rest.len() && !state.msg_buf.is_empty() && state.sock.is_some() {
90                self.flush_locked(state);
91                left = MSG_BUF_SIZE - state.msg_buf.len();
92            }
93            if left == 0 {
94                eprintln!("log client: messages to \"{}\" are lost", self.name);
95                break;
96            }
97            let take = left.min(rest.len());
98            state.msg_buf.extend_from_slice(&rest[..take]);
99            rest = &rest[take..];
100        }
101    }
102
103    /// C `logClientSend` (`logClient.c:202-221`) — the prefix, then the
104    /// message, under one lock so the two cannot interleave with another
105    /// thread's pair.
106    fn send(&self, message: &str) {
107        let mut state = self.state.lock().expect("log client");
108        if let Some(prefix) = PREFIX.lock().expect("log client prefix").as_deref() {
109            self.send_chunk(&mut state, prefix.as_bytes());
110        }
111        self.send_chunk(&mut state, message.as_bytes());
112    }
113
114    /// C `logClientFlush` (`logClient.c:222-273`): push what is buffered and
115    /// close on any write error, so the restart thread reconnects.
116    fn flush_locked(&self, state: &mut LogClientState) {
117        let Some(sock) = state.sock.as_mut() else {
118            return;
119        };
120        match sock.write_all(&state.msg_buf) {
121            Ok(()) => {
122                let _ = sock.flush();
123                state.msg_buf.clear();
124            }
125            Err(e) => {
126                eprintln!(
127                    "log client: lost contact with log server at '{}'\n because \"{e}\"",
128                    self.name
129                );
130                state.sock = None;
131            }
132        }
133    }
134
135    /// C `logClientConnect` (`logClient.c:308-424`), minus the non-blocking
136    /// dance: a blocking `connect` with a timeout reaches the same two states,
137    /// and the restart thread retries either way.
138    fn connect(&self) {
139        let sock = TcpStream::connect_timeout(&self.addr, RESTART_DELAY);
140        let mut state = self.state.lock().expect("log client");
141        match sock {
142            Ok(sock) => {
143                let _ = sock.set_nodelay(true);
144                state.sock = Some(sock);
145                state.connect_count += 1;
146                eprintln!("log client: connected to log server at '{}'", self.name);
147            }
148            Err(_) => {
149                // C prints its connect failure only once per distinct errno
150                // (`connFailStatus`), so a log server that is simply not
151                // running does not fill the console every 5 seconds. Silence
152                // here is the same choice made whole.
153                state.sock = None;
154            }
155        }
156    }
157}
158
159/// C `logClientRestart` (`logClient.c:426-449`): reconnect if down, flush,
160/// wait 5 s, repeat.
161fn restart_thread(client: Arc<LogClient>) {
162    loop {
163        let (connected, shutdown) = {
164            let state = client.state.lock().expect("log client");
165            (state.sock.is_some(), state.shutdown)
166        };
167        if shutdown {
168            return;
169        }
170        if !connected {
171            client.connect();
172        }
173        {
174            let mut state = client.state.lock().expect("log client");
175            client.flush_locked(&mut state);
176        }
177        let state = client.state.lock().expect("log client");
178        let _ = client
179            .wake
180            .wait_timeout_while(state, RESTART_DELAY, |s| !s.shutdown);
181    }
182}
183
184/// C `getConfig` (`iocLog.c:37-66`) — both variables, both diagnostics.
185fn get_config() -> Result<SocketAddr, IocLogError> {
186    let Some(port) = EPICS_IOC_LOG_PORT.long() else {
187        eprintln!(
188            "iocLog: EPICS environment variable \"{}\" undefined",
189            EPICS_IOC_LOG_PORT.name()
190        );
191        return Err(IocLogError::NoServerConfigured);
192    };
193    if !(0..=i64::from(u16::MAX)).contains(&port) {
194        eprintln!(
195            "iocLog: EPICS environment variable \"{}\" out of range",
196            EPICS_IOC_LOG_PORT.name()
197        );
198        return Err(IocLogError::NoServerConfigured);
199    }
200    // C `envGetInetAddrConfigParam` fails on an unset or unparsable value, and
201    // `EPICS_IOC_LOG_INET`'s compiled default is empty — so an IOC that was
202    // never told where its log server is reports the variable undefined rather
203    // than connecting somewhere.
204    let inet = EPICS_IOC_LOG_INET.get().unwrap_or_default();
205    let Ok(addr) = inet.trim().parse::<Ipv4Addr>() else {
206        eprintln!(
207            "iocLog: EPICS environment variable \"{}\" undefined",
208            EPICS_IOC_LOG_INET.name()
209        );
210        return Err(IocLogError::NoServerConfigured);
211    };
212    #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
213    Ok(SocketAddr::new(IpAddr::V4(addr), port as u16))
214}
215
216/// C `iocLogInit` (`iocLog.c:121-142`).
217///
218/// A no-op when logging is disabled or a client already exists, so a startup
219/// script may call it more than once. On success the client is registered as
220/// an errlog listener and every subsequent message reaches the log server.
221///
222/// # Errors
223/// Returns [`IocLogError`] when `EPICS_IOC_LOG_INET`/`EPICS_IOC_LOG_PORT` do
224/// not name a server, or when the reconnection thread cannot be created —
225/// exactly the two paths on which C returns `iocLogError`.
226pub fn ioc_log_init() -> Result<(), IocLogError> {
227    if IOC_LOG_DISABLE.load(Ordering::Relaxed) {
228        return Ok(());
229    }
230    if CLIENT.get().is_some() {
231        return Ok(());
232    }
233    let addr = get_config()?;
234    let client = Arc::new(LogClient {
235        addr,
236        name: addr.to_string(),
237        state: Mutex::new(LogClientState {
238            sock: None,
239            msg_buf: Vec::with_capacity(MSG_BUF_SIZE),
240            connect_count: 0,
241            shutdown: false,
242        }),
243        wake: Condvar::new(),
244    });
245    if CLIENT.set(Arc::clone(&client)).is_err() {
246        // Another thread won the race; its client is the one registered.
247        return Ok(());
248    }
249
250    let worker = Arc::clone(&client);
251    if crate::runtime::task::spawn_dedicated_thread(
252        "logRestart".to_string(),
253        crate::runtime::task::ThreadPriority::Low,
254        crate::runtime::task::StackSizeClass::Small,
255        move || restart_thread(worker),
256    )
257    .is_err()
258    {
259        eprintln!("log client: unable to start reconnection thread");
260        return Err(IocLogError::NoServerConfigured);
261    }
262
263    // C `logClientSendMessage` (`iocLog.c:79-84`) reads `iocLogDisable` on
264    // every message, not once at init, so `setIocLogDisable 1` silences a
265    // client that is already running.
266    let sender = Arc::clone(&client);
267    crate::runtime::log::errlog_add_listener(move |message| {
268        if !IOC_LOG_DISABLE.load(Ordering::Relaxed) {
269            sender.send(message);
270        }
271    });
272    Ok(())
273}
274
275/// C `iocLogPrefix` (`logClient.c:551-576`) — write-once.
276///
277/// The prefix is prepended to every message by every client, so C refuses to
278/// change one that is already in use and warns when the new value differs.
279/// A repeat of the SAME prefix is silent, which is what makes a startup script
280/// that is sourced twice harmless.
281pub fn ioc_log_prefix(prefix: &str) {
282    let mut current = PREFIX.lock().expect("log client prefix");
283    match current.as_deref() {
284        Some(existing) => {
285            if existing != prefix {
286                println!(
287                    "{} iocLogPrefix: The prefix was already set to \"{existing}\" and can't be changed.",
288                    crate::runtime::log::erl_warning()
289                );
290            }
291        }
292        None => *current = Some(prefix.to_string()),
293    }
294}
295
296/// The prefix in force, if one was set.
297#[must_use]
298pub fn ioc_log_prefix_get() -> Option<String> {
299    PREFIX.lock().expect("log client prefix").clone()
300}
301
302/// C `setIocLogDisable` (`libComRegister.c:226-229`).
303pub fn set_ioc_log_disable(disable: bool) {
304    IOC_LOG_DISABLE.store(disable, Ordering::Relaxed);
305}
306
307/// Whether forwarding is currently disabled.
308#[must_use]
309pub fn ioc_log_disabled() -> bool {
310    IOC_LOG_DISABLE.load(Ordering::Relaxed)
311}
312
313/// C `iocLogFlush` (`iocLog.c:70-75`) — push whatever is buffered now.
314pub fn ioc_log_flush() {
315    if let Some(client) = CLIENT.get() {
316        let mut state = client.state.lock().expect("log client");
317        client.flush_locked(&mut state);
318    }
319}
320
321/// C `iocLogShow`/`logClientShow` (`iocLog.c:145-152`, `logClient.c:513-545`).
322/// Returns the lines rather than printing them, so the iocsh command can send
323/// them through its own redirected output.
324#[must_use]
325pub fn ioc_log_show(level: u32) -> Vec<String> {
326    let Some(client) = CLIENT.get() else {
327        return Vec::new();
328    };
329    let state = client.state.lock().expect("log client");
330    let mut out = Vec::new();
331    if state.sock.is_some() {
332        out.push(format!(
333            "log client: connected to log server at '{}'",
334            client.name
335        ));
336    } else {
337        out.push(format!(
338            "log client: disconnected from log server at '{}'",
339            client.name
340        ));
341    }
342    if let Some(prefix) = PREFIX.lock().expect("log client prefix").as_deref() {
343        out.push(format!("log client: prefix is \"{prefix}\""));
344    }
345    if level > 0 {
346        out.push(format!(
347            "log client: sock {}, connect cycles = {}",
348            if state.sock.is_some() {
349                "OK"
350            } else {
351                "INVALID"
352            },
353            state.connect_count
354        ));
355    }
356    if level > 1 {
357        out.push(format!(
358            "log client: {} bytes in buffer",
359            state.msg_buf.len()
360        ));
361        if !state.msg_buf.is_empty() {
362            out.push("-------------------------".to_string());
363            out.push(String::from_utf8_lossy(&state.msg_buf).into_owned());
364            out.push("-------------------------".to_string());
365        }
366    }
367    out
368}
369
370#[cfg(test)]
371mod tests {
372    use super::*;
373    use serial_test::serial;
374    use std::io::Read;
375    use std::net::TcpListener;
376
377    /// Boundary: no log server configured. `EPICS_IOC_LOG_INET`'s compiled
378    /// default is empty, so an IOC that was never told where to log must
379    /// report the variable undefined and connect nowhere — C's `getConfig`
380    /// returns `iocLogError` before `logClientCreate` is reached
381    /// (`iocLog.c:56-63`).
382    #[test]
383    #[serial(ioc_log)]
384    fn with_no_inet_configured_init_declines_instead_of_connecting() {
385        unsafe {
386            std::env::remove_var("EPICS_IOC_LOG_INET");
387        }
388        assert_eq!(get_config(), Err(IocLogError::NoServerConfigured));
389    }
390
391    /// Boundary: the port must come from `EPICS_IOC_LOG_PORT`, whose
392    /// compiled default is 7004 (`envDefs`/`env_table`), and a value outside
393    /// `0..=65535` is refused rather than truncated.
394    #[test]
395    #[serial(ioc_log)]
396    fn the_port_defaults_to_7004_and_is_range_checked() {
397        unsafe {
398            std::env::set_var("EPICS_IOC_LOG_INET", "127.0.0.1");
399            std::env::remove_var("EPICS_IOC_LOG_PORT");
400        }
401        assert_eq!(get_config().expect("configured").port(), 7004);
402
403        unsafe {
404            std::env::set_var("EPICS_IOC_LOG_PORT", "70000");
405        }
406        assert_eq!(get_config(), Err(IocLogError::NoServerConfigured));
407        unsafe {
408            std::env::remove_var("EPICS_IOC_LOG_PORT");
409        }
410    }
411
412    /// Boundary: `iocLogPrefix` is write-once. C keeps the first value and
413    /// warns only when a LATER call differs (`logClient.c:560-573`), so a
414    /// startup script sourced twice is silent while a genuine conflict is
415    /// reported.
416    #[test]
417    #[serial(ioc_log)]
418    fn the_prefix_is_write_once_and_a_repeat_of_the_same_value_is_silent() {
419        *PREFIX.lock().expect("prefix") = None;
420        ioc_log_prefix("fac=SR ");
421        assert_eq!(ioc_log_prefix_get().as_deref(), Some("fac=SR "));
422        ioc_log_prefix("fac=SR ");
423        assert_eq!(ioc_log_prefix_get().as_deref(), Some("fac=SR "));
424        ioc_log_prefix("fac=BTS ");
425        assert_eq!(
426            ioc_log_prefix_get().as_deref(),
427            Some("fac=SR "),
428            "the first prefix stands; C refuses to change one already in use"
429        );
430        *PREFIX.lock().expect("prefix") = None;
431    }
432
433    /// The row's own observable: a log server receives the IOC's messages.
434    /// A real `TcpListener` stands in for `iocLogServer`, and the bytes it
435    /// reads must be the prefix followed by the errlog text.
436    #[test]
437    #[serial(ioc_log)]
438    fn a_log_server_receives_the_ioc_messages_with_the_prefix_prepended() {
439        let server = TcpListener::bind("127.0.0.1:0").expect("log server");
440        let port = server.local_addr().expect("addr").port();
441        unsafe {
442            std::env::set_var("EPICS_IOC_LOG_INET", "127.0.0.1");
443            std::env::set_var("EPICS_IOC_LOG_PORT", port.to_string());
444        }
445        *PREFIX.lock().expect("prefix") = None;
446        ioc_log_prefix("ioc=TEST ");
447        set_ioc_log_disable(false);
448        ioc_log_init().expect("the client must start");
449
450        let (mut peer, _) = server.accept().expect("the client must connect");
451        peer.set_read_timeout(Some(Duration::from_secs(5)))
452            .expect("read timeout");
453
454        crate::runtime::log::errlog_printf("bind failed\n");
455        crate::runtime::log::errlog_flush();
456
457        // The restart thread flushes every 5 s; push now so the test does not
458        // have to wait for it.
459        let mut buf = [0u8; 256];
460        let mut got = String::new();
461        for _ in 0..50 {
462            ioc_log_flush();
463            match peer.read(&mut buf) {
464                Ok(0) => break,
465                Ok(n) => {
466                    got.push_str(&String::from_utf8_lossy(&buf[..n]));
467                    break;
468                }
469                Err(_) => std::thread::sleep(Duration::from_millis(20)),
470            }
471        }
472        assert_eq!(
473            got, "ioc=TEST bind failed\n",
474            "the server must see the prefix then the message"
475        );
476
477        unsafe {
478            std::env::remove_var("EPICS_IOC_LOG_INET");
479            std::env::remove_var("EPICS_IOC_LOG_PORT");
480        }
481    }
482
483    /// Boundary: `setIocLogDisable 1` on a client that is ALREADY running.
484    /// C reads `iocLogDisable` inside `logClientSendMessage`, per message
485    /// (`iocLog.c:79-84`), so the switch takes effect without tearing the
486    /// connection down.
487    #[test]
488    #[serial(ioc_log)]
489    fn disabling_forwarding_silences_a_client_that_is_already_connected() {
490        let server = TcpListener::bind("127.0.0.1:0").expect("log server");
491        let port = server.local_addr().expect("addr").port();
492        unsafe {
493            std::env::set_var("EPICS_IOC_LOG_INET", "127.0.0.1");
494            std::env::set_var("EPICS_IOC_LOG_PORT", port.to_string());
495        }
496        *PREFIX.lock().expect("prefix") = None;
497        set_ioc_log_disable(false);
498        ioc_log_init().expect("the client must start");
499        let (mut peer, _) = server.accept().expect("the client must connect");
500        peer.set_read_timeout(Some(Duration::from_millis(300)))
501            .expect("read timeout");
502
503        set_ioc_log_disable(true);
504        crate::runtime::log::errlog_printf("suppressed\n");
505        crate::runtime::log::errlog_flush();
506        ioc_log_flush();
507
508        let mut buf = [0u8; 64];
509        let n = peer.read(&mut buf).unwrap_or(0);
510        assert_eq!(
511            n,
512            0,
513            "nothing may reach the server while iocLogDisable is set: {:?}",
514            String::from_utf8_lossy(&buf[..n])
515        );
516        set_ioc_log_disable(false);
517        unsafe {
518            std::env::remove_var("EPICS_IOC_LOG_INET");
519            std::env::remove_var("EPICS_IOC_LOG_PORT");
520        }
521    }
522}
epics_libcom_rs/runtime/log_client.rs

epics_libcom_rs/runtime/
log_client.rs
