[WIP] Set of attempts to optimize OTLP receiver/exporter

rust/otap-dataflow/Cargo.toml

@@ -121,6 +121,7 @@ weaver_resolved_schema = { git = "https://github.com/open-telemetry/weaver.git",
 weaver_resolver = { git = "https://github.com/open-telemetry/weaver.git", tag = "v0.17.0"}
 weaver_semconv = { git = "https://github.com/open-telemetry/weaver.git", tag = "v0.17.0"}
 zip = "=4.2.0"
+byte-unit = "5.1.6"
 
 [features]
 default = []

rust/otap-dataflow/configs/otlp-perf.yaml

@@ -13,7 +13,7 @@ nodes:
           - perf
         dispatch_strategy: round_robin
     config:
-      listening_addr: "127.0.0.1:4318"
+      listening_addr: "127.0.0.1:4317"
   perf:
     kind: exporter
     plugin_urn: "urn:otel:otap:perf:exporter"

rust/otap-dataflow/crates/config/Cargo.toml

@@ -20,3 +20,4 @@ serde_yaml = { workspace = true }
 miette = { workspace = true }
 urn = { workspace = true }
 schemars = { workspace = true }
+byte-unit = { workspace = true }

rust/otap-dataflow/crates/config/src/byte_units.rs

@@ -0,0 +1,126 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Support for byte units like "KB", "MB", "GB" in configuration.
+
+use byte_unit::Byte;
+use serde::de::Error as DeError;
+use serde::{Deserialize, Deserializer};
+
+#[derive(Deserialize)]
+#[serde(untagged)]
+enum Value {
+    Number(u64),
+    String(String),
+}
+
+/// Deserialize an optional byte size that can be specified either as a number (in bytes)
+/// or as a string with units (e.g., "1 KB", "2 MiB").
+pub fn deserialize<'de, D>(deserializer: D) -> Result<Option<u32>, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let value = Option::<Value>::deserialize(deserializer)?;
+    let Some(value) = value else {
+        return Ok(None);
+    };
+
+    let (bytes, repr) = match value {
+        Value::Number(value) => (value as u128, value.to_string()),
+        Value::String(text) => {
+            let parsed: Byte = text.parse().map_err(DeError::custom)?;
+            (parsed.as_u64() as u128, text)
+        }
+    };
+
+    if bytes > u32::MAX as u128 {
+        return Err(DeError::custom(format!(
+            "byte size {} exceeds u32::MAX ({} bytes)",
+            repr,
+            u32::MAX
+        )));
+    }
+
+    Ok(Some(bytes as u32))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::deserialize;
+    use serde::Deserialize;
+
+    #[derive(Debug, Deserialize)]
+    struct Holder {
+        #[serde(default, deserialize_with = "deserialize")]
+        value: Option<u32>,
+    }
+
+    fn de_yaml(input: &str) -> Result<Holder, serde_yaml::Error> {
+        serde_yaml::from_str::<Holder>(input)
+    }
+
+    #[test]
+    fn parses_number_as_bytes() {
+        let cfg = de_yaml("value: 1024").expect("should parse numeric bytes");
+        assert_eq!(cfg.value, Some(1024));
+    }
+
+    #[test]
+    fn parses_string_with_iec_units() {
+        // 1 KiB == 1024 bytes
+        let cfg = de_yaml("value: 1 KiB").expect("should parse 1 KiB");
+        assert_eq!(cfg.value, Some(1024));
+
+        // 2 MiB == 2 * 1024 * 1024 bytes
+        let cfg = de_yaml("value: '2 MiB'").expect("should parse 2 MiB");
+        assert_eq!(cfg.value, Some(2 * 1024 * 1024));
+    }
+
+    #[test]
+    fn parses_plain_string_number() {
+        let cfg = de_yaml("value: '2048'").expect("should parse plain numeric string");
+        assert_eq!(cfg.value, Some(2048));
+    }
+
+    #[test]
+    fn missing_value_is_none() {
+        let cfg = de_yaml("{}").expect("should parse with missing field as None");
+        assert_eq!(cfg.value, None);
+    }
+
+    #[test]
+    fn overflow_is_rejected() {
+        // 4 GiB == 4 * 1024^3 bytes = 4_294_967_296 > u32::MAX (4_294_967_295)
+        let err = de_yaml("value: 4 GiB").expect_err("should error for overflow");
+        let msg = err.to_string();
+        assert!(
+            msg.contains("exceeds u32::MAX"),
+            "unexpected error: {}",
+            msg
+        );
+    }
+
+    #[test]
+    fn parses_no_space_decimal_units() {
+        let cfg = de_yaml("value: 1KB").expect("should parse 1KB without space");
+        assert_eq!(cfg.value, Some(1000));
+
+        let cfg = de_yaml("value: 10MB").expect("should parse 10MB without space");
+        assert_eq!(cfg.value, Some(10_000_000));
+
+        // Lowercase 'b' should still be treated as bytes per crate behavior
+        let cfg = de_yaml("value: 1kb").expect("should parse 1kb as 1000 bits => 125 bytes");
+        assert_eq!(cfg.value, Some(125));
+    }
+
+    #[test]
+    fn parses_fractional_values_and_rounding() {
+        // Decimal unit with fraction
+        let cfg = de_yaml("value: '1.5 MB'").expect("should parse 1.5 MB");
+        assert_eq!(cfg.value, Some(1_500_000));
+
+        // Binary unit with fraction (exact)
+        let cfg = de_yaml("value: '0.5 KiB'").expect("should parse 0.5 KiB to 512 bytes");
+        assert_eq!(cfg.value, Some(512));
+    }
+}

