CSPTP source setup

Author: rnijveld

ntp-proto/src/algorithm/kalman/mod.rs

@@ -435,6 +435,21 @@ impl<C: NtpClock, SourceId: Hash + Eq + Copy + Debug + Send + 'static> TimeSyncC
         )
     }
 
+    fn add_two_way_source(
+        &mut self,
+        id: Self::SourceId,
+        source_config: SourceConfig,
+    ) -> Self::NtpSourceController {
+        self.sources.insert(id, (None, false));
+        KalmanSourceController::new(
+            id,
+            self.algo_config,
+            None,
+            source_config,
+            AveragingBuffer::default(),
+        )
+    }
+
     fn remove_source(&mut self, id: SourceId) {
         self.sources.remove(&id);
     }

ntp-proto/src/algorithm/mod.rs

@@ -90,6 +90,12 @@ pub trait TimeSyncController: Sized + Send + 'static {
         measurement_noise_estimate: f64,
         period: Option<f64>,
     ) -> Self::OneWaySourceController;
+    /// Create a new two way source with given identity (used e.g. with CSPTP sources)
+    fn add_two_way_source(
+        &mut self,
+        id: Self::SourceId,
+        source_config: SourceConfig,
+    ) -> Self::NtpSourceController;
     /// Notify the controller that a previous source has gone
     fn remove_source(&mut self, id: Self::SourceId);
     /// Notify the controller that the status of a source (whether

ntp-proto/src/lib.rs

@@ -119,7 +119,7 @@ mod exports {
         AcceptSynchronizationError, Measurement, NtpSource, NtpSourceAction,
         NtpSourceActionIterator, NtpSourceSnapshot, NtpSourceUpdate, ObservableSourceState,
         OneWaySource, OneWaySourceSnapshot, OneWaySourceUpdate, ProtocolVersion, Reach,
-        SourceNtsData,
+        SourceNtsData, TwoWaySource,
     };
     pub use super::system::{
         System, SystemAction, SystemActionIterator, SystemSnapshot, SystemSourceUpdate,

ntp-proto/src/system.rs

@@ -6,7 +6,7 @@ use std::time::Duration;
 use std::{fmt::Debug, hash::Hash};
 
 use crate::packet::v5::server_reference_id::{BloomFilter, ServerId};
-use crate::source::{NtpSourceUpdate, SourceSnapshot};
+use crate::source::{NtpSourceUpdate, SourceSnapshot, TwoWaySource};
 use crate::{NtpTimestamp, OneWaySource, OneWaySourceUpdate};
 use crate::{
     algorithm::{StateUpdate, TimeSyncController},
@@ -324,6 +324,19 @@ impl<SourceId: Hash + Eq + Copy + Debug, Controller: TimeSyncController<SourceId
         ))
     }
 
+    #[allow(clippy::type_complexity)]
+    pub fn create_csptp_source(
+        &mut self,
+        id: SourceId,
+        source_config: SourceConfig,
+    ) -> Result<TwoWaySource<Controller::NtpSourceController>, <Controller::Clock as NtpClock>::Error>
+    {
+        self.ensure_controller_control()?;
+        let controller = self.controller.add_two_way_source(id, source_config);
+        self.sources.insert(id, None);
+        Ok(TwoWaySource::new(controller))
+    }
+
     pub fn handle_source_remove(
         &mut self,
         id: SourceId,

ntpd/src/daemon/config/mod.rs

@@ -448,6 +448,7 @@ impl Config {
                 NtpSourceConfig::Sock(_) => count += 1,
                 #[cfg(feature = "pps")]
                 NtpSourceConfig::Pps(_) => {} // PPS sources don't count
+                NtpSourceConfig::Csptp(_) => count += 1,
             }
         }
         count
@@ -483,6 +484,7 @@ impl Config {
             NtpSourceConfig::Sock(_) => false,
             #[cfg(feature = "pps")]
             NtpSourceConfig::Pps(_) => false,
+            NtpSourceConfig::Csptp(_) => false,
             NtpSourceConfig::Standard(config) => {
                 matches!(config.first.ntp_version, ProtocolVersion::V5)
             }

ntpd/src/daemon/config/ntp_source.rs

@@ -400,6 +400,12 @@ impl<'de> Deserialize<'de> for PpsSourceConfig {
     }
 }
 
