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/// Subscription helper for `ams.asset_updated.<location>` broadcasts.
216/// Control programs use this to drive EL3356 SDO writes (and similar)
217/// when a load cell asset is registered, recalibrated, or retired.
218pub mod asset_watch;
219pub use asset_watch::{AssetUpdate, AssetWatch, AssetWatchStatus, AssetWatchTrigger};
220
221/// EtherCAT utilities (SDO client, etc.).
222pub mod ethercat;
223
224/// CiA 402 motion control: axis abstraction, traits, and types.
225pub mod motion;
226
227/// Shared memory utilities for external modules.
228pub mod shm;
229
230/// Lightweight process diagnostics (FD count, RSS).
231pub mod diagnostics;
232
233/// Banner Engineering device helpers (WLS15 IO-Link light strip, etc.).
234pub mod banner;
235
236/// Fixed-length string type for shared memory variables.
237pub mod fixed_string;
238pub use fixed_string::FixedString;
239
240// ============================================================================
241// Core Framework
242// ============================================================================
243
244/// Marker trait for generated GlobalMemory structs.
245///
246/// This trait is implemented by the auto-generated `GlobalMemory` struct
247/// that represents the shared memory layout. It serves as a marker for
248/// type safety in the control framework.
249///
250/// You don't need to implement this trait yourself - it's automatically
251/// implemented by the code generator.
252pub trait AutoCoreMemory {}
253
254/// Trait for detecting changes in memory structures.
255pub trait ChangeTracker {
256 /// Compare self with a previous state and return a list of changed fields.
257 /// Returns a vector of (field_name, new_value).
258 fn get_changes(&self, prev: &Self) -> Vec<(&'static str, serde_json::Value)>;
259
260 /// Unpack bit-mapped variables from their source words.
261 /// Called automatically after reading shared memory, before `process_tick`.
262 /// Auto-generated by codegen when bit-mapped variables exist; default is no-op.
263 fn unpack_bits(&mut self) {}
264
265 /// Pack bit-mapped variables back into their source words.
266 /// Called automatically after `process_tick`, before writing shared memory.
267 /// Only packs sources where at least one mapped bool changed since `pre_tick`.
268 /// Auto-generated by codegen when bit-mapped variables exist; default is no-op.
269 fn pack_bits(&mut self, _pre_tick: &Self) {}
270}
271
272/// Per-tick context passed to the control program by the framework.
273///
274/// `TickContext` bundles all per-cycle data into a single struct so that the
275/// [`ControlProgram::process_tick`] signature stays stable as new fields are
276/// added in the future (e.g., delta time, diagnostics).
277///
278/// The framework constructs a fresh `TickContext` each cycle, calls
279/// [`CommandClient::poll`] before handing it to the program, and writes
280/// the memory back to shared memory after `process_tick` returns.
281pub struct TickContext<'a, M> {
282 /// Mutable reference to the local shared memory copy.
283 pub gm: &'a mut M,
284 /// IPC command client for communicating with external modules.
285 pub client: &'a mut CommandClient,
286 /// Current cycle number (starts at 1, increments each tick).
287 pub cycle: u64,
288}
289
290/// The trait that defines a control program's logic.
291///
292/// Implement this trait to create your control program. The associated `Memory`
293/// type should be the generated `GlobalMemory` struct from your project.
294///
295/// # Memory Type Requirements
296///
297/// The `Memory` type must implement `Copy` to allow efficient synchronization
298/// between shared memory and local buffers. This is automatically satisfied
299/// by the generated `GlobalMemory` struct.
300///
301/// # Lifecycle
302///
303/// 1. `initialize` is called once at startup
304/// 2. `process_tick` is called repeatedly in the control loop with a
305/// [`TickContext`] that provides shared memory, the IPC client, and the
306/// current cycle number.
307///
308/// # Example
309///
310/// ```ignore
311/// use autocore_std::{ControlProgram, TickContext};
312///
313/// mod gm;
314/// use gm::GlobalMemory;
315///
316/// pub struct MyController {
317/// cycle_counter: u64,
318/// }
319///
320/// impl MyController {
321/// pub fn new() -> Self {
322/// Self { cycle_counter: 0 }
323/// }
324/// }
325///
326/// impl ControlProgram for MyController {
327/// type Memory = GlobalMemory;
328///
329/// fn initialize(&mut self, mem: &mut GlobalMemory) {
330/// // Set initial output states
331/// mem.outputs.ready = true;
332/// log::info!("Controller initialized");
333/// }
334///
335/// fn process_tick(&mut self, ctx: &mut TickContext<Self::Memory>) {
336/// self.cycle_counter = ctx.cycle;
337///
338/// // Your control logic here
339/// if ctx.gm.inputs.start && !ctx.gm.inputs.estop {
340/// ctx.gm.outputs.running = true;
341/// }
342/// }
343/// }
344/// ```
345pub trait ControlProgram {
346 /// The shared memory structure type (usually the generated `GlobalMemory`).
347 ///
348 /// Must implement `Copy` to allow efficient memory synchronization.
349 type Memory: Copy + ChangeTracker;
350
351 /// Called once when the control program starts.
352 ///
353 /// Use this to initialize output states, reset counters, or perform
354 /// any one-time setup. The default implementation does nothing.
355 ///
356 /// # Arguments
357 ///
358 /// * `mem` - Mutable reference to the shared memory. Changes are written
359 /// back to shared memory after this method returns.
360 fn initialize(&mut self, _mem: &mut Self::Memory) {}
361
362 /// The main control loop - called once per scan cycle.
363 ///
364 /// This is where your control logic lives. Read inputs from `ctx.gm`,
365 /// perform calculations, and write outputs back to `ctx.gm`. Use
366 /// `ctx.client` for IPC commands and `ctx.cycle` for the current cycle
367 /// number.
368 ///
369 /// The framework calls [`CommandClient::poll`] before each invocation,
370 /// so incoming responses are already buffered when your code runs.
371 ///
372 /// # Arguments
373 ///
374 /// * `ctx` - A [`TickContext`] containing the local shared memory copy,
375 /// the IPC command client, and the current cycle number.
376 ///
377 /// # Timing
378 ///
379 /// This method should complete within the scan cycle time. Long-running
380 /// operations will cause cycle overruns.
381 fn process_tick(&mut self, ctx: &mut TickContext<Self::Memory>);
382}
383
384/// Configuration for the [`ControlRunner`].
385///
386/// Specifies connection parameters, shared memory names, and logging settings.
387/// Use [`Default::default()`] for typical configurations.
388///
389/// # Example
390///
391/// ```
392/// use autocore_std::RunnerConfig;
393/// use log::LevelFilter;
394///
395/// let config = RunnerConfig {
396/// server_host: "192.168.1.100".to_string(),
397/// module_name: "my_controller".to_string(),
398/// shm_name: "my_project_shm".to_string(),
399/// tick_signal_name: "tick".to_string(),
400/// busy_signal_name: Some("busy".to_string()),
401/// log_level: LevelFilter::Debug,
402/// ..Default::default()
403/// };
404/// ```
405#[derive(Debug, Clone)]
406pub struct RunnerConfig {
407 /// Server host address (default: "127.0.0.1")
408 pub server_host: String,
409 /// WebSocket port for commands (default: 11969)
410 pub ws_port: u16,
411 /// Module name for identification (default: "control")
412 pub module_name: String,
413 /// Shared memory segment name (must match server configuration)
414 pub shm_name: String,
415 /// Name of the tick signal in shared memory (triggers each scan cycle)
416 pub tick_signal_name: String,
417 /// Optional name of the busy signal (set when cycle completes)
418 pub busy_signal_name: Option<String>,
419 /// Minimum log level to send to the server (default: Info)
420 pub log_level: LevelFilter,
421 /// UDP port for sending logs to the server (default: 39101)
422 pub log_udp_port: u16,
423}
424
425/// Default WebSocket port for autocore-server
426pub const DEFAULT_WS_PORT: u16 = 11969;
427
428impl Default for RunnerConfig {
429 fn default() -> Self {
430 Self {
431 server_host: "127.0.0.1".to_string(),
432 ws_port: DEFAULT_WS_PORT,
433 module_name: "control".to_string(),
434 shm_name: "autocore_cyclic".to_string(),
435 tick_signal_name: "tick".to_string(),
436 busy_signal_name: None,
437 log_level: LevelFilter::Info,
438 log_udp_port: logger::DEFAULT_LOG_UDP_PORT,
439 }
440 }
441}
442
443
444/// The main execution engine for control programs.
445///
446/// `ControlRunner` handles all the infrastructure required to run a control program:
447///
448/// - Reading memory layout from the server's layout file
449/// - Opening and mapping shared memory
450/// - Setting up synchronization signals
451/// - Running the real-time control loop
452/// - Sending log messages to the server
453///
454/// # Usage
455///
456/// ```ignore
457/// use autocore_std::{ControlRunner, RunnerConfig};
458///
459/// let config = RunnerConfig {
460/// shm_name: "my_project_shm".to_string(),
461/// tick_signal_name: "tick".to_string(),
462/// ..Default::default()
463/// };
464///
465/// ControlRunner::new(MyProgram::new())
466/// .config(config)
467/// .run()?; // Blocks forever
468/// ```
469///
470/// # Control Loop
471///
472/// The runner executes a synchronous control loop:
473///
474/// 1. **Wait** - Blocks until the tick signal is set by the server
475/// 2. **Read** - Copies shared memory to a local buffer (acquire barrier)
476/// 3. **Execute** - Calls your `process_tick` method
477/// 4. **Write** - Copies local buffer back to shared memory (release barrier)
478/// 5. **Signal** - Sets the busy signal (if configured) to indicate completion
479///
480/// This ensures your code always sees a consistent snapshot of the data
481/// and that your writes are atomically visible to other processes.
482pub struct ControlRunner<P: ControlProgram> {
483 config: RunnerConfig,
484 program: P,
485}
486
487impl<P: ControlProgram> ControlRunner<P> {
488 /// Creates a new runner for the given control program.
489 ///
490 /// Uses default configuration. Call [`.config()`](Self::config) to customize.
491 ///
492 /// # Arguments
493 ///
494 /// * `program` - Your control program instance
495 ///
496 /// # Example
497 ///
498 /// ```ignore
499 /// let runner = ControlRunner::new(MyProgram::new());
500 /// ```
501 pub fn new(program: P) -> Self {
502 Self {
503 config: RunnerConfig::default(),
504 program,
505 }
506 }
507
508 /// Sets the configuration for this runner.
509 ///
510 /// # Arguments
511 ///
512 /// * `config` - The configuration to use
513 ///
514 /// # Example
515 ///
516 /// ```ignore
517 /// ControlRunner::new(MyProgram::new())
518 /// .config(RunnerConfig {
519 /// shm_name: "custom_shm".to_string(),
520 /// ..Default::default()
521 /// })
522 /// .run()?;
523 /// ```
524 pub fn config(mut self, config: RunnerConfig) -> Self {
525 self.config = config;
526 self
527 }
528
529 /// Starts the control loop.
530 ///
531 /// This method blocks indefinitely, running the control loop until
532 /// an error occurs or the process is terminated.
533 ///
534 /// # Returns
535 ///
536 /// Returns `Ok(())` only if the loop exits cleanly (which typically
537 /// doesn't happen). Returns an error if:
538 ///
539 /// - IPC connection fails
540 /// - Shared memory cannot be opened
541 /// - Signal offsets cannot be found
542 /// - A critical error occurs during execution
543 ///
544 /// # Example
545 ///
546 /// ```ignore
547 /// fn main() -> anyhow::Result<()> {
548 /// ControlRunner::new(MyProgram::new())
549 /// .config(config)
550 /// .run()
551 /// }
552 /// ```
553 pub fn run(mut self) -> Result<()> {
554 // Initialize UDP logger FIRST (before any log statements)
555 if let Err(e) = logger::init_udp_logger(
556 &self.config.server_host,
557 self.config.log_udp_port,
558 self.config.log_level,
559 "control",
560 ) {
561 eprintln!("Warning: Failed to initialize UDP logger: {}", e);
562 // Continue anyway - logging will just go nowhere
563 }
564
565 // Multi-threaded runtime so spawned WS read/write tasks can run
566 // alongside the synchronous control loop.
567 let rt = tokio::runtime::Builder::new_multi_thread()
568 .worker_threads(2)
569 .enable_all()
570 .build()?;
571
572 rt.block_on(async {
573 log::info!("AutoCore Control Runner Starting...");
574
575 // 1. Connect to server via WebSocket and get layout
576 let ws_url = format!("ws://{}:{}/ws/", self.config.server_host, self.config.ws_port);
577 log::info!("Connecting to server at {}", ws_url);
578
579 let (ws_stream, _) = connect_async(&ws_url).await
580 .map_err(|e| anyhow!("Failed to connect to server at {}: {}", ws_url, e))?;
581
582 let (mut write, mut read) = ws_stream.split();
583
584 // Send gm.get_layout request
585 let request = CommandMessage::request("gm.get_layout", serde_json::Value::Null);
586 let transaction_id = request.transaction_id;
587 let request_json = serde_json::to_string(&request)?;
588
589 write.send(Message::Text(request_json)).await
590 .map_err(|e| anyhow!("Failed to send layout request: {}", e))?;
591
592 // Wait for response with matching transaction_id
593 let timeout = Duration::from_secs(10);
594 let start = std::time::Instant::now();
595 let mut layout: Option<HashMap<String, serde_json::Value>> = None;
596
597 while start.elapsed() < timeout {
598 match tokio::time::timeout(Duration::from_secs(1), read.next()).await {
599 Ok(Some(Ok(Message::Text(text)))) => {
600 if let Ok(response) = serde_json::from_str::<CommandMessage>(&text) {
601 if response.transaction_id == transaction_id {
602 if !response.success {
603 return Err(anyhow!("Server error: {}", response.error_message));
604 }
605 layout = Some(serde_json::from_value(response.data)?);
606 break;
607 }
608 // Skip broadcasts and other messages
609 if response.message_type == MessageType::Broadcast {
610 continue;
611 }
612 }
613 }
614 Ok(Some(Ok(_))) => continue,
615 Ok(Some(Err(e))) => return Err(anyhow!("WebSocket error: {}", e)),
616 Ok(None) => return Err(anyhow!("Server closed connection")),
617 Err(_) => continue, // Timeout on single read, keep trying
618 }
619 }
620
621 let layout = layout.ok_or_else(|| anyhow!("Timeout waiting for layout response"))?;
622 log::info!("Layout received with {} entries.", layout.len());
623
624 // Set up channels and background tasks for shared WebSocket access.
625 // This allows both the control loop (gm.write) and CommandClient (IPC
626 // commands) to share the write half, while routing incoming responses
627 // to the CommandClient.
628 let (ws_write_tx, mut ws_write_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
629 let (response_tx, response_rx) = tokio::sync::mpsc::unbounded_channel::<CommandMessage>();
630
631 // Background task: WS write loop
632 // Reads serialized messages from ws_write_rx and sends them over the WebSocket.
633 tokio::spawn(async move {
634 while let Some(msg_json) = ws_write_rx.recv().await {
635 if let Err(e) = write.send(Message::Text(msg_json)).await {
636 log::error!("WebSocket write error: {}", e);
637 break;
638 }
639 }
640 });
641
642 // Background task: WS read loop
643 // Reads all incoming WebSocket messages. Routes Response and
644 // Broadcast messages to response_tx; CommandClient dispatches
645 // them onward (responses by transaction_id, broadcasts by
646 // topic into per-topic buffers that subscribers drain via
647 // `take_broadcasts`).
648 tokio::spawn(async move {
649 while let Some(result) = read.next().await {
650 match result {
651 Ok(Message::Text(text)) => {
652 if let Ok(msg) = serde_json::from_str::<CommandMessage>(&text) {
653 if matches!(
654 msg.message_type,
655 MessageType::Response | MessageType::Broadcast,
656 ) {
657 if response_tx.send(msg).is_err() {
658 break; // receiver dropped
659 }
660 }
661 // Other message types are ignored
662 }
663 }
664 Ok(Message::Close(_)) => {
665 log::info!("WebSocket closed by server");
666 break;
667 }
668 Err(e) => {
669 log::error!("WebSocket read error: {}", e);
670 break;
671 }
672 _ => {} // Ping/Pong/Binary - ignore
673 }
674 }
675 });
676
677 // Construct CommandClient — owned by the runner, passed to the
678 // program via TickContext each cycle.
679 let mut command_client = CommandClient::new(ws_write_tx.clone(), response_rx);
680
681 // 2. Find Signal Offsets
682 let tick_offset = self.find_offset(&layout, &self.config.tick_signal_name)?;
683 let busy_offset = if let Some(name) = &self.config.busy_signal_name {
684 Some(self.find_offset(&layout, name)?)
685 } else {
686 None
687 };
688
689 // 4. Open Shared Memory
690 let shmem = ShmemConf::new().os_id(&self.config.shm_name).open()?;
691 let base_ptr = shmem.as_ptr();
692 log::info!("Shared Memory '{}' mapped.", self.config.shm_name);
693
694 // Wait for the server to finish applying initial values before we
695 // read SHM. Without this, our startup read can race with the
696 // server's initial-value writes; we would then clobber any
697 // initials when we write local_mem back below.
698 if let Some(ready_info) = layout.get("__ready__") {
699 let ready_offset = ready_info.get("offset")
700 .and_then(|v| v.as_u64())
701 .ok_or_else(|| anyhow!("__ready__ layout entry has no offset"))? as usize;
702 let ready_ptr = unsafe { base_ptr.add(ready_offset) as *const u32 };
703 let start = std::time::Instant::now();
704 let timeout = Duration::from_secs(10);
705 loop {
706 let val = unsafe { std::ptr::read_volatile(ready_ptr) };
707 if val == 1 {
708 fence(Ordering::Acquire);
709 log::info!("Server ready flag observed after {:?}", start.elapsed());
710 break;
711 }
712 if start.elapsed() > timeout {
713 log::warn!("Timed out waiting for server ready flag; proceeding anyway (initials may be zeroed)");
714 break;
715 }
716 std::thread::sleep(Duration::from_millis(5));
717 }
718 } else {
719 log::warn!("No __ready__ flag in layout; server may predate ready-flag protocol. Initial values may race.");
720 }
721
722 // 5. Setup Pointers
723 // SAFETY: We trust the server's layout matches the generated GlobalMemory struct.
724 let gm = unsafe { &mut *(base_ptr as *mut P::Memory) };
725
726 // Get tick event from shared memory
727 log::info!("Setting up tick event at offset {} (base_ptr: {:p})", tick_offset, base_ptr);
728 let (tick_event, _) = unsafe {
729 Event::from_existing(base_ptr.add(tick_offset))
730 }.map_err(|e| anyhow!("Failed to open tick event: {:?}", e))?;
731 log::info!("Tick event ready");
732
733 // Busy signal event (optional)
734 let busy_event = busy_offset.map(|offset| {
735 unsafe { Event::from_existing(base_ptr.add(offset)) }
736 .map(|(event, _)| event)
737 .ok()
738 }).flatten();
739
740 // 6. Initialize local memory buffer and user program
741 // We use a local copy for the control loop to ensure:
742 // - Consistent snapshot of inputs at start of cycle
743 // - Atomic commit of outputs at end of cycle
744 // - Proper memory barriers for cross-process visibility
745 let mut local_mem: P::Memory = unsafe { std::ptr::read_volatile(gm) };
746 let mut prev_mem: P::Memory = local_mem; // Snapshot for change detection
747
748 fence(Ordering::Acquire); // Ensure we see all prior writes from other processes
749
750 self.program.initialize(&mut local_mem);
751
752 // Write back any changes from initialize
753 fence(Ordering::Release);
754 unsafe { std::ptr::write_volatile(gm, local_mem) };
755
756 // Set up signal handler for graceful shutdown
757 let running = Arc::new(AtomicBool::new(true));
758 let r = running.clone();
759
760 // Only set handler if not already set
761 if let Err(e) = ctrlc::set_handler(move || {
762 r.store(false, Ordering::SeqCst);
763 }) {
764 log::warn!("Failed to set signal handler: {}", e);
765 }
766
767 log::info!("Entering Control Loop - waiting for first tick...");
768 let mut cycle_count: u64 = 0;
769 let mut consecutive_timeouts: u32 = 0;
770
771 while running.load(Ordering::SeqCst) {
772 // Wait for Tick - Event-based synchronization
773 // Use a timeout (1s) to allow checking the running flag periodically
774 match tick_event.wait(Timeout::Val(Duration::from_secs(1))) {
775 Ok(_) => {
776 consecutive_timeouts = 0;
777 },
778 Err(e) => {
779 // Check for timeout
780 let err_str = format!("{:?}", e);
781 if err_str.contains("Timeout") {
782 consecutive_timeouts += 1;
783 if consecutive_timeouts == 10 {
784 log::error!(
785 "TICK STALL: {} consecutive timeouts! cycle={} pending={} responses={} fds={} rss_kb={}",
786 consecutive_timeouts,
787 cycle_count,
788 command_client.pending_count(),
789 command_client.response_count(),
790 diagnostics::count_open_fds(),
791 diagnostics::get_rss_kb(),
792 );
793 }
794 if consecutive_timeouts > 10 && consecutive_timeouts % 60 == 0 {
795 log::error!(
796 "TICK STALL continues: {} consecutive timeouts, cycle={}",
797 consecutive_timeouts,
798 cycle_count,
799 );
800 }
801 continue;
802 }
803 return Err(anyhow!("Tick wait failed: {:?}", e));
804 }
805 }
806
807 if !running.load(Ordering::SeqCst) {
808 log::info!("Shutdown signal received, exiting control loop.");
809 break;
810 }
811
812 cycle_count += 1;
813 if cycle_count == 1 {
814 log::info!("First tick received!");
815 }
816
817 // // Periodic diagnostics (every 30s at 100 Hz)
818 // if cycle_count % 3000 == 0 {
819 // log::info!(
820 // "DIAG cycle={} pending={} responses={} fds={} rss_kb={}",
821 // cycle_count,
822 // command_client.pending_count(),
823 // command_client.response_count(),
824 // diagnostics::count_open_fds(),
825 // diagnostics::get_rss_kb(),
826 // );
827 // }
828
829 // === INPUT PHASE ===
830 // Read all variables from shared memory into local buffer.
831 // This gives us a consistent snapshot of inputs for this cycle.
832 // Acquire fence ensures we see all writes from other processes (server, modules).
833 local_mem = unsafe { std::ptr::read_volatile(gm) };
834
835 // Update prev_mem before execution to track changes made IN THIS CYCLE
836 // Actually, we want to know what changed in SHM relative to what we last knew,
837 // OR what WE changed relative to what we read?
838 // The user wants "writes on shared variables" to be broadcast.
839 // Typically outputs.
840 // If inputs changed (from other source), broadcasting them again is fine too.
841 // Let's capture state BEFORE execution (which is what we just read from SHM).
842 prev_mem = local_mem;
843
844 fence(Ordering::Acquire);
845
846 // Unpack bit-mapped variables from their source words.
847 local_mem.unpack_bits();
848
849 // Snapshot after unpack — used by pack_bits to detect which
850 // bools the control program actually changed.
851 let pre_tick = local_mem;
852
853 // === EXECUTE PHASE ===
854 // Poll IPC responses so they are available during process_tick.
855 command_client.poll();
856
857 // Execute user logic on the local copy.
858 // All reads/writes during process_tick operate on local_mem.
859 let mut ctx = TickContext {
860 gm: &mut local_mem,
861 client: &mut command_client,
862 cycle: cycle_count,
863 };
864 self.program.process_tick(&mut ctx);
865
866 // === OUTPUT PHASE ===
867 // Pack bit-mapped variables back into their source words,
868 // but only for sources where a mapped bool actually changed.
869 local_mem.pack_bits(&pre_tick);
870
871 // Write all variables from local buffer back to shared memory.
872 // Release fence ensures our writes are visible to other processes.
873 fence(Ordering::Release);
874 unsafe { std::ptr::write_volatile(gm, local_mem) };
875
876 // === CHANGE DETECTION & NOTIFICATION ===
877 let changes = local_mem.get_changes(&prev_mem);
878 if !changes.is_empty() {
879 // Construct bulk write message
880 let mut data_map = serde_json::Map::new();
881 for (key, val) in changes {
882 data_map.insert(key.to_string(), val);
883 }
884
885 let msg = CommandMessage::request("gm.write", serde_json::Value::Object(data_map));
886 let msg_json = serde_json::to_string(&msg).unwrap_or_default();
887
888 // Send via the shared write channel (non-blocking)
889 if let Err(e) = ws_write_tx.send(msg_json) {
890 log::error!("Failed to send updates: {}", e);
891 }
892 }
893
894 // Signal Busy/Done event
895 if let Some(ref busy_ev) = busy_event {
896 let _ = busy_ev.set(EventState::Signaled);
897 }
898 }
899
900 Ok(())
901 })
902 }
903
904 fn find_offset(&self, layout: &HashMap<String, serde_json::Value>, name: &str) -> Result<usize> {
905 let info = layout.get(name).ok_or_else(|| anyhow!("Signal '{}' not found in layout", name))?;
906 info.get("offset")
907 .and_then(|v| v.as_u64())
908 .map(|v| v as usize)
909 .ok_or_else(|| anyhow!("Invalid offset for '{}'", name))
910 }
911}
912
913/// Generates the standard `main` function for a control program.
914///
915/// This macro reduces boilerplate by creating a properly configured `main`
916/// function that initializes and runs your control program.
917///
918/// # Arguments
919///
920/// * `$prog_type` - The type of your control program (must implement [`ControlProgram`])
921/// * `$shm_name` - The shared memory segment name (string literal)
922/// * `$tick_signal` - The tick signal name in shared memory (string literal)
923///
924/// # Example
925///
926/// ```ignore
927/// mod gm;
928/// use gm::GlobalMemory;
929///
930/// pub struct MyProgram;
931///
932/// impl MyProgram {
933/// pub fn new() -> Self { Self }
934/// }
935///
936/// impl autocore_std::ControlProgram for MyProgram {
937/// type Memory = GlobalMemory;
938///
939/// fn process_tick(&mut self, ctx: &mut autocore_std::TickContext<Self::Memory>) {
940/// // Your logic here
941/// }
942/// }
943///
944/// // This generates the main function
945/// autocore_std::autocore_main!(MyProgram, "my_project_shm", "tick");
946/// ```
947///
948/// # Generated Code
949///
950/// The macro expands to:
951///
952/// ```ignore
953/// fn main() -> anyhow::Result<()> {
954/// let config = autocore_std::RunnerConfig {
955/// server_host: "127.0.0.1".to_string(),
956/// ws_port: autocore_std::DEFAULT_WS_PORT,
957/// module_name: "control".to_string(),
958/// shm_name: "my_project_shm".to_string(),
959/// tick_signal_name: "tick".to_string(),
960/// busy_signal_name: None,
961/// log_level: log::LevelFilter::Info,
962/// log_udp_port: autocore_std::logger::DEFAULT_LOG_UDP_PORT,
963/// };
964///
965/// autocore_std::ControlRunner::new(MyProgram::new())
966/// .config(config)
967/// .run()
968/// }
969/// ```
970#[macro_export]
971macro_rules! autocore_main {
972 ($prog_type:ty, $shm_name:expr, $tick_signal:expr) => {
973 fn main() -> anyhow::Result<()> {
974 let config = autocore_std::RunnerConfig {
975 server_host: "127.0.0.1".to_string(),
976 ws_port: autocore_std::DEFAULT_WS_PORT,
977 module_name: "control".to_string(),
978 shm_name: $shm_name.to_string(),
979 tick_signal_name: $tick_signal.to_string(),
980 busy_signal_name: None,
981 log_level: log::LevelFilter::Info,
982 log_udp_port: autocore_std::logger::DEFAULT_LOG_UDP_PORT,
983 };
984
985 autocore_std::ControlRunner::new(<$prog_type>::new())
986 .config(config)
987 .run()
988 }
989 };
990}
991