autocore_std/lib.rs
1//! # AutoCore Standard Library
2//!
3//! The standard library for writing AutoCore control programs. This crate provides
4//! everything you need to build real-time control applications that integrate with
5//! the AutoCore server ecosystem.
6//!
7//! ## Overview
8//!
9//! AutoCore control programs run as separate processes that communicate with the
10//! autocore-server via shared memory and IPC. This library handles all the low-level
11//! details, allowing you to focus on your control logic.
12//!
13//! ```text
14//! ┌─────────────────────────┐ ┌─────────────────────────┐
15//! │ autocore-server │ │ Your Control Program │
16//! │ │ │ │
17//! │ ┌─────────────────┐ │ │ ┌─────────────────┐ │
18//! │ │ Shared Memory │◄───┼─────┼──│ ControlRunner │ │
19//! │ │ (GlobalMemory) │ │ │ │ │ │
20//! │ └─────────────────┘ │ │ │ ┌─────────────┐ │ │
21//! │ │ │ │ │ Your Logic │ │ │
22//! │ ┌─────────────────┐ │ │ │ └─────────────┘ │ │
23//! │ │ Tick Signal │────┼─────┼──│ │ │
24//! │ └─────────────────┘ │ │ └─────────────────┘ │
25//! └─────────────────────────┘ └─────────────────────────┘
26//! ```
27//!
28//! ## Quick Start
29//!
30//! 1. Create a new control project using `acctl`:
31//! ```bash
32//! acctl clone <server-ip> <project-name>
33//! ```
34//!
35//! 2. Implement the [`ControlProgram`] trait:
36//! ```ignore
37//! use autocore_std::{ControlProgram, TickContext};
38//! use autocore_std::fb::RTrig;
39//!
40//! // GlobalMemory is generated from your project.json
41//! mod gm;
42//! use gm::GlobalMemory;
43//!
44//! pub struct MyProgram {
45//! start_button: RTrig,
46//! }
47//!
48//! impl MyProgram {
49//! pub fn new() -> Self {
50//! Self {
51//! start_button: RTrig::new(),
52//! }
53//! }
54//! }
55//!
56//! impl ControlProgram for MyProgram {
57//! type Memory = GlobalMemory;
58//!
59//! fn process_tick(&mut self, ctx: &mut TickContext<Self::Memory>) {
60//! // Detect rising edge on start button
61//! if self.start_button.call(ctx.gm.inputs.start_button) {
62//! ctx.gm.outputs.motor_running = true;
63//! autocore_std::log::info!("Motor started!");
64//! }
65//! }
66//! }
67//! ```
68//!
69//! 3. Use the [`autocore_main!`] macro for the entry point:
70//! ```ignore
71//! autocore_std::autocore_main!(MyProgram, "my_project_shm", "tick");
72//! ```
73//!
74//! ## Function Blocks (IEC 61131-3 Inspired)
75//!
76//! This library includes standard function blocks commonly used in PLC programming:
77//!
78//! - [`fb::RTrig`] - Rising edge detector (false→true transition)
79//! - [`fb::FTrig`] - Falling edge detector (true→false transition)
80//! - [`fb::Ton`] - Timer On Delay (output after delay)
81//! - [`fb::BitResetOnDelay`] - Resets a boolean after it has been true for a duration
82//! - [`fb::SimpleTimer`] - Simple one-shot timer (NOT IEC 61131-3, for imperative use)
83//! - [`fb::StateMachine`] - State machine helper with automatic timer management
84//! - [`fb::RunningAverage`] - Accumulates values and computes their arithmetic mean
85//! - [`fb::Beeper`] - Audible beeper controller with configurable beep sequences
86//! - [`fb::Heartbeat`] - Monitors a remote heartbeat counter for connection loss
87//!
88//! ### Example: Edge Detection
89//!
90//! ```
91//! use autocore_std::fb::RTrig;
92//!
93//! let mut trigger = RTrig::new();
94//!
95//! // First call with false - no edge
96//! assert_eq!(trigger.call(false), false);
97//!
98//! // Rising edge detected!
99//! assert_eq!(trigger.call(true), true);
100//!
101//! // Still true, but no edge (already high)
102//! assert_eq!(trigger.call(true), false);
103//!
104//! // Back to false
105//! assert_eq!(trigger.call(false), false);
106//!
107//! // Another rising edge
108//! assert_eq!(trigger.call(true), true);
109//! ```
110//!
111//! ### Example: Timer
112//!
113//! ```
114//! use autocore_std::fb::Ton;
115//! use std::time::Duration;
116//!
117//! let mut timer = Ton::new();
118//! let delay = Duration::from_millis(100);
119//!
120//! // Timer not enabled - output is false
121//! assert_eq!(timer.call(false, delay), false);
122//!
123//! // Enable timer - starts counting
124//! assert_eq!(timer.call(true, delay), false);
125//!
126//! // Still counting...
127//! std::thread::sleep(Duration::from_millis(50));
128//! assert_eq!(timer.call(true, delay), false);
129//! assert!(timer.et < delay); // Elapsed time < preset
130//!
131//! // After delay elapsed
132//! std::thread::sleep(Duration::from_millis(60));
133//! assert_eq!(timer.call(true, delay), true); // Output is now true!
134//! ```
135//!
136//! ## Logging
137//!
138//! Control programs can send log messages to the autocore-server for display in the
139//! web console. Logging is handled automatically when using [`ControlRunner`].
140//!
141//! ```ignore
142//! use autocore_std::log;
143//!
144//! log::trace!("Detailed trace message");
145//! log::debug!("Debug information");
146//! log::info!("Normal operation message");
147//! log::warn!("Warning condition detected");
148//! log::error!("Error occurred!");
149//! ```
150//!
151//! See the [`logger`] module for advanced configuration.
152//!
153//! ## Memory Synchronization
154//!
155//! The [`ControlRunner`] handles all shared memory synchronization automatically:
156//!
157//! 1. **Wait for tick** - Blocks until the server signals a new cycle
158//! 2. **Read inputs** - Copies shared memory to local buffer (atomic snapshot)
159//! 3. **Execute logic** - Your `process_tick` runs on the local buffer
160//! 4. **Write outputs** - Copies local buffer back to shared memory
161//!
162//! This ensures your control logic always sees a consistent view of the data,
163//! even when other processes are modifying shared memory.
164
165#![warn(missing_docs)]
166#![warn(rustdoc::missing_crate_level_docs)]
167#![doc(html_root_url = "https://docs.rs/autocore-std/3.3.0")]
168
169use anyhow::{anyhow, Result};
170use futures_util::{SinkExt, StreamExt};
171use log::LevelFilter;
172use mechutil::ipc::{CommandMessage, MessageType};
173use raw_sync::events::{Event, EventInit, EventState};
174use raw_sync::Timeout;
175use shared_memory::ShmemConf;
176use std::collections::HashMap;
177use std::sync::atomic::{fence, Ordering, AtomicBool};
178use std::sync::Arc;
179use std::time::Duration;
180use tokio_tungstenite::{connect_async, tungstenite::Message};
181
182/// UDP logger for sending log messages to autocore-server.
183///
184/// This module provides a non-blocking logger implementation that sends log messages
185/// via UDP to the autocore-server. Messages are batched and sent asynchronously to
186/// avoid impacting the control loop timing.
187///
188/// # Example
189///
190/// ```ignore
191/// use autocore_std::logger;
192/// use log::LevelFilter;
193///
194/// // Initialize the logger (done automatically by ControlRunner)
195/// logger::init_udp_logger("127.0.0.1", 39101, LevelFilter::Info, "control")?;
196///
197/// // Now you can use the log macros
198/// log::info!("System initialized");
199/// ```
200pub mod logger;
201
202// Re-export log crate for convenience - control programs can use autocore_std::log::info!() etc.
203pub use log;
204
205/// Function blocks for control programs (IEC 61131-3 inspired).
206pub mod fb;
207
208/// Interface protocols for communication between control programs and external sources.
209pub mod iface;
210
211/// Client for sending IPC commands to external modules via WebSocket.
212pub mod command_client;
213pub use command_client::CommandClient;
214
215/// EtherCAT utilities (SDO client, etc.).
216pub mod ethercat;
217
218/// CiA 402 motion control: axis abstraction, traits, and types.
219pub mod motion;
220
221/// Shared memory utilities for external modules.
222pub mod shm;
223
224/// Lightweight process diagnostics (FD count, RSS).
225pub mod diagnostics;
226
227/// Banner Engineering device helpers (WLS15 IO-Link light strip, etc.).
228pub mod banner;
229
230/// Fixed-length string type for shared memory variables.
231pub mod fixed_string;
232pub use fixed_string::FixedString;
233
234// ============================================================================
235// Core Framework
236// ============================================================================
237
238/// Marker trait for generated GlobalMemory structs.
239///
240/// This trait is implemented by the auto-generated `GlobalMemory` struct
241/// that represents the shared memory layout. It serves as a marker for
242/// type safety in the control framework.
243///
244/// You don't need to implement this trait yourself - it's automatically
245/// implemented by the code generator.
246pub trait AutoCoreMemory {}
247
248/// Trait for detecting changes in memory structures.
249pub trait ChangeTracker {
250 /// Compare self with a previous state and return a list of changed fields.
251 /// Returns a vector of (field_name, new_value).
252 fn get_changes(&self, prev: &Self) -> Vec<(&'static str, serde_json::Value)>;
253
254 /// Unpack bit-mapped variables from their source words.
255 /// Called automatically after reading shared memory, before `process_tick`.
256 /// Auto-generated by codegen when bit-mapped variables exist; default is no-op.
257 fn unpack_bits(&mut self) {}
258
259 /// Pack bit-mapped variables back into their source words.
260 /// Called automatically after `process_tick`, before writing shared memory.
261 /// Only packs sources where at least one mapped bool changed since `pre_tick`.
262 /// Auto-generated by codegen when bit-mapped variables exist; default is no-op.
263 fn pack_bits(&mut self, _pre_tick: &Self) {}
264}
265
266/// Per-tick context passed to the control program by the framework.
267///
268/// `TickContext` bundles all per-cycle data into a single struct so that the
269/// [`ControlProgram::process_tick`] signature stays stable as new fields are
270/// added in the future (e.g., delta time, diagnostics).
271///
272/// The framework constructs a fresh `TickContext` each cycle, calls
273/// [`CommandClient::poll`] before handing it to the program, and writes
274/// the memory back to shared memory after `process_tick` returns.
275pub struct TickContext<'a, M> {
276 /// Mutable reference to the local shared memory copy.
277 pub gm: &'a mut M,
278 /// IPC command client for communicating with external modules.
279 pub client: &'a mut CommandClient,
280 /// Current cycle number (starts at 1, increments each tick).
281 pub cycle: u64,
282}
283
284/// The trait that defines a control program's logic.
285///
286/// Implement this trait to create your control program. The associated `Memory`
287/// type should be the generated `GlobalMemory` struct from your project.
288///
289/// # Memory Type Requirements
290///
291/// The `Memory` type must implement `Copy` to allow efficient synchronization
292/// between shared memory and local buffers. This is automatically satisfied
293/// by the generated `GlobalMemory` struct.
294///
295/// # Lifecycle
296///
297/// 1. `initialize` is called once at startup
298/// 2. `process_tick` is called repeatedly in the control loop with a
299/// [`TickContext`] that provides shared memory, the IPC client, and the
300/// current cycle number.
301///
302/// # Example
303///
304/// ```ignore
305/// use autocore_std::{ControlProgram, TickContext};
306///
307/// mod gm;
308/// use gm::GlobalMemory;
309///
310/// pub struct MyController {
311/// cycle_counter: u64,
312/// }
313///
314/// impl MyController {
315/// pub fn new() -> Self {
316/// Self { cycle_counter: 0 }
317/// }
318/// }
319///
320/// impl ControlProgram for MyController {
321/// type Memory = GlobalMemory;
322///
323/// fn initialize(&mut self, mem: &mut GlobalMemory) {
324/// // Set initial output states
325/// mem.outputs.ready = true;
326/// log::info!("Controller initialized");
327/// }
328///
329/// fn process_tick(&mut self, ctx: &mut TickContext<Self::Memory>) {
330/// self.cycle_counter = ctx.cycle;
331///
332/// // Your control logic here
333/// if ctx.gm.inputs.start && !ctx.gm.inputs.estop {
334/// ctx.gm.outputs.running = true;
335/// }
336/// }
337/// }
338/// ```
339pub trait ControlProgram {
340 /// The shared memory structure type (usually the generated `GlobalMemory`).
341 ///
342 /// Must implement `Copy` to allow efficient memory synchronization.
343 type Memory: Copy + ChangeTracker;
344
345 /// Called once when the control program starts.
346 ///
347 /// Use this to initialize output states, reset counters, or perform
348 /// any one-time setup. The default implementation does nothing.
349 ///
350 /// # Arguments
351 ///
352 /// * `mem` - Mutable reference to the shared memory. Changes are written
353 /// back to shared memory after this method returns.
354 fn initialize(&mut self, _mem: &mut Self::Memory) {}
355
356 /// The main control loop - called once per scan cycle.
357 ///
358 /// This is where your control logic lives. Read inputs from `ctx.gm`,
359 /// perform calculations, and write outputs back to `ctx.gm`. Use
360 /// `ctx.client` for IPC commands and `ctx.cycle` for the current cycle
361 /// number.
362 ///
363 /// The framework calls [`CommandClient::poll`] before each invocation,
364 /// so incoming responses are already buffered when your code runs.
365 ///
366 /// # Arguments
367 ///
368 /// * `ctx` - A [`TickContext`] containing the local shared memory copy,
369 /// the IPC command client, and the current cycle number.
370 ///
371 /// # Timing
372 ///
373 /// This method should complete within the scan cycle time. Long-running
374 /// operations will cause cycle overruns.
375 fn process_tick(&mut self, ctx: &mut TickContext<Self::Memory>);
376}
377
378/// Configuration for the [`ControlRunner`].
379///
380/// Specifies connection parameters, shared memory names, and logging settings.
381/// Use [`Default::default()`] for typical configurations.
382///
383/// # Example
384///
385/// ```
386/// use autocore_std::RunnerConfig;
387/// use log::LevelFilter;
388///
389/// let config = RunnerConfig {
390/// server_host: "192.168.1.100".to_string(),
391/// module_name: "my_controller".to_string(),
392/// shm_name: "my_project_shm".to_string(),
393/// tick_signal_name: "tick".to_string(),
394/// busy_signal_name: Some("busy".to_string()),
395/// log_level: LevelFilter::Debug,
396/// ..Default::default()
397/// };
398/// ```
399#[derive(Debug, Clone)]
400pub struct RunnerConfig {
401 /// Server host address (default: "127.0.0.1")
402 pub server_host: String,
403 /// WebSocket port for commands (default: 11969)
404 pub ws_port: u16,
405 /// Module name for identification (default: "control")
406 pub module_name: String,
407 /// Shared memory segment name (must match server configuration)
408 pub shm_name: String,
409 /// Name of the tick signal in shared memory (triggers each scan cycle)
410 pub tick_signal_name: String,
411 /// Optional name of the busy signal (set when cycle completes)
412 pub busy_signal_name: Option<String>,
413 /// Minimum log level to send to the server (default: Info)
414 pub log_level: LevelFilter,
415 /// UDP port for sending logs to the server (default: 39101)
416 pub log_udp_port: u16,
417}
418
419/// Default WebSocket port for autocore-server
420pub const DEFAULT_WS_PORT: u16 = 11969;
421
422impl Default for RunnerConfig {
423 fn default() -> Self {
424 Self {
425 server_host: "127.0.0.1".to_string(),
426 ws_port: DEFAULT_WS_PORT,
427 module_name: "control".to_string(),
428 shm_name: "autocore_cyclic".to_string(),
429 tick_signal_name: "tick".to_string(),
430 busy_signal_name: None,
431 log_level: LevelFilter::Info,
432 log_udp_port: logger::DEFAULT_LOG_UDP_PORT,
433 }
434 }
435}
436
437
438/// The main execution engine for control programs.
439///
440/// `ControlRunner` handles all the infrastructure required to run a control program:
441///
442/// - Reading memory layout from the server's layout file
443/// - Opening and mapping shared memory
444/// - Setting up synchronization signals
445/// - Running the real-time control loop
446/// - Sending log messages to the server
447///
448/// # Usage
449///
450/// ```ignore
451/// use autocore_std::{ControlRunner, RunnerConfig};
452///
453/// let config = RunnerConfig {
454/// shm_name: "my_project_shm".to_string(),
455/// tick_signal_name: "tick".to_string(),
456/// ..Default::default()
457/// };
458///
459/// ControlRunner::new(MyProgram::new())
460/// .config(config)
461/// .run()?; // Blocks forever
462/// ```
463///
464/// # Control Loop
465///
466/// The runner executes a synchronous control loop:
467///
468/// 1. **Wait** - Blocks until the tick signal is set by the server
469/// 2. **Read** - Copies shared memory to a local buffer (acquire barrier)
470/// 3. **Execute** - Calls your `process_tick` method
471/// 4. **Write** - Copies local buffer back to shared memory (release barrier)
472/// 5. **Signal** - Sets the busy signal (if configured) to indicate completion
473///
474/// This ensures your code always sees a consistent snapshot of the data
475/// and that your writes are atomically visible to other processes.
476pub struct ControlRunner<P: ControlProgram> {
477 config: RunnerConfig,
478 program: P,
479}
480
481impl<P: ControlProgram> ControlRunner<P> {
482 /// Creates a new runner for the given control program.
483 ///
484 /// Uses default configuration. Call [`.config()`](Self::config) to customize.
485 ///
486 /// # Arguments
487 ///
488 /// * `program` - Your control program instance
489 ///
490 /// # Example
491 ///
492 /// ```ignore
493 /// let runner = ControlRunner::new(MyProgram::new());
494 /// ```
495 pub fn new(program: P) -> Self {
496 Self {
497 config: RunnerConfig::default(),
498 program,
499 }
500 }
501
502 /// Sets the configuration for this runner.
503 ///
504 /// # Arguments
505 ///
506 /// * `config` - The configuration to use
507 ///
508 /// # Example
509 ///
510 /// ```ignore
511 /// ControlRunner::new(MyProgram::new())
512 /// .config(RunnerConfig {
513 /// shm_name: "custom_shm".to_string(),
514 /// ..Default::default()
515 /// })
516 /// .run()?;
517 /// ```
518 pub fn config(mut self, config: RunnerConfig) -> Self {
519 self.config = config;
520 self
521 }
522
523 /// Starts the control loop.
524 ///
525 /// This method blocks indefinitely, running the control loop until
526 /// an error occurs or the process is terminated.
527 ///
528 /// # Returns
529 ///
530 /// Returns `Ok(())` only if the loop exits cleanly (which typically
531 /// doesn't happen). Returns an error if:
532 ///
533 /// - IPC connection fails
534 /// - Shared memory cannot be opened
535 /// - Signal offsets cannot be found
536 /// - A critical error occurs during execution
537 ///
538 /// # Example
539 ///
540 /// ```ignore
541 /// fn main() -> anyhow::Result<()> {
542 /// ControlRunner::new(MyProgram::new())
543 /// .config(config)
544 /// .run()
545 /// }
546 /// ```
547 pub fn run(mut self) -> Result<()> {
548 // Initialize UDP logger FIRST (before any log statements)
549 if let Err(e) = logger::init_udp_logger(
550 &self.config.server_host,
551 self.config.log_udp_port,
552 self.config.log_level,
553 "control",
554 ) {
555 eprintln!("Warning: Failed to initialize UDP logger: {}", e);
556 // Continue anyway - logging will just go nowhere
557 }
558
559 // Multi-threaded runtime so spawned WS read/write tasks can run
560 // alongside the synchronous control loop.
561 let rt = tokio::runtime::Builder::new_multi_thread()
562 .worker_threads(2)
563 .enable_all()
564 .build()?;
565
566 rt.block_on(async {
567 log::info!("AutoCore Control Runner Starting...");
568
569 // 1. Connect to server via WebSocket and get layout
570 let ws_url = format!("ws://{}:{}/ws/", self.config.server_host, self.config.ws_port);
571 log::info!("Connecting to server at {}", ws_url);
572
573 let (ws_stream, _) = connect_async(&ws_url).await
574 .map_err(|e| anyhow!("Failed to connect to server at {}: {}", ws_url, e))?;
575
576 let (mut write, mut read) = ws_stream.split();
577
578 // Send gm.get_layout request
579 let request = CommandMessage::request("gm.get_layout", serde_json::Value::Null);
580 let transaction_id = request.transaction_id;
581 let request_json = serde_json::to_string(&request)?;
582
583 write.send(Message::Text(request_json)).await
584 .map_err(|e| anyhow!("Failed to send layout request: {}", e))?;
585
586 // Wait for response with matching transaction_id
587 let timeout = Duration::from_secs(10);
588 let start = std::time::Instant::now();
589 let mut layout: Option<HashMap<String, serde_json::Value>> = None;
590
591 while start.elapsed() < timeout {
592 match tokio::time::timeout(Duration::from_secs(1), read.next()).await {
593 Ok(Some(Ok(Message::Text(text)))) => {
594 if let Ok(response) = serde_json::from_str::<CommandMessage>(&text) {
595 if response.transaction_id == transaction_id {
596 if !response.success {
597 return Err(anyhow!("Server error: {}", response.error_message));
598 }
599 layout = Some(serde_json::from_value(response.data)?);
600 break;
601 }
602 // Skip broadcasts and other messages
603 if response.message_type == MessageType::Broadcast {
604 continue;
605 }
606 }
607 }
608 Ok(Some(Ok(_))) => continue,
609 Ok(Some(Err(e))) => return Err(anyhow!("WebSocket error: {}", e)),
610 Ok(None) => return Err(anyhow!("Server closed connection")),
611 Err(_) => continue, // Timeout on single read, keep trying
612 }
613 }
614
615 let layout = layout.ok_or_else(|| anyhow!("Timeout waiting for layout response"))?;
616 log::info!("Layout received with {} entries.", layout.len());
617
618 // Set up channels and background tasks for shared WebSocket access.
619 // This allows both the control loop (gm.write) and CommandClient (IPC
620 // commands) to share the write half, while routing incoming responses
621 // to the CommandClient.
622 let (ws_write_tx, mut ws_write_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
623 let (response_tx, response_rx) = tokio::sync::mpsc::unbounded_channel::<CommandMessage>();
624
625 // Background task: WS write loop
626 // Reads serialized messages from ws_write_rx and sends them over the WebSocket.
627 tokio::spawn(async move {
628 while let Some(msg_json) = ws_write_rx.recv().await {
629 if let Err(e) = write.send(Message::Text(msg_json)).await {
630 log::error!("WebSocket write error: {}", e);
631 break;
632 }
633 }
634 });
635
636 // Background task: WS read loop
637 // Reads all incoming WebSocket messages. Routes Response messages to
638 // response_tx for the CommandClient; ignores broadcasts and others.
639 tokio::spawn(async move {
640 while let Some(result) = read.next().await {
641 match result {
642 Ok(Message::Text(text)) => {
643 if let Ok(msg) = serde_json::from_str::<CommandMessage>(&text) {
644 if msg.message_type == MessageType::Response {
645 if response_tx.send(msg).is_err() {
646 break; // receiver dropped
647 }
648 }
649 // Broadcasts and other message types are ignored
650 }
651 }
652 Ok(Message::Close(_)) => {
653 log::info!("WebSocket closed by server");
654 break;
655 }
656 Err(e) => {
657 log::error!("WebSocket read error: {}", e);
658 break;
659 }
660 _ => {} // Ping/Pong/Binary - ignore
661 }
662 }
663 });
664
665 // Construct CommandClient — owned by the runner, passed to the
666 // program via TickContext each cycle.
667 let mut command_client = CommandClient::new(ws_write_tx.clone(), response_rx);
668
669 // 2. Find Signal Offsets
670 let tick_offset = self.find_offset(&layout, &self.config.tick_signal_name)?;
671 let busy_offset = if let Some(name) = &self.config.busy_signal_name {
672 Some(self.find_offset(&layout, name)?)
673 } else {
674 None
675 };
676
677 // 4. Open Shared Memory
678 let shmem = ShmemConf::new().os_id(&self.config.shm_name).open()?;
679 let base_ptr = shmem.as_ptr();
680 log::info!("Shared Memory '{}' mapped.", self.config.shm_name);
681
682 // Wait for the server to finish applying initial values before we
683 // read SHM. Without this, our startup read can race with the
684 // server's initial-value writes; we would then clobber any
685 // initials when we write local_mem back below.
686 if let Some(ready_info) = layout.get("__ready__") {
687 let ready_offset = ready_info.get("offset")
688 .and_then(|v| v.as_u64())
689 .ok_or_else(|| anyhow!("__ready__ layout entry has no offset"))? as usize;
690 let ready_ptr = unsafe { base_ptr.add(ready_offset) as *const u32 };
691 let start = std::time::Instant::now();
692 let timeout = Duration::from_secs(10);
693 loop {
694 let val = unsafe { std::ptr::read_volatile(ready_ptr) };
695 if val == 1 {
696 fence(Ordering::Acquire);
697 log::info!("Server ready flag observed after {:?}", start.elapsed());
698 break;
699 }
700 if start.elapsed() > timeout {
701 log::warn!("Timed out waiting for server ready flag; proceeding anyway (initials may be zeroed)");
702 break;
703 }
704 std::thread::sleep(Duration::from_millis(5));
705 }
706 } else {
707 log::warn!("No __ready__ flag in layout; server may predate ready-flag protocol. Initial values may race.");
708 }
709
710 // 5. Setup Pointers
711 // SAFETY: We trust the server's layout matches the generated GlobalMemory struct.
712 let gm = unsafe { &mut *(base_ptr as *mut P::Memory) };
713
714 // Get tick event from shared memory
715 log::info!("Setting up tick event at offset {} (base_ptr: {:p})", tick_offset, base_ptr);
716 let (tick_event, _) = unsafe {
717 Event::from_existing(base_ptr.add(tick_offset))
718 }.map_err(|e| anyhow!("Failed to open tick event: {:?}", e))?;
719 log::info!("Tick event ready");
720
721 // Busy signal event (optional)
722 let busy_event = busy_offset.map(|offset| {
723 unsafe { Event::from_existing(base_ptr.add(offset)) }
724 .map(|(event, _)| event)
725 .ok()
726 }).flatten();
727
728 // 6. Initialize local memory buffer and user program
729 // We use a local copy for the control loop to ensure:
730 // - Consistent snapshot of inputs at start of cycle
731 // - Atomic commit of outputs at end of cycle
732 // - Proper memory barriers for cross-process visibility
733 let mut local_mem: P::Memory = unsafe { std::ptr::read_volatile(gm) };
734 let mut prev_mem: P::Memory = local_mem; // Snapshot for change detection
735
736 fence(Ordering::Acquire); // Ensure we see all prior writes from other processes
737
738 self.program.initialize(&mut local_mem);
739
740 // Write back any changes from initialize
741 fence(Ordering::Release);
742 unsafe { std::ptr::write_volatile(gm, local_mem) };
743
744 // Set up signal handler for graceful shutdown
745 let running = Arc::new(AtomicBool::new(true));
746 let r = running.clone();
747
748 // Only set handler if not already set
749 if let Err(e) = ctrlc::set_handler(move || {
750 r.store(false, Ordering::SeqCst);
751 }) {
752 log::warn!("Failed to set signal handler: {}", e);
753 }
754
755 log::info!("Entering Control Loop - waiting for first tick...");
756 let mut cycle_count: u64 = 0;
757 let mut consecutive_timeouts: u32 = 0;
758
759 while running.load(Ordering::SeqCst) {
760 // Wait for Tick - Event-based synchronization
761 // Use a timeout (1s) to allow checking the running flag periodically
762 match tick_event.wait(Timeout::Val(Duration::from_secs(1))) {
763 Ok(_) => {
764 consecutive_timeouts = 0;
765 },
766 Err(e) => {
767 // Check for timeout
768 let err_str = format!("{:?}", e);
769 if err_str.contains("Timeout") {
770 consecutive_timeouts += 1;
771 if consecutive_timeouts == 10 {
772 log::error!(
773 "TICK STALL: {} consecutive timeouts! cycle={} pending={} responses={} fds={} rss_kb={}",
774 consecutive_timeouts,
775 cycle_count,
776 command_client.pending_count(),
777 command_client.response_count(),
778 diagnostics::count_open_fds(),
779 diagnostics::get_rss_kb(),
780 );
781 }
782 if consecutive_timeouts > 10 && consecutive_timeouts % 60 == 0 {
783 log::error!(
784 "TICK STALL continues: {} consecutive timeouts, cycle={}",
785 consecutive_timeouts,
786 cycle_count,
787 );
788 }
789 continue;
790 }
791 return Err(anyhow!("Tick wait failed: {:?}", e));
792 }
793 }
794
795 if !running.load(Ordering::SeqCst) {
796 log::info!("Shutdown signal received, exiting control loop.");
797 break;
798 }
799
800 cycle_count += 1;
801 if cycle_count == 1 {
802 log::info!("First tick received!");
803 }
804
805 // // Periodic diagnostics (every 30s at 100 Hz)
806 // if cycle_count % 3000 == 0 {
807 // log::info!(
808 // "DIAG cycle={} pending={} responses={} fds={} rss_kb={}",
809 // cycle_count,
810 // command_client.pending_count(),
811 // command_client.response_count(),
812 // diagnostics::count_open_fds(),
813 // diagnostics::get_rss_kb(),
814 // );
815 // }
816
817 // === INPUT PHASE ===
818 // Read all variables from shared memory into local buffer.
819 // This gives us a consistent snapshot of inputs for this cycle.
820 // Acquire fence ensures we see all writes from other processes (server, modules).
821 local_mem = unsafe { std::ptr::read_volatile(gm) };
822
823 // Update prev_mem before execution to track changes made IN THIS CYCLE
824 // Actually, we want to know what changed in SHM relative to what we last knew,
825 // OR what WE changed relative to what we read?
826 // The user wants "writes on shared variables" to be broadcast.
827 // Typically outputs.
828 // If inputs changed (from other source), broadcasting them again is fine too.
829 // Let's capture state BEFORE execution (which is what we just read from SHM).
830 prev_mem = local_mem;
831
832 fence(Ordering::Acquire);
833
834 // Unpack bit-mapped variables from their source words.
835 local_mem.unpack_bits();
836
837 // Snapshot after unpack — used by pack_bits to detect which
838 // bools the control program actually changed.
839 let pre_tick = local_mem;
840
841 // === EXECUTE PHASE ===
842 // Poll IPC responses so they are available during process_tick.
843 command_client.poll();
844
845 // Execute user logic on the local copy.
846 // All reads/writes during process_tick operate on local_mem.
847 let mut ctx = TickContext {
848 gm: &mut local_mem,
849 client: &mut command_client,
850 cycle: cycle_count,
851 };
852 self.program.process_tick(&mut ctx);
853
854 // === OUTPUT PHASE ===
855 // Pack bit-mapped variables back into their source words,
856 // but only for sources where a mapped bool actually changed.
857 local_mem.pack_bits(&pre_tick);
858
859 // Write all variables from local buffer back to shared memory.
860 // Release fence ensures our writes are visible to other processes.
861 fence(Ordering::Release);
862 unsafe { std::ptr::write_volatile(gm, local_mem) };
863
864 // === CHANGE DETECTION & NOTIFICATION ===
865 let changes = local_mem.get_changes(&prev_mem);
866 if !changes.is_empty() {
867 // Construct bulk write message
868 let mut data_map = serde_json::Map::new();
869 for (key, val) in changes {
870 data_map.insert(key.to_string(), val);
871 }
872
873 let msg = CommandMessage::request("gm.write", serde_json::Value::Object(data_map));
874 let msg_json = serde_json::to_string(&msg).unwrap_or_default();
875
876 // Send via the shared write channel (non-blocking)
877 if let Err(e) = ws_write_tx.send(msg_json) {
878 log::error!("Failed to send updates: {}", e);
879 }
880 }
881
882 // Signal Busy/Done event
883 if let Some(ref busy_ev) = busy_event {
884 let _ = busy_ev.set(EventState::Signaled);
885 }
886 }
887
888 Ok(())
889 })
890 }
891
892 fn find_offset(&self, layout: &HashMap<String, serde_json::Value>, name: &str) -> Result<usize> {
893 let info = layout.get(name).ok_or_else(|| anyhow!("Signal '{}' not found in layout", name))?;
894 info.get("offset")
895 .and_then(|v| v.as_u64())
896 .map(|v| v as usize)
897 .ok_or_else(|| anyhow!("Invalid offset for '{}'", name))
898 }
899}
900
901/// Generates the standard `main` function for a control program.
902///
903/// This macro reduces boilerplate by creating a properly configured `main`
904/// function that initializes and runs your control program.
905///
906/// # Arguments
907///
908/// * `$prog_type` - The type of your control program (must implement [`ControlProgram`])
909/// * `$shm_name` - The shared memory segment name (string literal)
910/// * `$tick_signal` - The tick signal name in shared memory (string literal)
911///
912/// # Example
913///
914/// ```ignore
915/// mod gm;
916/// use gm::GlobalMemory;
917///
918/// pub struct MyProgram;
919///
920/// impl MyProgram {
921/// pub fn new() -> Self { Self }
922/// }
923///
924/// impl autocore_std::ControlProgram for MyProgram {
925/// type Memory = GlobalMemory;
926///
927/// fn process_tick(&mut self, ctx: &mut autocore_std::TickContext<Self::Memory>) {
928/// // Your logic here
929/// }
930/// }
931///
932/// // This generates the main function
933/// autocore_std::autocore_main!(MyProgram, "my_project_shm", "tick");
934/// ```
935///
936/// # Generated Code
937///
938/// The macro expands to:
939///
940/// ```ignore
941/// fn main() -> anyhow::Result<()> {
942/// let config = autocore_std::RunnerConfig {
943/// server_host: "127.0.0.1".to_string(),
944/// ws_port: autocore_std::DEFAULT_WS_PORT,
945/// module_name: "control".to_string(),
946/// shm_name: "my_project_shm".to_string(),
947/// tick_signal_name: "tick".to_string(),
948/// busy_signal_name: None,
949/// log_level: log::LevelFilter::Info,
950/// log_udp_port: autocore_std::logger::DEFAULT_LOG_UDP_PORT,
951/// };
952///
953/// autocore_std::ControlRunner::new(MyProgram::new())
954/// .config(config)
955/// .run()
956/// }
957/// ```
958#[macro_export]
959macro_rules! autocore_main {
960 ($prog_type:ty, $shm_name:expr, $tick_signal:expr) => {
961 fn main() -> anyhow::Result<()> {
962 let config = autocore_std::RunnerConfig {
963 server_host: "127.0.0.1".to_string(),
964 ws_port: autocore_std::DEFAULT_WS_PORT,
965 module_name: "control".to_string(),
966 shm_name: $shm_name.to_string(),
967 tick_signal_name: $tick_signal.to_string(),
968 busy_signal_name: None,
969 log_level: log::LevelFilter::Info,
970 log_udp_port: autocore_std::logger::DEFAULT_LOG_UDP_PORT,
971 };
972
973 autocore_std::ControlRunner::new(<$prog_type>::new())
974 .config(config)
975 .run()
976 }
977 };
978}
979