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