rust/otap-dataflow/crates/config/src/lib.rs

@@ -13,6 +13,7 @@
 //! A pipeline is a collection of nodes interconnected in a directed acyclic graph (DAG).
 
 use std::borrow::Cow;
+pub mod byte_units;
 pub mod engine;
 pub mod error;
 pub mod experimental;

rust/otap-dataflow/crates/engine/src/local/processor.rs

@@ -44,6 +44,7 @@ use otap_df_telemetry::error::Error as TelemetryError;
 use otap_df_telemetry::metrics::{MetricSet, MetricSetHandler};
 use otap_df_telemetry::reporter::MetricsReporter;
 use std::collections::HashMap;
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 /// A trait for processors in the pipeline (!Send definition).
@@ -91,7 +92,10 @@ pub trait Processor<PData> {
 #[derive(Clone)]
 pub struct EffectHandler<PData> {
     pub(crate) core: EffectHandlerCore<PData>,
+    senders: Arc<EffectHandlerSenders<PData>>,
+}
 
+struct EffectHandlerSenders<PData> {
     /// A sender used to forward messages from the processor.
     /// Supports multiple named output ports.
     msg_senders: HashMap<PortName, LocalSender<PData>>,
@@ -120,11 +124,12 @@ impl<PData> EffectHandler<PData> {
             None
         };
 
-        EffectHandler {
-            core,
+        let senders = Arc::new(EffectHandlerSenders {
             msg_senders,
             default_sender,
-        }
+        });
+
+        EffectHandler { core, senders }
     }
 
     /// Returns the id of the processor associated with this handler.
@@ -136,7 +141,7 @@ impl<PData> EffectHandler<PData> {
     /// Returns the list of connected out ports for this processor.
     #[must_use]
     pub fn connected_ports(&self) -> Vec<PortName> {
-        self.msg_senders.keys().cloned().collect()
+        self.senders.msg_senders.keys().cloned().collect()
     }
 
     /// Sends a message to the next node(s) in the pipeline using the default port.
@@ -150,7 +155,7 @@ impl<PData> EffectHandler<PData> {
     /// if the default port is not configured.
     #[inline]
     pub async fn send_message(&self, data: PData) -> Result<(), TypedError<PData>> {
-        match &self.default_sender {
+        match &self.senders.default_sender {
             Some(sender) => sender
                 .send(data)
                 .await
@@ -178,7 +183,7 @@ impl<PData> EffectHandler<PData> {
         P: Into<PortName>,
     {
         let port_name: PortName = port.into();
-        match self.msg_senders.get(&port_name) {
+        match self.senders.msg_senders.get(&port_name) {
             Some(sender) => sender
                 .send(data)
                 .await

rust/otap-dataflow/crates/engine/src/local/receiver.rs

@@ -46,6 +46,7 @@ use otap_df_telemetry::metrics::{MetricSet, MetricSetHandler};
 use otap_df_telemetry::reporter::MetricsReporter;
 use std::collections::HashMap;
 use std::net::SocketAddr;
+use std::sync::Arc;
 use std::time::Duration;
 use tokio::net::{TcpListener, UdpSocket};
 
@@ -124,7 +125,10 @@ impl<PData> ControlChannel<PData> {
 #[derive(Clone)]
 pub struct EffectHandler<PData> {
     core: EffectHandlerCore<PData>,
+    senders: Arc<EffectHandlerSenders<PData>>,
+}
 
+struct EffectHandlerSenders<PData> {
     /// A sender used to forward messages from the receiver.
     /// Supports multiple named output ports.
     msg_senders: HashMap<PortName, LocalSender<PData>>,
@@ -155,11 +159,12 @@ impl<PData> EffectHandler<PData> {
             None
         };
 
-        EffectHandler {
-            core,
+        let senders = Arc::new(EffectHandlerSenders {
             msg_senders,
             default_sender,
-        }
+        });
+
+        EffectHandler { core, senders }
     }
 
     /// Returns the id of the receiver associated with this handler.
@@ -171,7 +176,7 @@ impl<PData> EffectHandler<PData> {
     /// Returns the list of connected out ports for this receiver.
     #[must_use]
     pub fn connected_ports(&self) -> Vec<PortName> {
-        self.msg_senders.keys().cloned().collect()
+        self.senders.msg_senders.keys().cloned().collect()
     }
 
     /// Sends a message to the next node(s) in the pipeline using the default port.
@@ -185,7 +190,7 @@ impl<PData> EffectHandler<PData> {
     /// [`TypedError::Error::ReceiverError`] if the default port is not configured.
     #[inline]
     pub async fn send_message(&self, data: PData) -> Result<(), TypedError<PData>> {
-        match &self.default_sender {
+        match &self.senders.default_sender {
             Some(sender) => sender
                 .send(data)
                 .await
@@ -213,7 +218,7 @@ impl<PData> EffectHandler<PData> {
         P: Into<PortName>,
     {
         let port_name: PortName = port.into();
-        match self.msg_senders.get(&port_name) {
+        match self.senders.msg_senders.get(&port_name) {
             Some(sender) => sender
                 .send(data)
                 .await

rust/otap-dataflow/crates/engine/src/shared/processor.rs

@@ -43,6 +43,7 @@ use otap_df_telemetry::error::Error as TelemetryError;
 use otap_df_telemetry::metrics::{MetricSet, MetricSetHandler};
 use otap_df_telemetry::reporter::MetricsReporter;
 use std::collections::HashMap;
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 /// A trait for processors in the pipeline (Send definition).
@@ -90,7 +91,10 @@ pub trait Processor<PData> {
 #[derive(Clone)]
 pub struct EffectHandler<PData> {
     pub(crate) core: EffectHandlerCore<PData>,
+    senders: Arc<EffectHandlerSenders<PData>>,
+}
 
+struct EffectHandlerSenders<PData> {
     /// A sender used to forward messages from the processor.
     /// Supports multiple named output ports.
     msg_senders: HashMap<PortName, SharedSender<PData>>,
@@ -119,11 +123,12 @@ impl<PData> EffectHandler<PData> {
             None
         };
 
-        EffectHandler {
-            core,
+        let senders = Arc::new(EffectHandlerSenders {
             msg_senders,
             default_sender,
-        }
+        });
+
+        EffectHandler { core, senders }
     }
 
     /// Returns the id of the processor associated with this handler.
@@ -135,7 +140,7 @@ impl<PData> EffectHandler<PData> {
     /// Returns the list of connected out ports for this processor.
     #[must_use]
     pub fn connected_ports(&self) -> Vec<PortName> {
-        self.msg_senders.keys().cloned().collect()
+        self.senders.msg_senders.keys().cloned().collect()
     }
 
     /// Sends a message to the next node(s) in the pipeline.
@@ -145,7 +150,7 @@ impl<PData> EffectHandler<PData> {
     /// Returns an [`Error::ProcessorError`] if the message could not be routed to a port.
     #[inline]
     pub async fn send_message(&self, data: PData) -> Result<(), TypedError<PData>> {
-        match &self.default_sender {
+        match &self.senders.default_sender {
             Some(sender) => sender
                 .send(data)
                 .await
@@ -168,7 +173,7 @@ impl<PData> EffectHandler<PData> {
         P: Into<PortName>,
     {
         let port_name: PortName = port.into();
-        match self.msg_senders.get(&port_name) {
+        match self.senders.msg_senders.get(&port_name) {
             Some(sender) => sender
                 .send(data)
                 .await