+#[derive(Deserialize, Debug, PartialEq, Clone)]
+#[serde(rename_all = "kebab-case", deny_unknown_fields)]
+pub struct CsptpSourceConfig {
+    pub address: CsptpAddress,
+}
+
 #[derive(Debug, Deserialize, PartialEq, Clone)]
 #[serde(tag = "mode")]
 pub enum NtpSourceConfig {
@@ -417,6 +423,8 @@ pub enum NtpSourceConfig {
     #[cfg(feature = "pps")]
     #[serde(rename = "pps")]
     Pps(PpsSourceConfig),
+    #[serde(rename = "csptp")]
+    Csptp(CsptpSourceConfig),
 }
 
 /// A normalized address has a host and a port part. However, the host may be
@@ -463,6 +471,9 @@ pub struct NtpAddress(pub NormalizedAddress);
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct NtsKeAddress(pub NormalizedAddress);
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CsptpAddress(pub NormalizedAddress);
+
 impl<'de> Deserialize<'de> for NtpAddress {
     fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
     where
@@ -487,6 +498,18 @@ impl<'de> Deserialize<'de> for NtsKeAddress {
     }
 }
 
+impl<'de> Deserialize<'de> for CsptpAddress {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let s = String::deserialize(deserializer)?;
+        Ok(NormalizedAddress::from_string_csptp(s)
+            .map_err(serde::de::Error::custom)?
+            .into())
+    }
+}
+
 impl From<NormalizedAddress> for NtpAddress {
     fn from(addr: NormalizedAddress) -> Self {
         Self(addr)
@@ -499,6 +522,12 @@ impl From<NormalizedAddress> for NtsKeAddress {
     }
 }
 
+impl From<NormalizedAddress> for CsptpAddress {
+    fn from(addr: NormalizedAddress) -> Self {
+        Self(addr)
+    }
+}
+
 impl Deref for NtsKeAddress {
     type Target = NormalizedAddress;
 
@@ -515,9 +544,18 @@ impl Deref for NtpAddress {
     }
 }
 
+impl Deref for CsptpAddress {
+    type Target = NormalizedAddress;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
 impl NormalizedAddress {
     const NTP_DEFAULT_PORT: u16 = 123;
     const NTS_KE_DEFAULT_PORT: u16 = 4460;
+    const CSPTP_DEFAULT_PORT: u16 = 319;
 
     /// Specifically, this adds the `:123` port if no port is specified
     pub(crate) fn from_string_ntp(address: String) -> std::io::Result<Self> {
@@ -545,6 +583,19 @@ impl NormalizedAddress {
         })
     }
 
+    /// Specifically, this adds the `:319` port if no port is specified
+    pub(crate) fn from_string_csptp(address: String) -> std::io::Result<Self> {
+        let (server_name, port) = Self::from_string_help(address, Self::CSPTP_DEFAULT_PORT)?;
+
+        Ok(Self {
+            server_name,
+            port,
+
+            #[cfg(test)]
+            hardcoded_dns_resolve: HardcodedDnsResolve::default(),
+        })
+    }
+
     fn from_string_help(address: String, default_port: u16) -> std::io::Result<(String, u16)> {
         if address.split(':').count() > 2 {
             // IPv6, try to parse it as such
@@ -673,6 +724,7 @@ mod tests {
             NtpSourceConfig::Sock(_c) => "".to_string(),
             #[cfg(feature = "pps")]
             NtpSourceConfig::Pps(_c) => "".to_string(),
+            NtpSourceConfig::Csptp(c) => c.address.to_string(),
         }
     }
 

ntpd/src/daemon/csptp_source.rs

@@ -0,0 +1,60 @@
+use std::net::SocketAddr;
+
+use ntp_proto::{NtpClock, NtpDuration, SourceController, TwoWaySource};
+use timestamped_socket::{
+    interface::InterfaceName,
+    socket::{Connected, Socket},
+};
+use tracing::{Instrument, Span, instrument};
+
+use crate::daemon::{config::TimestampMode, ntp_source::SourceChannels, spawn::SourceId};
+
+pub(crate) struct CsptpSourceTask<
+    C: 'static + NtpClock + Send,
+    Controller: SourceController<MeasurementDelay = NtpDuration>,
+> {
+    index: SourceId,
+    clock: C,
+    interface: Option<InterfaceName>,
+    timestamp_mode: TimestampMode,
+    name: String,
+    source_addr: SocketAddr,
+    socket: Option<Socket<SocketAddr, Connected>>,
+    channels: SourceChannels<Controller::ControllerMessage, Controller::SourceMessage>,
+
+    source: TwoWaySource<Controller>,
+}
+
+impl<C: 'static + NtpClock + Send, Controller: SourceController<MeasurementDelay = NtpDuration>>
+    CsptpSourceTask<C, Controller>
+{
+    #[allow(clippy::too_many_arguments)]
+    #[instrument(level = tracing::Level::ERROR, name = "CSPTP Source", skip(timestamp_mode, clock, channels, source))]
+    pub fn spawn(
+        index: SourceId,
+        name: String,
+        source_addr: SocketAddr,
+        interface: Option<InterfaceName>,
+        clock: C,
+        timestamp_mode: TimestampMode,
+        channels: SourceChannels<Controller::ControllerMessage, Controller::SourceMessage>,
+        source: TwoWaySource<Controller>,
+    ) -> tokio::task::JoinHandle<()> {
+        tokio::spawn(
+            (async move {
+                let mut process = CsptpSourceTask {
+                    index,
+                    name,
+                    clock,
+                    channels,
+                    interface,
+                    timestamp_mode,
+                    source_addr,
+                    socket: None,
+                    source,
+                };
+            })
+            .instrument(Span::current()),
+        )
+    }
+}

ntpd/src/daemon/mod.rs

@@ -1,6 +1,7 @@
 mod clock;
 pub mod config;
 mod csptp_server;
+mod csptp_source;
 pub mod keyexchange;
 mod local_ip_provider;
 mod ntp_source;