umshctl/
lan.rs

1//! The LAN virtual radio: a soft ULCP device in this process whose PHY is
2//! the LAN bearer's multicast group.
3//!
4//! The device is the production session on a file-backed platform, so it
5//! has an identity and saved configuration of its own that persist from
6//! one invocation to the next. The tool reaches it over an in-memory byte
7//! pipe carrying the same HDLC-Lite framing a wire does, and everything
8//! above that—attach, the mesh session, every command—is unchanged.
9//!
10//! What the device transmits is sent to the group, where every radio with
11//! its bearer enabled hears it; what any participant sends to the group
12//! is received unmeasured, since nothing here measured it.
13
14use std::net::SocketAddrV4;
15use std::path::PathBuf;
16use std::time::Duration;
17
18use anyhow::{Context as _, Result, anyhow, bail};
19use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream};
20use tokio::sync::mpsc;
21use umsh::tokio_support::LanConfig;
22use umsh_ulcp_simdev::host::HostPlatform;
23use umsh_ulcp_simdev::lan::{self, HostEvent, StateLock};
24use umsh_ulcp_simdev::soft::SoftDevice;
25
26/// How much of the pipe to take in one read. ULCP frames are small; this
27/// only bounds a burst.
28const READ_CHUNK: usize = 4096;
29
30/// How long an open waits for a previous virtual radio in this process
31/// to let go of its state.
32const LOCK_PATIENCE: Duration = Duration::from_secs(2);
33
34/// Where the virtual radio keeps its identity and saved configuration.
35fn device_dir() -> Result<PathBuf> {
36    crate::connection::state_dir()
37        .map(|dir| dir.join("lan-device"))
38        .ok_or_else(|| anyhow!("no HOME directory to keep the LAN device's state in"))
39}
40
41/// Start a virtual radio and return the host's end of the link to it.
42/// A `group` stands in for the radio's own until it exits, as an unsaved
43/// write of `PROP_LAN_BEARER_GROUP` would.
44///
45/// The device runs on this thread's `LocalSet` until the host's end is
46/// dropped.
47pub async fn open(group: Option<SocketAddrV4>) -> Result<DuplexStream> {
48    let dir = device_dir()?;
49    let platform =
50        HostPlatform::new(&dir).with_context(|| format!("creating {}", dir.display()))?;
51    let deadline = tokio::time::Instant::now() + LOCK_PATIENCE;
52    let lock = loop {
53        if let Some(lock) =
54            StateLock::try_acquire(&dir).with_context(|| format!("locking {}", dir.display()))?
55        {
56            break lock;
57        }
58        if tokio::time::Instant::now() >= deadline {
59            bail!(
60                "another umshctl is already the LAN virtual radio ({} is locked)",
61                dir.display()
62            );
63        }
64        tokio::time::sleep(Duration::from_millis(50)).await;
65    };
66    let (host, device_end) = tokio::io::duplex(READ_CHUNK);
67    let mut device = SoftDevice::new(
68        lan::session_config(concat!("umshctl-lan/", env!("CARGO_PKG_VERSION"))),
69        platform,
70    );
71    // Asking for the LAN is asking for it on. A LoRa radio comes up off
72    // so that a throwaway identity never reaches the air; this one has no
73    // node of its own and sends only its host's frames.
74    if !device.session().settings().enabled {
75        let _ = device.session_mut().toggle_radio(&mut |_| {});
76    }
77    if let Some(group) = group {
78        device
79            .session_mut()
80            .set_lan_bearer_group(Some(umsh::ulcp_wire::lan::Group {
81                address: group.ip().octets(),
82                port: group.port(),
83            }));
84    }
85    let (events, host_events) = mpsc::unbounded_channel();
86    let (wire_tx, wire_rx) = mpsc::unbounded_channel();
87    let _ = events.send(HostEvent::Attach(Box::new(move |wire| {
88        let _ = wire_tx.send(wire);
89    })));
90    tokio::task::spawn_local(async move {
91        // Nobody is told of a queued frame: the tool is attached for as
92        // long as the device runs.
93        lan::run(device, LanConfig::lan(), host_events, |_| {}).await;
94        drop(lock);
95    });
96    tokio::task::spawn_local(relay(device_end, events, wire_rx));
97    Ok(host)
98}
99
100/// Carry the pipe to the device and back until the host lets go, which
101/// stops the device.
102async fn relay(
103    mut pipe: DuplexStream,
104    events: mpsc::UnboundedSender<HostEvent>,
105    mut wire: mpsc::UnboundedReceiver<Vec<u8>>,
106) {
107    let mut buf = vec![0u8; READ_CHUNK];
108    loop {
109        tokio::select! {
110            read = pipe.read(&mut buf) => match read {
111                Ok(0) | Err(_) => break,
112                Ok(len) => {
113                    if events.send(HostEvent::Bytes(buf[..len].to_vec())).is_err() {
114                        break;
115                    }
116                }
117            },
118            frame = wire.recv() => match frame {
119                Some(frame) if pipe.write_all(&frame).await.is_ok() => {}
120                _ => break,
121            },
122        }
123    }
124}
umshctl/lan.rs

umshctl/
lan.rs
