fix: Update PTP socket and clock init for statime-linux API

Co-authored-by: aider (gemini/gemini-2.5-pro-preview-05-06) <aider@aider.chat>
This commit is contained in:
Chaos Rogers 2025-07-10 17:50:58 +01:00
parent 277701a87f
commit b862e8d307
2 changed files with 38 additions and 12 deletions

View file

@ -17,3 +17,4 @@ tokio = { version = "1", features = ["full"] }
log = "0.4" log = "0.4"
env_logger = "0.11" env_logger = "0.11"
rand = "0.8" rand = "0.8"
socket2 = "0.4"

View file

@ -11,9 +11,12 @@ use statime::{
time::{Duration as PtpDuration, Interval}, time::{Duration as PtpDuration, Interval},
OverlayClock, PtpInstance, SharedClock, OverlayClock, PtpInstance, SharedClock,
}; };
use statime_linux::{clock::LinuxClock, socket::udp::LinuxUdpHandles}; use socket2::{Domain, Protocol, Socket, Type};
use statime_linux::clock::LinuxClock;
use std::net::SocketAddrV4;
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
use std::time::Duration; use std::time::Duration;
use tokio::net::UdpSocket;
use tokio::time::{sleep, Instant}; use tokio::time::{sleep, Instant};
pub async fn start_ptp_client(state: Arc<Mutex<LtcState>>, config: Arc<Mutex<Config>>) { pub async fn start_ptp_client(state: Arc<Mutex<LtcState>>, config: Arc<Mutex<Config>>) {
@ -94,12 +97,31 @@ async fn run_ptp_session(
}; };
// 4. Create Clock and Filter // 4. Create Clock and Filter
let clock = SharedClock::new(OverlayClock::new(LinuxClock::new())); let clock = SharedClock::new(OverlayClock::new(LinuxClock::open("/dev/ptp0")?));
let filter_config = 0.1; // Filter coefficient let filter_config = 0.1; // Filter coefficient
// 5. Create network handles // 5. Create network handles
let (mut event_handle, mut general_handle) = fn create_socket(
LinuxUdpHandles::new(&port_config, &interface)?; interface: &str,
port: u16,
) -> Result<UdpSocket, Box<dyn std::error::Error + Send + Sync>> {
let socket = Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP))?;
#[cfg(target_os = "linux")]
if let Err(e) = socket.bind_device(Some(interface.as_bytes())) {
log::warn!(
"Failed to bind to device '{}', maybe you need to be root? Error: {}",
interface,
e
);
}
socket.set_reuse_address(true)?;
let address = SocketAddrV4::new("0.0.0.0".parse().unwrap(), port);
socket.bind(&address.into())?;
Ok(UdpSocket::from_std(socket.into())?)
}
let event_socket = create_socket(&interface, 319)?;
let general_socket = create_socket(&interface, 320)?;
// 6. Add port and run BMCA // 6. Add port and run BMCA
let mut port = ptp_instance.add_port(port_config, filter_config, clock, thread_rng())?; let mut port = ptp_instance.add_port(port_config, filter_config, clock, thread_rng())?;
@ -109,24 +131,27 @@ async fn run_ptp_session(
let mut last_state_update = Instant::now(); let mut last_state_update = Instant::now();
let mut actions: Vec<_> = initial_actions.collect(); let mut actions: Vec<_> = initial_actions.collect();
let mut event_buf = [0u8; 1500];
let mut general_buf = [0u8; 1500];
loop { loop {
for action in actions { for action in actions {
match action { match action {
PortAction::SendEvent { PortAction::SendEvent {
data, data,
link_local: _, destination,
.. ..
} => { } => {
if let Err(e) = event_handle.send(data, None).await { if let Err(e) = event_socket.send_to(data, destination.into()).await {
log::error!("Error sending PTP event packet: {}", e); log::error!("Error sending PTP event packet: {}", e);
} }
} }
PortAction::SendGeneral { PortAction::SendGeneral {
data, data,
link_local: _, destination,
..
} => { } => {
if let Err(e) = general_handle.send(data, None).await { if let Err(e) = general_socket.send_to(data, destination.into()).await {
log::error!("Error sending PTP general packet: {}", e); log::error!("Error sending PTP general packet: {}", e);
} }
} }
@ -155,11 +180,11 @@ async fn run_ptp_session(
_ = tokio::time::sleep_until(timer_instant) => { _ = tokio::time::sleep_until(timer_instant) => {
actions.extend(running_port.handle_timer()); actions.extend(running_port.handle_timer());
} }
Ok((message, source_address)) = event_handle.recv() => { Ok((len, source_address)) = event_socket.recv_from(&mut event_buf) => {
actions.extend(running_port.handle_message(&message, source_address)); actions.extend(running_port.handle_message(&event_buf[..len], source_address));
} }
Ok((message, source_address)) = general_handle.recv() => { Ok((len, source_address)) = general_socket.recv_from(&mut general_buf) => {
actions.extend(running_port.handle_message(&message, source_address)); actions.extend(running_port.handle_message(&general_buf[..len], source_address));
} }
} }