rlg/
engine.rs

1// engine.rs
2// Copyright © 2024-2026 RustLogs (RLG). All rights reserved.
3// SPDX-License-Identifier: Apache-2.0
4// SPDX-License-Identifier: MIT
5
6//! Near-lock-free ingestion engine backed by a bounded ring buffer.
7//!
8//! The global [`ENGINE`][crate::engine::ENGINE] accepts
9//! [`LogEvent`][crate::engine::LogEvent]s via
10//! [`LockFreeEngine::ingest()`][crate::engine::LockFreeEngine::ingest]
11//! using only atomic operations. A dedicated background thread drains events
12//! in batches of 64 and writes them through [`PlatformSink`](crate::sink::PlatformSink).
13//!
14//! **The Mutex is never locked on the hot path.** It exists solely for
15//! `shutdown()` to join the flusher thread.
16
17use crate::log_level::LogLevel;
18use crate::sharded_queue::ShardedQueue;
19#[cfg(not(miri))]
20use crate::sink::PlatformSink;
21use crate::tui::TuiMetrics;
22#[cfg(not(miri))]
23use crate::tui::spawn_tui_thread;
24use std::fmt;
25use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
26use std::sync::{Arc, LazyLock, Mutex};
27use std::thread;
28#[cfg(not(miri))]
29use std::time::Duration;
30
31/// Capacity of the lock-free ring buffer (number of log events).
32const RING_BUFFER_CAPACITY: usize = 65_536;
33
34/// Maximum number of events drained per flusher wake-up cycle.
35#[cfg(not(miri))]
36const MAX_DRAIN_BATCH_SIZE: usize = 64;
37
38/// A structured log event passed through the ring buffer.
39///
40/// The caller pays only for a `Log` move (~128-byte memcpy).
41/// Serialization happens on the flusher thread.
42#[derive(Debug, Clone)]
43pub struct LogEvent {
44    /// Severity level of this event.
45    pub level: LogLevel,
46    /// Numeric severity for fast level-gating comparisons.
47    pub level_num: u8,
48    /// Structured log data. Formatted on the flusher thread, not here.
49    pub log: crate::log::Log,
50}
51
52/// The near-lock-free ingestion engine.
53///
54/// Owns the ring buffer, flusher thread, and TUI metrics counters.
55/// Access the global instance via [`ENGINE`].
56pub struct LockFreeEngine {
57    /// Bounded, sharded ring buffer.
58    ///
59    /// - Default build: one shard — semantically identical to the
60    ///   direct `ArrayQueue` use in prior releases.
61    /// - `fast-queue` feature: eight shards — reduces producer-side
62    ///   cache-line contention on the underlying atomic tag when
63    ///   many threads ingest concurrently.
64    ///
65    /// See `docs/adr/0009-sharded-producer-queue.md`.
66    queue: Arc<ShardedQueue>,
67    /// Signals the flusher thread to drain and exit.
68    shutdown_flag: Arc<AtomicBool>,
69    /// Atomic counters consumed by the opt-in TUI dashboard.
70    metrics: Arc<TuiMetrics>,
71    /// Minimum severity level. Events below this are dropped at `ingest()`.
72    filter_level: AtomicU8,
73    /// Flusher thread handle for lock-free `unpark()`. No Mutex involved.
74    flusher_thread_handle: Option<thread::Thread>,
75    /// `JoinHandle` for `shutdown()` only. **Never locked on the hot path.**
76    flusher_join: Mutex<Option<thread::JoinHandle<()>>>,
77}
78
79impl fmt::Debug for LockFreeEngine {
80    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
81        f.debug_struct("LockFreeEngine")
82            .field("queue", &self.queue)
83            .field("shutdown_flag", &self.shutdown_flag)
84            .field("metrics", &self.metrics)
85            .field("filter_level", &self.filter_level)
86            .field(
87                "flusher_thread_handle",
88                &self
89                    .flusher_thread_handle
90                    .as_ref()
91                    .map(thread::Thread::id),
92            )
93            .finish_non_exhaustive()
94    }
95}
96
97/// Global engine instance, lazily initialized on first access.
98pub static ENGINE: LazyLock<LockFreeEngine> =
99    LazyLock::new(|| LockFreeEngine::new(RING_BUFFER_CAPACITY));
100
101impl LockFreeEngine {
102    /// Create a new engine with the given buffer capacity and spawn the flusher.
103    ///
104    /// # Panics
105    ///
106    /// Panics if the OS cannot spawn the background flusher thread.
107    #[must_use]
108    pub fn new(capacity: usize) -> Self {
109        let queue = Arc::new(ShardedQueue::new(capacity));
110        let shutdown_flag = Arc::new(AtomicBool::new(false));
111        let metrics = Arc::new(TuiMetrics::default());
112        let filter_level = AtomicU8::new(0); // Default to ALL
113
114        // Under MIRI, skip spawning background threads to avoid
115        // "main thread terminated without waiting" errors.
116        #[cfg(not(miri))]
117        let flusher_handle = {
118            let flusher_queue = queue.clone();
119            let flusher_shutdown = shutdown_flag.clone();
120
121            // Spawn lightweight OS thread (Runtime Agnostic)
122            let handle = thread::Builder::new()
123                .name("rlg-flusher".into())
124                .spawn(move || {
125                    use std::io::Write;
126                    let mut sink = PlatformSink::native();
127                    let mut fmt_buf = Vec::with_capacity(512);
128
129                    loop {
130                        let mut batch: [Option<LogEvent>;
131                            MAX_DRAIN_BATCH_SIZE] =
132                            std::array::from_fn(|_| None);
133                        let mut count = 0;
134                        while count < MAX_DRAIN_BATCH_SIZE {
135                            match flusher_queue.pop() {
136                                Some(event) => {
137                                    batch[count] = Some(event);
138                                    count += 1;
139                                }
140                                None => break,
141                            }
142                        }
143                        for event in batch.iter().flatten() {
144                            fmt_buf.clear();
145                            let _ = writeln!(fmt_buf, "{}", &event.log);
146                            sink.emit(event.level.as_str(), &fmt_buf);
147                        }
148
149                        if flusher_shutdown.load(Ordering::Relaxed)
150                            && flusher_queue.is_empty()
151                        {
152                            break;
153                        }
154
155                        // Park briefly as fallback; real wakeup comes from unpark() in ingest().
156                        thread::park_timeout(Duration::from_millis(5));
157                    }
158                })
159                .expect(
160                    "Failed to spawn rlg-flusher background thread",
161                );
162
163            // Spawn the TUI dashboard thread if RLG_TUI=1
164            if std::env::var("RLG_TUI").is_ok_and(|v| v == "1") {
165                spawn_tui_thread(
166                    metrics.clone(),
167                    shutdown_flag.clone(),
168                );
169            }
170
171            Some(handle)
172        };
173
174        #[cfg(miri)]
175        let flusher_handle: Option<thread::JoinHandle<()>> = None;
176
177        let flusher_thread_handle =
178            flusher_handle.as_ref().map(|h| h.thread().clone());
179
180        Self {
181            queue,
182            shutdown_flag,
183            metrics,
184            filter_level,
185            flusher_thread_handle,
186            flusher_join: Mutex::new(flusher_handle),
187        }
188    }
189
190    /// Appends an event to the ring buffer.
191    ///
192    /// If the buffer is full, the oldest event is evicted to make room.
193    /// Dropped events are tracked via `TuiMetrics::dropped_events`.
194    pub fn ingest(&self, event: LogEvent) {
195        if event.level_num < self.filter_level.load(Ordering::Acquire) {
196            return;
197        }
198
199        self.metrics.inc_events();
200        self.metrics.inc_level(event.level);
201
202        if event.level_num >= LogLevel::ERROR.to_numeric() {
203            self.metrics.inc_errors();
204        }
205
206        // If the buffer is full, evict and retry with bounded retries.
207        //
208        // `pop_local` targets the same shard as `push` so the eviction
209        // makes room for the retry on the shard the producer is
210        // actually contending on. Under the default (1-shard) build
211        // this is identical to the historical pop-then-push loop.
212        if let Err(rejected) = self.queue.push(event) {
213            self.metrics.inc_dropped();
214            let mut to_push = rejected;
215            for _ in 0..3 {
216                let _ = self.queue.pop_local();
217                match self.queue.push(to_push) {
218                    Ok(()) => break,
219                    Err(e) => to_push = e,
220                }
221            }
222        }
223
224        // Wake the flusher thread — no Mutex on the hot path.
225        if let Some(thread) = &self.flusher_thread_handle {
226            thread.unpark();
227        }
228    }
229
230    /// Sets the global log level filter.
231    pub fn set_filter(&self, level: u8) {
232        self.filter_level.store(level, Ordering::Release);
233    }
234
235    /// Returns the current global log level filter.
236    #[must_use]
237    pub fn filter_level(&self) -> u8 {
238        self.filter_level.load(Ordering::Relaxed)
239    }
240
241    /// Increments the format counter in the TUI metrics.
242    pub fn inc_format(&self, format: crate::log_format::LogFormat) {
243        self.metrics.inc_format(format);
244    }
245
246    /// Increments the active span count in the TUI metrics.
247    pub fn inc_spans(&self) {
248        self.metrics.inc_spans();
249    }
250
251    /// Decrements the active span count in the TUI metrics.
252    pub fn dec_spans(&self) {
253        self.metrics.dec_spans();
254    }
255
256    /// Returns the current number of active spans.
257    #[must_use]
258    pub fn active_spans(&self) -> usize {
259        self.metrics.active_spans.load(Ordering::Relaxed)
260    }
261
262    /// Applies configuration settings to the engine.
263    ///
264    /// Sets the log level filter from the config. File sink construction
265    /// and rotation are handled by the flusher thread at startup via
266    /// [`PlatformSink::from_config`](crate::sink::PlatformSink::from_config).
267    pub fn apply_config(&self, config: &crate::config::Config) {
268        self.set_filter(config.log_level.to_numeric());
269    }
270
271    /// Safely halts the background thread, flushing pending logs.
272    ///
273    /// Signals the flusher thread to stop and waits for it to finish
274    /// draining any remaining events from the queue.
275    pub fn shutdown(&self) {
276        self.shutdown_flag.store(true, Ordering::SeqCst);
277        // Wake the flusher so it can drain and exit.
278        if let Some(thread) = &self.flusher_thread_handle {
279            thread.unpark();
280        }
281        if let Ok(mut guard) = self.flusher_join.lock()
282            && let Some(handle) = guard.take()
283        {
284            let _ = handle.join();
285        }
286    }
287}
288
289/// Zero-Allocation Serializer Helper
290#[derive(Debug, Clone, Copy)]
291pub struct FastSerializer;
292
293impl FastSerializer {
294    /// Appends a u64 integer to a buffer using `itoa` without allocating a String.
295    pub fn append_u64(buf: &mut Vec<u8>, val: u64) {
296        let mut buffer = itoa::Buffer::new();
297        buf.extend_from_slice(buffer.format(val).as_bytes());
298    }
299
300    /// Appends an f64 float to a buffer using `ryu` without allocating a String.
301    pub fn append_f64(buf: &mut Vec<u8>, val: f64) {
302        let mut buffer = ryu::Buffer::new();
303        buf.extend_from_slice(buffer.format(val).as_bytes());
304    }
305}
306
307#[cfg(test)]
308#[cfg_attr(miri, allow(unused_imports))]
309mod tests {
310    use super::*;
311    use crate::LogLevel;
312    use crate::log::Log;
313
314    fn make_event(level: LogLevel) -> LogEvent {
315        LogEvent {
316            level,
317            level_num: level.to_numeric(),
318            log: Log::build(level, "test"),
319        }
320    }
321
322    #[test]
323    #[cfg_attr(miri, ignore)]
324    fn fast_serializer_round_trip() {
325        let mut buf = Vec::new();
326        FastSerializer::append_u64(&mut buf, 1234);
327        FastSerializer::append_f64(&mut buf, 3.5);
328        assert_eq!(buf, b"12343.5");
329    }
330
331    #[test]
332    #[cfg_attr(miri, ignore)]
333    fn ingest_overfills_small_queue_and_drops() {
334        // Capacity 1 — every ingest after the first hits the retry loop
335        // (covers the `Err(e) => to_push = e` continuation in the retry).
336        let engine = LockFreeEngine::new(1);
337        for _ in 0..32 {
338            engine.ingest(make_event(LogLevel::INFO));
339        }
340        engine.shutdown();
341    }
342
343    #[test]
344    #[cfg_attr(miri, ignore)]
345    fn ingest_under_concurrent_overfill_hits_retry_err_branch() {
346        // 8 producer threads hammer a capacity-1 queue concurrently.
347        // Statistically guaranteed to hit the `Err(e) => to_push = e`
348        // arm in the retry loop (line 204 in src/engine.rs) when a
349        // peer thread refills the slot between `pop` and `push`.
350        let engine = Arc::new(LockFreeEngine::new(1));
351        let mut handles = Vec::new();
352        for _ in 0..8 {
353            let e = engine.clone();
354            handles.push(thread::spawn(move || {
355                for _ in 0..2_000 {
356                    e.ingest(make_event(LogLevel::INFO));
357                }
358            }));
359        }
360        for h in handles {
361            h.join().unwrap();
362        }
363        engine.shutdown();
364    }
365
366    #[test]
367    #[cfg_attr(miri, ignore)]
368    fn ingest_below_filter_short_circuits() {
369        let engine = LockFreeEngine::new(8);
370        engine.set_filter(LogLevel::ERROR.to_numeric());
371        // DEBUG is below ERROR — should be filtered out before push.
372        engine.ingest(make_event(LogLevel::DEBUG));
373        engine.shutdown();
374    }
375
376    #[test]
377    #[cfg_attr(miri, ignore)]
378    fn engine_records_errors_and_spans() {
379        let engine = LockFreeEngine::new(8);
380        engine.ingest(make_event(LogLevel::ERROR));
381        engine.inc_format(crate::log_format::LogFormat::JSON);
382        engine.inc_spans();
383        engine.inc_spans();
384        assert_eq!(engine.active_spans(), 2);
385        engine.dec_spans();
386        assert_eq!(engine.active_spans(), 1);
387        engine.shutdown();
388    }
389
390    #[test]
391    #[cfg_attr(miri, ignore)]
392    fn shutdown_is_idempotent() {
393        let engine = LockFreeEngine::new(4);
394        engine.ingest(make_event(LogLevel::INFO));
395        engine.shutdown();
396        // Second shutdown: flusher_join has already been drained, so the
397        // `Some(handle) = guard.take()` branch is None this time.
398        engine.shutdown();
399    }
400
401    #[test]
402    fn apply_config_updates_filter() {
403        let engine = LockFreeEngine::new(4);
404        let cfg = crate::config::Config {
405            log_level: LogLevel::WARN,
406            ..crate::config::Config::default()
407        };
408        engine.apply_config(&cfg);
409        assert_eq!(engine.filter_level(), LogLevel::WARN.to_numeric());
410        engine.shutdown();
411    }
412}
rlg/engine.rs

rlg/
engine.rs
