rns_net/interface/
mod.rs

1//! Network interface abstractions.
2
3#[cfg(feature = "iface-auto")]
4pub mod auto;
5#[cfg(feature = "iface-backbone")]
6pub mod backbone;
7#[cfg(feature = "iface-i2p")]
8pub mod i2p;
9#[cfg(feature = "iface-kiss")]
10pub mod kiss_iface;
11#[cfg(feature = "iface-local")]
12pub mod local;
13#[cfg(feature = "iface-pipe")]
14pub mod pipe;
15pub mod registry;
16#[cfg(feature = "iface-rnode")]
17pub mod rnode;
18#[cfg(feature = "iface-serial")]
19pub mod serial_iface;
20#[cfg(feature = "iface-tcp")]
21pub mod tcp;
22#[cfg(feature = "iface-tcp")]
23pub mod tcp_server;
24#[cfg(feature = "iface-udp")]
25pub mod udp;
26
27use std::any::Any;
28use std::collections::HashMap;
29use std::io;
30use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
31use std::sync::mpsc::{sync_channel, SyncSender, TrySendError};
32use std::sync::{Arc, Mutex, MutexGuard};
33use std::thread;
34use std::time::{Duration, Instant};
35
36use crate::event::EventSender;
37use crate::ifac::IfacState;
38use rns_core::transport::types::{InterfaceId, InterfaceInfo};
39
40/// Bind a socket to a specific network interface using `SO_BINDTODEVICE`.
41///
42/// Requires `CAP_NET_RAW` or root on Linux.
43#[cfg(target_os = "linux")]
44pub fn bind_to_device(fd: std::os::unix::io::RawFd, device: &str) -> io::Result<()> {
45    let dev_bytes = device.as_bytes();
46    if dev_bytes.len() >= libc::IFNAMSIZ {
47        return Err(io::Error::new(
48            io::ErrorKind::InvalidInput,
49            format!("device name too long: {}", device),
50        ));
51    }
52    let ret = unsafe {
53        libc::setsockopt(
54            fd,
55            libc::SOL_SOCKET,
56            libc::SO_BINDTODEVICE,
57            dev_bytes.as_ptr() as *const libc::c_void,
58            dev_bytes.len() as libc::socklen_t,
59        )
60    };
61    if ret != 0 {
62        return Err(io::Error::last_os_error());
63    }
64    Ok(())
65}
66
67/// Writable end of an interface. Held by the driver.
68///
69/// Each implementation wraps a socket + framing.
70pub trait Writer: Send {
71    fn send_frame(&mut self, data: &[u8]) -> io::Result<()>;
72}
73
74pub const DEFAULT_ASYNC_WRITER_QUEUE_CAPACITY: usize = 256;
75
76pub(crate) fn lock_or_recover<'a, T>(mutex: &'a Mutex<T>, label: &str) -> MutexGuard<'a, T> {
77    match mutex.lock() {
78        Ok(guard) => guard,
79        Err(poisoned) => {
80            log::warn!("recovering poisoned mutex: {}", label);
81            poisoned.into_inner()
82        }
83    }
84}
85
86#[derive(Clone, Default)]
87pub struct ListenerControl {
88    stop: Arc<AtomicBool>,
89}
90
91impl ListenerControl {
92    pub fn new() -> Self {
93        Self {
94            stop: Arc::new(AtomicBool::new(false)),
95        }
96    }
97
98    pub fn request_stop(&self) {
99        self.stop.store(true, Ordering::Relaxed);
100    }
101
102    pub fn should_stop(&self) -> bool {
103        self.stop.load(Ordering::Relaxed)
104    }
105}
106
107#[derive(Clone, Default)]
108pub struct AsyncWriterMetrics {
109    queued_frames: Arc<AtomicUsize>,
110    worker_alive: Arc<AtomicBool>,
111}
112
113impl AsyncWriterMetrics {
114    pub fn queued_frames(&self) -> usize {
115        self.queued_frames.load(Ordering::Relaxed)
116    }
117
118    pub fn worker_alive(&self) -> bool {
119        self.worker_alive.load(Ordering::Relaxed)
120    }
121}
122
123struct AsyncWriter {
124    tx: SyncSender<Vec<u8>>,
125    metrics: AsyncWriterMetrics,
126}
127
128impl Writer for AsyncWriter {
129    fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
130        if !self.metrics.worker_alive() {
131            return Err(io::Error::new(
132                io::ErrorKind::BrokenPipe,
133                "interface writer worker is offline",
134            ));
135        }
136
137        match self.tx.try_send(data.to_vec()) {
138            Ok(()) => {
139                self.metrics.queued_frames.fetch_add(1, Ordering::Relaxed);
140                Ok(())
141            }
142            Err(TrySendError::Full(_)) => Err(io::Error::new(
143                io::ErrorKind::WouldBlock,
144                "interface writer queue is full",
145            )),
146            Err(TrySendError::Disconnected(_)) => {
147                self.metrics.worker_alive.store(false, Ordering::Relaxed);
148                Err(io::Error::new(
149                    io::ErrorKind::BrokenPipe,
150                    "interface writer worker disconnected",
151                ))
152            }
153        }
154    }
155}
156
157pub fn wrap_async_writer(
158    writer: Box<dyn Writer>,
159    interface_id: InterfaceId,
160    interface_name: &str,
161    event_tx: EventSender,
162    queue_capacity: usize,
163) -> (Box<dyn Writer>, AsyncWriterMetrics) {
164    let (tx, rx) = sync_channel::<Vec<u8>>(queue_capacity.max(1));
165    let metrics = AsyncWriterMetrics {
166        queued_frames: Arc::new(AtomicUsize::new(0)),
167        worker_alive: Arc::new(AtomicBool::new(true)),
168    };
169    let metrics_thread = metrics.clone();
170    let name = interface_name.to_string();
171
172    let spawn_result = thread::Builder::new()
173        .name(format!("iface-writer-{}", interface_id.0))
174        .spawn(move || async_writer_loop(writer, rx, interface_id, name, event_tx, metrics_thread));
175
176    if let Err(err) = spawn_result {
177        metrics.worker_alive.store(false, Ordering::Relaxed);
178        log::error!(
179            "[{}:{}] failed to spawn async writer thread: {}",
180            interface_name,
181            interface_id.0,
182            err
183        );
184        return (Box::new(DirectWriterFallback), metrics);
185    }
186
187    (
188        Box::new(AsyncWriter {
189            tx,
190            metrics: metrics.clone(),
191        }),
192        metrics,
193    )
194}
195
196struct DirectWriterFallback;
197
198impl Writer for DirectWriterFallback {
199    fn send_frame(&mut self, _data: &[u8]) -> io::Result<()> {
200        Err(io::Error::other("interface writer worker unavailable"))
201    }
202}
203
204fn async_writer_loop(
205    mut writer: Box<dyn Writer>,
206    rx: std::sync::mpsc::Receiver<Vec<u8>>,
207    interface_id: InterfaceId,
208    interface_name: String,
209    event_tx: EventSender,
210    metrics: AsyncWriterMetrics,
211) {
212    while let Ok(frame) = rx.recv() {
213        metrics.queued_frames.fetch_sub(1, Ordering::Relaxed);
214        if let Err(err) = writer.send_frame(&frame) {
215            metrics.worker_alive.store(false, Ordering::Relaxed);
216            log::warn!(
217                "[{}:{}] async writer exiting after send failure: {}",
218                interface_name,
219                interface_id.0,
220                err
221            );
222            let _ = event_tx.send(crate::event::Event::InterfaceDown(interface_id));
223            return;
224        }
225    }
226
227    metrics.worker_alive.store(false, Ordering::Relaxed);
228}
229
230pub use crate::common::interface_stats::{InterfaceStats, ANNOUNCE_SAMPLE_MAX};
231
232use crate::common::management::InterfaceStatusView;
233
234/// Everything the driver tracks per interface.
235pub struct InterfaceEntry {
236    pub id: InterfaceId,
237    pub info: InterfaceInfo,
238    pub writer: Box<dyn Writer>,
239    pub async_writer_metrics: Option<AsyncWriterMetrics>,
240    /// Administrative enable/disable state.
241    pub enabled: bool,
242    pub online: bool,
243    /// True for dynamically spawned interfaces (e.g. TCP server clients).
244    /// These are fully removed on InterfaceDown rather than just marked offline.
245    pub dynamic: bool,
246    /// IFAC state for this interface, if access codes are enabled.
247    pub ifac: Option<IfacState>,
248    /// Traffic statistics.
249    pub stats: InterfaceStats,
250    /// Human-readable interface type string (e.g. "TCPClientInterface").
251    pub interface_type: String,
252    /// Next time a send should be retried after a transient WouldBlock.
253    pub send_retry_at: Option<Instant>,
254    /// Current retry backoff for transient send failures.
255    pub send_retry_backoff: Duration,
256}
257
258/// Result of starting an interface via a factory.
259pub enum StartResult {
260    /// One writer, registered immediately (TcpClient, Udp, Serial, etc.)
261    Simple {
262        id: InterfaceId,
263        info: InterfaceInfo,
264        writer: Box<dyn Writer>,
265        interface_type_name: String,
266    },
267    /// Spawns a listener; dynamic interfaces arrive via Event::InterfaceUp (TcpServer, Auto, I2P, etc.)
268    Listener { control: Option<ListenerControl> },
269    /// Multiple subinterfaces from one config (RNode).
270    Multi(Vec<SubInterface>),
271}
272
273/// A single subinterface returned from a multi-interface factory.
274pub struct SubInterface {
275    pub id: InterfaceId,
276    pub info: InterfaceInfo,
277    pub writer: Box<dyn Writer>,
278    pub interface_type_name: String,
279}
280
281/// Context passed to [`InterfaceFactory::start()`].
282pub struct StartContext {
283    pub tx: EventSender,
284    pub next_dynamic_id: Arc<AtomicU64>,
285    pub mode: u8,
286    pub recursive_prs: bool,
287    pub announces_from_internal: bool,
288    pub ingress_control: rns_core::transport::types::IngressControlConfig,
289}
290
291/// Opaque interface config data. Each factory downcasts to its concrete type.
292pub trait InterfaceConfigData: Send + Any {
293    fn as_any(&self) -> &dyn Any;
294    fn into_any(self: Box<Self>) -> Box<dyn Any>;
295}
296
297impl<T: Send + 'static> InterfaceConfigData for T {
298    fn as_any(&self) -> &dyn Any {
299        self
300    }
301
302    fn into_any(self: Box<Self>) -> Box<dyn Any> {
303        self
304    }
305}
306
307/// Factory that can parse config and start an interface type.
308pub trait InterfaceFactory: Send + Sync {
309    /// Config-file type name, e.g. "TCPClientInterface".
310    fn type_name(&self) -> &str;
311
312    /// Default IFAC size (bytes). 8 for serial/kiss/rnode, 16 for others.
313    fn default_ifac_size(&self) -> usize {
314        16
315    }
316
317    /// Parse from key-value params (config file or external).
318    fn parse_config(
319        &self,
320        name: &str,
321        id: InterfaceId,
322        params: &HashMap<String, String>,
323    ) -> Result<Box<dyn InterfaceConfigData>, String>;
324
325    /// Start the interface from parsed config.
326    fn start(
327        &self,
328        config: Box<dyn InterfaceConfigData>,
329        ctx: StartContext,
330    ) -> io::Result<StartResult>;
331}
332
333impl InterfaceStatusView for InterfaceEntry {
334    fn id(&self) -> InterfaceId {
335        self.id
336    }
337    fn info(&self) -> &InterfaceInfo {
338        &self.info
339    }
340    fn online(&self) -> bool {
341        self.online
342    }
343    fn stats(&self) -> &InterfaceStats {
344        &self.stats
345    }
346}
347
348#[cfg(test)]
349mod tests {
350    use super::*;
351    use crate::event::Event;
352    use rns_core::constants;
353    use std::sync::mpsc;
354
355    struct MockWriter {
356        sent: Vec<Vec<u8>>,
357    }
358
359    impl MockWriter {
360        fn new() -> Self {
361            MockWriter { sent: Vec::new() }
362        }
363    }
364
365    impl Writer for MockWriter {
366        fn send_frame(&mut self, data: &[u8]) -> io::Result<()> {
367            self.sent.push(data.to_vec());
368            Ok(())
369        }
370    }
371
372    #[test]
373    fn interface_entry_construction() {
374        let entry = InterfaceEntry {
375            id: InterfaceId(1),
376            info: InterfaceInfo {
377                id: InterfaceId(1),
378                name: String::new(),
379                mode: constants::MODE_FULL,
380                recursive_prs: false,
381                announces_from_internal: true,
382                out_capable: true,
383                in_capable: true,
384                bitrate: None,
385                airtime_profile: None,
386                announce_rate_target: None,
387                announce_rate_grace: 0,
388                announce_rate_penalty: 0.0,
389                announce_cap: constants::ANNOUNCE_CAP,
390                is_local_client: false,
391                wants_tunnel: false,
392                tunnel_id: None,
393                mtu: constants::MTU as u32,
394                ia_freq: 0.0,
395                ip_freq: 0.0,
396                op_freq: 0.0,
397                op_samples: 0,
398                started: 0.0,
399                ingress_control: rns_core::transport::types::IngressControlConfig::disabled(),
400            },
401            writer: Box::new(MockWriter::new()),
402            async_writer_metrics: None,
403            enabled: true,
404            online: false,
405            dynamic: false,
406            ifac: None,
407            stats: InterfaceStats::default(),
408            interface_type: String::new(),
409            send_retry_at: None,
410            send_retry_backoff: Duration::ZERO,
411        };
412        assert_eq!(entry.id, InterfaceId(1));
413        assert!(!entry.online);
414        assert!(!entry.dynamic);
415    }
416
417    #[test]
418    fn mock_writer_captures_bytes() {
419        let mut writer = MockWriter::new();
420        writer.send_frame(b"hello").unwrap();
421        writer.send_frame(b"world").unwrap();
422        assert_eq!(writer.sent.len(), 2);
423        assert_eq!(writer.sent[0], b"hello");
424        assert_eq!(writer.sent[1], b"world");
425    }
426
427    #[test]
428    fn writer_send_frame_produces_output() {
429        let mut writer = MockWriter::new();
430        let data = vec![0x01, 0x02, 0x03];
431        writer.send_frame(&data).unwrap();
432        assert_eq!(writer.sent[0], data);
433    }
434
435    struct BlockingWriter {
436        entered_tx: mpsc::Sender<()>,
437        release_rx: mpsc::Receiver<()>,
438    }
439
440    impl Writer for BlockingWriter {
441        fn send_frame(&mut self, _data: &[u8]) -> io::Result<()> {
442            let _ = self.entered_tx.send(());
443            let _ = self.release_rx.recv();
444            Ok(())
445        }
446    }
447
448    struct FailingWriter;
449
450    impl Writer for FailingWriter {
451        fn send_frame(&mut self, _data: &[u8]) -> io::Result<()> {
452            Err(io::Error::new(io::ErrorKind::BrokenPipe, "boom"))
453        }
454    }
455
456    #[test]
457    fn async_writer_returns_wouldblock_when_queue_is_full() {
458        let (event_tx, _event_rx) = crate::event::channel();
459        let (entered_tx, entered_rx) = mpsc::channel();
460        let (release_tx, release_rx) = mpsc::channel();
461        let (mut writer, metrics) = wrap_async_writer(
462            Box::new(BlockingWriter {
463                entered_tx,
464                release_rx,
465            }),
466            InterfaceId(7),
467            "test",
468            event_tx,
469            1,
470        );
471
472        writer.send_frame(&[1]).unwrap();
473        entered_rx.recv_timeout(Duration::from_secs(1)).unwrap();
474        writer.send_frame(&[2]).unwrap();
475        assert_eq!(metrics.queued_frames(), 1);
476        let err = writer.send_frame(&[3]).unwrap_err();
477        assert_eq!(err.kind(), io::ErrorKind::WouldBlock);
478
479        let _ = release_tx.send(());
480    }
481
482    #[test]
483    fn async_writer_reports_interface_down_after_worker_failure() {
484        let (event_tx, event_rx) = crate::event::channel();
485        let (mut writer, metrics) =
486            wrap_async_writer(Box::new(FailingWriter), InterfaceId(9), "fail", event_tx, 2);
487
488        writer.send_frame(&[1]).unwrap();
489        let event = event_rx.recv_timeout(Duration::from_secs(1)).unwrap();
490        assert!(matches!(event, Event::InterfaceDown(InterfaceId(9))));
491        assert!(!metrics.worker_alive());
492    }
493}
rns_net/interface/mod.rs

rns_net/interface/
mod.rs
