grafton_ndi/
runtime.rs

1//! NDI runtime management and initialization.
2
3use std::sync::{Condvar, Mutex, MutexGuard};
4
5use crate::{ndi_lib::*, Error, Result};
6
7/// Runtime lifecycle phase.
8///
9/// Transitions:
10/// - `Uninitialized` → `Initializing` → `Running` (on success)
11/// - `Uninitialized` → `Initializing` → `Failed` (on failure)
12/// - `Running` → `Destroying` → `Uninitialized` (on last release)
13/// - `Failed` → `Initializing` (retry allowed)
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
15enum Phase {
16    /// Runtime not initialized. Ready for initialization.
17    Uninitialized,
18    /// Initialization in progress. Other threads should wait.
19    Initializing,
20    /// Runtime is active with one or more live handles.
21    Running,
22    /// Destruction in progress. Other threads should wait.
23    Destroying,
24    /// Last initialization attempt failed. Retry is allowed.
25    Failed,
26}
27
28/// Internal state protected by the mutex.
29struct RuntimeState {
30    phase: Phase,
31    refcount: usize,
32}
33
34impl RuntimeState {
35    const fn new() -> Self {
36        Self {
37            phase: Phase::Uninitialized,
38            refcount: 0,
39        }
40    }
41}
42
43// Private SDK hooks let tests exercise the production state machine.
44trait RuntimeBackend {
45    fn initialize(&self) -> bool;
46    fn destroy(&self);
47}
48
49struct SdkBackend;
50
51impl RuntimeBackend for SdkBackend {
52    fn initialize(&self) -> bool {
53        unsafe { NDIlib_initialize() }
54    }
55
56    fn destroy(&self) {
57        unsafe { NDIlib_destroy() };
58    }
59}
60
61/// Process-global runtime manager for NDI.
62///
63/// This implementation uses a `Mutex` + `Condvar` state machine that:
64/// - Allows re-initialization after teardown
65/// - Allows retry after initialization failure
66/// - Avoids spin loops by using `Condvar` waits
67/// - Maintains the invariant: `NDI::new()` returns `Ok` only when runtime is initialized
68struct RuntimeManager<B = SdkBackend> {
69    state: Mutex<RuntimeState>,
70    cv: Condvar,
71    backend: B,
72}
73
74impl RuntimeManager {
75    const fn new() -> Self {
76        Self::with_backend(SdkBackend)
77    }
78}
79
80impl<B: RuntimeBackend> RuntimeManager<B> {
81    const fn with_backend(backend: B) -> Self {
82        Self {
83            state: Mutex::new(RuntimeState::new()),
84            cv: Condvar::new(),
85            backend,
86        }
87    }
88
89    /// Recover from mutex poisoning, preferring progress over panic.
90    fn recover_guard<'a>(
91        result: std::sync::LockResult<MutexGuard<'a, RuntimeState>>,
92    ) -> MutexGuard<'a, RuntimeState> {
93        result.unwrap_or_else(|poisoned| poisoned.into_inner())
94    }
95
96    fn acquire(&self) -> Result<()> {
97        let mut guard = Self::recover_guard(self.state.lock());
98
99        loop {
100            match guard.phase {
101                Phase::Uninitialized | Phase::Failed => {
102                    // We're responsible for initialization
103                    guard.phase = Phase::Initializing;
104                    drop(guard);
105
106                    // Call NDIlib_initialize outside the lock
107                    #[cfg(all(target_os = "windows", debug_assertions))]
108                    {
109                        if std::env::var("CI").is_ok() {
110                            eprintln!("[NDI] Initializing NDI runtime in CI environment...");
111                            if let Ok(sdk_dir) = std::env::var("NDI_SDK_DIR") {
112                                eprintln!("[NDI] NDI_SDK_DIR: {}", sdk_dir);
113                            }
114                        }
115                    }
116
117                    let succeeded = self.backend.initialize();
118
119                    // Re-acquire lock to update state
120                    guard = Self::recover_guard(self.state.lock());
121
122                    if succeeded {
123                        guard.phase = Phase::Running;
124                        guard.refcount = 1;
125                        self.cv.notify_all();
126                        return Ok(());
127                    } else {
128                        guard.phase = Phase::Failed;
129                        self.cv.notify_all();
130                        return Err(Error::InitializationFailed(
131                            "NDIlib_initialize failed".into(),
132                        ));
133                    }
134                }
135
136                Phase::Initializing | Phase::Destroying => {
137                    // Wait for state transition to complete
138                    guard = Self::recover_guard(self.cv.wait(guard));
139                    // Loop again to check the new state
140                }
141
142                Phase::Running => {
143                    // Runtime is already initialized, just increment refcount
144                    guard.refcount = guard.refcount.checked_add(1).ok_or_else(|| {
145                        Error::InitializationFailed("NDI runtime reference count exhausted".into())
146                    })?;
147                    return Ok(());
148                }
149            }
150        }
151    }
152
153    fn release(&self) {
154        let mut guard = Self::recover_guard(self.state.lock());
155
156        debug_assert!(
157            guard.refcount > 0,
158            "release() called when refcount was already 0 (double-free or unbalanced release)"
159        );
160        debug_assert!(
161            guard.phase == Phase::Running,
162            "release() called when phase was {:?}, expected Running",
163            guard.phase
164        );
165
166        guard.refcount -= 1;
167
168        if guard.refcount == 0 {
169            // Last reference - destroy the runtime
170            guard.phase = Phase::Destroying;
171            drop(guard);
172
173            // Call NDIlib_destroy outside the lock
174            self.backend.destroy();
175
176            // Re-acquire lock to reset state
177            let mut guard = Self::recover_guard(self.state.lock());
178            guard.phase = Phase::Uninitialized;
179            self.cv.notify_all();
180        }
181    }
182
183    fn is_running(&self) -> bool {
184        let guard = Self::recover_guard(self.state.lock());
185        guard.phase == Phase::Running && guard.refcount > 0
186    }
187}
188
189static RUNTIME: RuntimeManager = RuntimeManager::new();
190
191/// Manages the NDI runtime lifecycle.
192///
193/// The `NDI` struct is the entry point for all NDI operations. It ensures the NDI
194/// runtime is properly initialized and cleaned up. Multiple instances can exist
195/// simultaneously - they share the same underlying runtime through reference counting.
196/// Create handles with [`NDI::new`] or [`Clone::clone`]; direct construction is private
197/// so every handle owns a reference to the initialized runtime.
198///
199/// # Examples
200///
201/// ```no_run
202/// use grafton_ndi::NDI;
203///
204/// # fn main() -> Result<(), grafton_ndi::Error> {
205/// // Create an NDI instance
206/// let ndi = NDI::new()?;
207///
208/// // The runtime stays alive as long as any NDI instance exists
209/// let ndi2 = ndi.clone(); // Cheap reference-counted clone
210///
211/// // Runtime is automatically cleaned up when all instances are dropped
212/// # Ok(())
213/// # }
214/// ```
215#[derive(Debug)]
216pub struct NDI {
217    _private: (),
218}
219
220impl NDI {
221    /// Creates a new NDI instance.
222    ///
223    /// This checked constructor is thread-safe and can be called from multiple
224    /// threads. It initializes the runtime when necessary; otherwise it acquires
225    /// another runtime reference. Cloning a live handle also acquires a reference.
226    /// The runtime is destroyed when the last handle is dropped, including handles
227    /// retained internally by senders, receivers, and finders.
228    ///
229    /// # Errors
230    ///
231    /// Returns [`Error::InitializationFailed`] if the NDI SDK fails to initialize
232    /// or the process has exhausted the runtime handle reference count.
233    ///
234    /// # Examples
235    ///
236    /// ```
237    /// # use grafton_ndi::NDI;
238    /// # fn main() -> Result<(), grafton_ndi::Error> {
239    /// let ndi = NDI::new()?;
240    /// // Use NDI operations...
241    /// # Ok(())
242    /// # }
243    /// ```
244    pub fn new() -> Result<Self> {
245        RUNTIME.acquire()?;
246        Ok(Self { _private: () })
247    }
248
249    /// Checks if the current CPU is supported by the NDI SDK.
250    ///
251    /// The NDI SDK requires certain CPU features (e.g., SSE4.2 on x86_64).
252    ///
253    /// # Examples
254    ///
255    /// ```
256    /// if grafton_ndi::NDI::is_supported_cpu() {
257    ///     println!("CPU is supported by NDI");
258    /// } else {
259    ///     eprintln!("CPU lacks required features for NDI");
260    /// }
261    /// ```
262    pub fn is_supported_cpu() -> bool {
263        unsafe { NDIlib_is_supported_CPU() }
264    }
265
266    /// Returns the version string of the NDI runtime.
267    ///
268    /// # Errors
269    ///
270    /// Returns an error if the version string cannot be retrieved or contains
271    /// invalid UTF-8.
272    ///
273    /// # Examples
274    ///
275    /// ```no_run
276    /// # use grafton_ndi::NDI;
277    /// # fn main() -> Result<(), grafton_ndi::Error> {
278    /// match NDI::version() {
279    ///     Ok(version) => println!("NDI version: {}", version),
280    ///     Err(e) => eprintln!("Failed to get version: {}", e),
281    /// }
282    /// # Ok(())
283    /// # }
284    /// ```
285    pub fn version() -> Result<String> {
286        unsafe {
287            let version_ptr = NDIlib_version();
288            if version_ptr.is_null() {
289                return Err(Error::NullPointer("NDIlib_version".into()));
290            }
291            let c_str = std::ffi::CStr::from_ptr(version_ptr);
292            c_str
293                .to_str()
294                .map(|s| s.to_owned())
295                .map_err(|e| Error::InvalidUtf8(e.to_string()))
296        }
297    }
298    /// Checks if the NDI runtime is currently initialized.
299    ///
300    /// This can be useful for diagnostic purposes or conditional initialization.
301    ///
302    /// # Examples
303    ///
304    /// ```
305    /// if grafton_ndi::NDI::is_running() {
306    ///     println!("NDI runtime is active");
307    /// }
308    /// ```
309    pub fn is_running() -> bool {
310        RUNTIME.is_running()
311    }
312}
313
314impl Clone for NDI {
315    /// Acquires another reference to the active runtime.
316    ///
317    /// # Panics
318    ///
319    /// Panics if the runtime handle reference count is exhausted. The original
320    /// handle and runtime remain valid.
321    fn clone(&self) -> Self {
322        RUNTIME
323            .acquire()
324            .expect("Unable to acquire another reference to the existing NDI runtime");
325        Self { _private: () }
326    }
327}
328
329impl Drop for NDI {
330    fn drop(&mut self) {
331        RUNTIME.release();
332    }
333}
334
335#[cfg(test)]
336mod tests {
337    use super::*;
338    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
339    use std::sync::Arc;
340    use std::thread;
341    use std::time::Duration;
342
343    #[derive(Default)]
344    struct MockBackend {
345        init_count: AtomicUsize,
346        destroy_count: AtomicUsize,
347        init_should_fail: AtomicBool,
348        init_delay_ms: AtomicUsize,
349        destroy_delay_ms: AtomicUsize,
350    }
351
352    impl RuntimeBackend for MockBackend {
353        fn initialize(&self) -> bool {
354            thread::sleep(Duration::from_millis(
355                self.init_delay_ms.load(Ordering::Acquire) as u64,
356            ));
357            self.init_count.fetch_add(1, Ordering::AcqRel);
358            !self.init_should_fail.load(Ordering::Acquire)
359        }
360
361        fn destroy(&self) {
362            thread::sleep(Duration::from_millis(
363                self.destroy_delay_ms.load(Ordering::Acquire) as u64,
364            ));
365            self.destroy_count.fetch_add(1, Ordering::AcqRel);
366        }
367    }
368
369    impl RuntimeManager<MockBackend> {
370        fn phase(&self) -> Phase {
371            Self::recover_guard(self.state.lock()).phase
372        }
373
374        fn refcount(&self) -> usize {
375            Self::recover_guard(self.state.lock()).refcount
376        }
377    }
378
379    #[test]
380    fn refcount_overflow_preserves_live_runtime() {
381        let manager = RuntimeManager::with_backend(MockBackend::default());
382        manager.acquire().unwrap();
383        manager.state.lock().unwrap().refcount = usize::MAX;
384        assert!(manager.acquire().is_err());
385        assert_eq!(manager.refcount(), usize::MAX);
386        assert!(manager.is_running());
387        assert_eq!(manager.backend.destroy_count.load(Ordering::Acquire), 0);
388        // Restore the injected count so the acquired reference can be released.
389        manager.state.lock().unwrap().refcount = 1;
390        manager.release();
391        assert_eq!(manager.backend.destroy_count.load(Ordering::Acquire), 1);
392    }
393
394    // ========== Lifecycle Invariant Tests ==========
395
396    #[test]
397    fn test_reinit_after_teardown() {
398        // Issue requirement: create NDI, drop all, create NDI again
399        // => init called twice, destroy called twice, both NDI::new() return Ok
400        let manager = Arc::new(RuntimeManager::with_backend(MockBackend::default()));
401
402        // First cycle
403        manager.acquire().expect("First init should succeed");
404        assert_eq!(manager.backend.init_count.load(Ordering::Acquire), 1);
405        assert!(manager.is_running());
406
407        manager.release();
408        assert_eq!(manager.backend.destroy_count.load(Ordering::Acquire), 1);
409        assert!(!manager.is_running());
410        assert_eq!(manager.phase(), Phase::Uninitialized);
411
412        // Second cycle - must re-initialize
413        manager.acquire().expect("Second init should succeed");
414        assert_eq!(manager.backend.init_count.load(Ordering::Acquire), 2);
415        assert!(manager.is_running());
416
417        manager.release();
418        assert_eq!(manager.backend.destroy_count.load(Ordering::Acquire), 2);
419        assert!(!manager.is_running());
420    }
421
422    #[test]
423    fn test_init_failure_retry() {
424        // Issue requirement: first init fails => error, next call succeeds => Ok
425        let manager = Arc::new(RuntimeManager::with_backend(MockBackend::default()));
426
427        // Configure first init to fail
428        manager
429            .backend
430            .init_should_fail
431            .store(true, Ordering::Release);
432
433        let result = manager.acquire();
434        assert!(result.is_err());
435        assert_eq!(manager.backend.init_count.load(Ordering::Acquire), 1);
436        assert_eq!(manager.phase(), Phase::Failed);
437        assert!(!manager.is_running());
438
439        // Configure next init to succeed
440        manager
441            .backend
442            .init_should_fail
443            .store(false, Ordering::Release);
444
445        let result = manager.acquire();
446        assert!(result.is_ok());
447        assert_eq!(manager.backend.init_count.load(Ordering::Acquire), 2);
448        assert!(manager.is_running());
449
450        manager.release();
451    }
452
453    #[test]
454    fn test_no_ok_while_uninitialized() {
455        // Issue requirement: cannot get Ok from acquire without init in that cycle
456        let manager = Arc::new(RuntimeManager::with_backend(MockBackend::default()));
457
458        // Verify initial state
459        assert_eq!(manager.phase(), Phase::Uninitialized);
460        assert!(!manager.is_running());
461        assert_eq!(manager.backend.init_count.load(Ordering::Acquire), 0);
462
463        // Acquire must call init
464        manager.acquire().expect("Should succeed");
465        assert_eq!(manager.backend.init_count.load(Ordering::Acquire), 1);
466
467        // Clone/acquire with running does NOT call init
468        manager.acquire().expect("Clone should succeed");
469        assert_eq!(manager.backend.init_count.load(Ordering::Acquire), 1);
470        assert_eq!(manager.refcount(), 2);
471
472        // Release both
473        manager.release();
474        manager.release();
475
476        // After teardown, next acquire MUST call init
477        assert_eq!(manager.phase(), Phase::Uninitialized);
478        manager.acquire().expect("Re-init should succeed");
479        assert_eq!(manager.backend.init_count.load(Ordering::Acquire), 2);
480
481        manager.release();
482    }
483
484    #[test]
485    fn test_destroy_called_exactly_once_per_cycle() {
486        // Issue requirement: destroy called exactly once per successful init cycle
487        let manager = Arc::new(RuntimeManager::with_backend(MockBackend::default()));
488
489        // First cycle with multiple refs
490        manager.acquire().expect("Init");
491        manager.acquire().expect("Clone 1");
492        manager.acquire().expect("Clone 2");
493        assert_eq!(manager.refcount(), 3);
494        assert_eq!(manager.backend.destroy_count.load(Ordering::Acquire), 0);
495
496        manager.release();
497        assert_eq!(manager.backend.destroy_count.load(Ordering::Acquire), 0);
498        manager.release();
499        assert_eq!(manager.backend.destroy_count.load(Ordering::Acquire), 0);
500        manager.release(); // Last one
501        assert_eq!(manager.backend.destroy_count.load(Ordering::Acquire), 1);
502
503        // Second cycle
504        manager.acquire().expect("Re-init");
505        manager.acquire().expect("Clone");
506        assert_eq!(manager.backend.destroy_count.load(Ordering::Acquire), 1);
507
508        manager.release();
509        manager.release();
510        assert_eq!(manager.backend.destroy_count.load(Ordering::Acquire), 2);
511    }
512
513    // ========== Concurrency Stress Tests ==========
514
515    #[test]
516    fn test_concurrent_acquire_single_init() {
517        // Issue requirement: concurrent NDI::new() calls result in at most one init per cycle
518        let manager = Arc::new(RuntimeManager::with_backend(MockBackend::default()));
519
520        // Add a small delay to init to increase chance of race
521        manager.backend.init_delay_ms.store(5, Ordering::Release);
522
523        let handles: Vec<_> = (0..10)
524            .map(|_| {
525                let mgr = Arc::clone(&manager);
526                thread::spawn(move || mgr.acquire())
527            })
528            .collect();
529
530        let results: Vec<_> = handles.into_iter().map(|h| h.join().unwrap()).collect();
531
532        // All should succeed
533        for result in &results {
534            assert!(result.is_ok());
535        }
536
537        // Only one init call should have happened
538        assert_eq!(manager.backend.init_count.load(Ordering::Acquire), 1);
539        assert_eq!(manager.refcount(), 10);
540
541        // Cleanup
542        for _ in 0..10 {
543            manager.release();
544        }
545    }
546
547    #[test]
548    fn test_concurrent_acquire_during_destroy() {
549        // Issue requirement: callers block during destroy, then succeed after
550        let manager = Arc::new(RuntimeManager::with_backend(MockBackend::default()));
551
552        // Initialize
553        manager.acquire().expect("Init");
554        assert!(manager.is_running());
555
556        // Add delay to destroy
557        manager
558            .backend
559            .destroy_delay_ms
560            .store(50, Ordering::Release);
561
562        let mgr_clone = Arc::clone(&manager);
563
564        // Start a thread that will try to acquire while destroy is in progress
565        let acquirer = thread::spawn(move || {
566            // Wait a bit to ensure destroy starts
567            thread::sleep(Duration::from_millis(10));
568            mgr_clone.acquire()
569        });
570
571        // Trigger destroy
572        manager.release();
573
574        // The acquirer should have blocked and then succeeded
575        let result = acquirer.join().unwrap();
576        assert!(result.is_ok());
577
578        // A new init cycle should have started
579        assert_eq!(manager.backend.init_count.load(Ordering::Acquire), 2);
580        assert_eq!(manager.backend.destroy_count.load(Ordering::Acquire), 1);
581
582        manager.release();
583    }
584
585    #[test]
586    fn test_mixed_acquire_release_concurrent() {
587        // Stress test with mixed operations
588        let manager = Arc::new(RuntimeManager::with_backend(MockBackend::default()));
589        let success_count = Arc::new(AtomicUsize::new(0));
590        let failure_count = Arc::new(AtomicUsize::new(0));
591
592        let handles: Vec<_> = (0..20)
593            .map(|i| {
594                let mgr = Arc::clone(&manager);
595                let success = Arc::clone(&success_count);
596                let failure = Arc::clone(&failure_count);
597
598                thread::spawn(move || {
599                    for _ in 0..10 {
600                        match mgr.acquire() {
601                            Ok(()) => {
602                                success.fetch_add(1, Ordering::Relaxed);
603                                // Hold for a bit
604                                if i % 3 == 0 {
605                                    thread::sleep(Duration::from_micros(100));
606                                }
607                                mgr.release();
608                            }
609                            Err(_) => {
610                                failure.fetch_add(1, Ordering::Relaxed);
611                            }
612                        }
613                    }
614                })
615            })
616            .collect();
617
618        for handle in handles {
619            handle.join().expect("Thread panicked");
620        }
621
622        // All should succeed (no failures configured)
623        assert_eq!(success_count.load(Ordering::Relaxed), 200);
624        assert_eq!(failure_count.load(Ordering::Relaxed), 0);
625
626        // Final state should be clean
627        assert!(!manager.is_running());
628        assert_eq!(manager.refcount(), 0);
629        assert_eq!(manager.phase(), Phase::Uninitialized);
630    }
631
632    #[test]
633    fn test_concurrent_init_with_failures() {
634        // Test that failures during concurrent init are handled correctly
635        let manager = Arc::new(RuntimeManager::with_backend(MockBackend::default()));
636        manager
637            .backend
638            .init_should_fail
639            .store(true, Ordering::Release);
640        manager.backend.init_delay_ms.store(5, Ordering::Release);
641
642        let handles: Vec<_> = (0..5)
643            .map(|_| {
644                let mgr = Arc::clone(&manager);
645                thread::spawn(move || mgr.acquire())
646            })
647            .collect();
648
649        let results: Vec<_> = handles.into_iter().map(|h| h.join().unwrap()).collect();
650
651        // All should fail
652        for result in &results {
653            assert!(result.is_err());
654        }
655
656        // State should be Failed
657        assert_eq!(manager.phase(), Phase::Failed);
658        assert!(!manager.is_running());
659
660        // Now allow init to succeed
661        manager
662            .backend
663            .init_should_fail
664            .store(false, Ordering::Release);
665
666        // Retry should work
667        manager.acquire().expect("Retry should succeed");
668        assert!(manager.is_running());
669
670        manager.release();
671    }
672
673    #[test]
674    fn test_client_handles_can_be_released_and_recreated() {
675        // Independent clients can share the runtime, release it, and start again.
676        let manager = Arc::new(RuntimeManager::with_backend(MockBackend::default()));
677
678        // First client initializes the runtime.
679        manager.acquire().expect("First client");
680        assert_eq!(manager.backend.init_count.load(Ordering::Acquire), 1);
681
682        // Additional clients acquire independent references.
683        manager.acquire().expect("Second client");
684        manager.acquire().expect("Third client");
685        assert_eq!(manager.refcount(), 3);
686
687        // All clients release their handles.
688        manager.release();
689        manager.release();
690        manager.release();
691
692        // Runtime should be destroyed
693        assert_eq!(manager.backend.destroy_count.load(Ordering::Acquire), 1);
694        assert!(!manager.is_running());
695
696        // A later client must reinitialize the runtime.
697        manager.acquire().expect("New client after teardown");
698        assert_eq!(manager.backend.init_count.load(Ordering::Acquire), 2);
699        assert!(manager.is_running());
700
701        manager.release();
702    }
703}
grafton_ndi/runtime.rs

grafton_ndi/
runtime.rs
