Configuration handling (#57)

* configuration access fixed

* performance_receive and performance_receive benchmark samples added

* commenting fixed

* import fixed

Author: rex-schilasky

Cargo.toml

@@ -9,6 +9,8 @@ members = [
     "rustecal-types-protobuf",
     "rustecal-types-serde",
     "rustecal-types-string",
+    "rustecal-samples/benchmarks/performance_receive",
+    "rustecal-samples/benchmarks/performance_send",
     "rustecal-samples/monitoring/logging_receive",
     "rustecal-samples/monitoring/monitoring_receive",
     "rustecal-samples/pubsub/blob_send",

rustecal-core/src/configuration.rs

@@ -1,13 +1,13 @@
 //! configuration.rs
 //!
-//! Provides a safe Rust wrapper around the eCAL C Configuration API
+//! Provides a safe Rust wrapper around the eCAL C Configuration API.
 //!
 //! This module exposes a `Configuration` struct that manages an
 //! `eCAL_Configuration` instance via FFI. It supports initializing
 //! default settings or loading from a YAML file, and automatically
 //! frees the underlying C object on drop.
 
-use std::{ffi::{CStr, CString}, path::Path};
+use std::{ffi::{CStr, CString}, path::Path, ops::{Deref, DerefMut}};
 use thiserror::Error;
 use rustecal_sys as sys;
 
@@ -31,33 +31,25 @@ unsafe impl Sync for Configuration {}
 impl Configuration {
   /// Creates a new Configuration with default values loaded via eCAL_Configuration_InitFromConfig
   pub fn new() -> Result<Self, ConfigError> {
-    // Allocate new eCAL_Configuration
     let cfg = unsafe { sys::eCAL_Configuration_New() };
     if cfg.is_null() {
       return Err(ConfigError::NullPointer);
     }
-    // Initialize configuration with default settings
     unsafe { sys::eCAL_Configuration_InitFromConfig(cfg) };
     Ok(Configuration { inner: cfg })
   }
 
   /// Loads a Configuration from a YAML file at the given path
   pub fn from_file(path: &str) -> Result<Self, ConfigError> {
-    // Check that the file exists
     if !Path::new(path).exists() {
       return Err(ConfigError::InvalidPath(path.to_string()));
     }
-    // Convert Rust &str to CString
     let c_path = CString::new(path).map_err(|_| ConfigError::InvalidPath(path.to_string()))?;
-
-    // Allocate new eCAL_Configuration
     let cfg = unsafe { sys::eCAL_Configuration_New() };
     if cfg.is_null() {
       return Err(ConfigError::NullPointer);
     }
-    // Load configuration from file (void return type)
     unsafe { sys::eCAL_Configuration_InitFromFile(cfg, c_path.as_ptr()) };
-
     Ok(Configuration { inner: cfg })
   }
 
@@ -79,6 +71,21 @@ impl Configuration {
   }
 }
 
+/// Allow transparent access to the underlying C struct
+impl Deref for Configuration {
+  type Target = sys::eCAL_Configuration;
+  fn deref(&self) -> &Self::Target {
+    unsafe { &*self.inner }
+  }
+}
+
+/// Allow mutable access to the underlying C struct
+impl DerefMut for Configuration {
+  fn deref_mut(&mut self) -> &mut Self::Target {
+    unsafe { &mut *self.inner }
+  }
+}
+
 impl Drop for Configuration {
   /// Deletes the underlying eCAL_Configuration on drop
   fn drop(&mut self) {

rustecal-samples/benchmarks/performance_receive/.gitignore

@@ -0,0 +1 @@
+/target

rustecal-samples/benchmarks/performance_receive/Cargo.toml

@@ -0,0 +1,8 @@
+[package]
+name = "performance_receive"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+rustecal = { path = "../../../rustecal", features = ["pubsub"] }
+rustecal-types-bytes = { path = "../../../rustecal-types-bytes" }

rustecal-samples/benchmarks/performance_receive/src/main.rs

@@ -0,0 +1,74 @@
+//! A performance benchmark subscriber in Rust, modeled on the eCAL C++ sample.
+//!
+
+use std::{sync::{Arc, Mutex, atomic::Ordering}, thread, time::{Duration, Instant}};
+use rustecal::{Ecal, EcalComponents, TypedSubscriber};
+use rustecal::pubsub::typed_subscriber::Received;
+use rustecal_types_bytes::BytesMessage;
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // initialize eCAL
+    Ecal::initialize(Some("performance receive rust"), EcalComponents::DEFAULT, None)?;
+
+    // create a typed subscriber for raw bytes
+    let mut subscriber: TypedSubscriber<BytesMessage> = TypedSubscriber::new("Performance")?;
+
+    // shared counters & timer
+    let msgs  = Arc::new(std::sync::atomic::AtomicU64::new(0));
+    let bytes = Arc::new(std::sync::atomic::AtomicU64::new(0));
+    let start = Arc::new(Mutex::new(Instant::now()));
+
+    // register the receive‐callback
+    {
+        let msgs  = Arc::clone(&msgs);
+        let bytes = Arc::clone(&bytes);
+        let start = Arc::clone(&start);
+
+        subscriber.set_callback(move |msg: Received<BytesMessage>| {
+            let buffer = &msg.payload.data;
+            if buffer.is_empty() {
+                // nothing to do
+                return;
+            }
+
+            // update counters
+            msgs.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+            bytes.fetch_add(buffer.len() as u64, std::sync::atomic::Ordering::Relaxed);
+
+            // lock the timer, compute & maybe print
+            let mut start_lock = start.lock().unwrap();
+            let elapsed = start_lock.elapsed();
+            if elapsed >= Duration::from_secs(1) {
+                let m = msgs.swap(0, Ordering::Relaxed);
+                let b = bytes.swap(0, Ordering::Relaxed);
+
+                let secs = elapsed.as_secs_f64();
+                let kbyte_s    = (b as f64 / 1024.0) / secs;
+                let mbyte_s    = kbyte_s / 1024.0;
+                let gbyte_s    = mbyte_s / 1024.0;
+                let msg_s      = (m as f64) / secs;
+                let latency_us = (secs * 1e6) / (m as f64);
+
+                println!("Payload size      : {} kB", buffer.len() / 1024);
+                println!("Throughput (kB/s) : {:.0}", kbyte_s);
+                println!("Throughput (MB/s) : {:.2}", mbyte_s);
+                println!("Throughput (GB/s) : {:.3}", gbyte_s);
+                println!("Messages/s        : {:.0}", msg_s);
+                println!("Latency (µs)      : {:.2}", latency_us);
+                println!();
+
+                // reset the timer
+                *start_lock = Instant::now();
+            }
+        });
+    }
+
+    // keep the thread alive so callbacks can run
+    while Ecal::ok() {
+        thread::sleep(Duration::from_millis(100));
+    }
+
+    // clean up and finalize eCAL
+    Ecal::finalize();
+    Ok(())
+}

rustecal-samples/benchmarks/performance_send/.gitignore

@@ -0,0 +1 @@
+/target

rustecal-samples/benchmarks/performance_send/Cargo.toml

@@ -0,0 +1,8 @@
+[package]
+name = "performance_send"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+rustecal = { path = "../../../rustecal", features = ["pubsub"] }
+rustecal-types-bytes = { path = "../../../rustecal-types-bytes" }

rustecal-samples/benchmarks/performance_send/src/main.rs

@@ -0,0 +1,111 @@
+//! A performance benchmark publisher in Rust, modeled on the eCAL C++ sample.
+//!
+//! This will send messages of the given size in a tight loop, logging
+//! throughput every second.
+
+use std::{env, sync::Arc, time::{Duration, Instant}};
+use rustecal::{Ecal, EcalComponents, Configuration, TypedPublisher};
+use rustecal_types_bytes::BytesMessage;
+
+// performance settings
+const ZERO_COPY:               bool  = true;
+const BUFFER_COUNT:            u32   = 1;
+const ACKNOWLEDGE_TIMEOUT_MS:  i32   = 50;
+const PAYLOAD_SIZE_DEFAULT:    usize = 8 * 1024 * 1024;
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // parse payload size from CLI (or use default)
+    let args: Vec<String> = env::args().collect();
+    let mut payload_size = if args.len() > 1 {
+        args[1].parse::<usize>().unwrap_or(PAYLOAD_SIZE_DEFAULT)
+    } else {
+        PAYLOAD_SIZE_DEFAULT
+    };
+    if payload_size == 0 {
+        payload_size = 1;
+    }
+
+    // log performance settings
+    println!("Zero copy mode: {}", ZERO_COPY);
+    println!("Number of write buffers: {}", BUFFER_COUNT);
+    println!("Acknowledge timeout: {} ms", ACKNOWLEDGE_TIMEOUT_MS);
+    println!("Payload size: {} bytes", payload_size);
+    println!();
+
+    // prepare and tweak eCAL Configuration
+    let mut cfg = Configuration::new()?;
+    cfg.publisher.layer.shm.zero_copy_mode         = ZERO_COPY as i32;
+    cfg.publisher.layer.shm.memfile_buffer_count   = BUFFER_COUNT;
+    cfg.publisher.layer.shm.acknowledge_timeout_ms = ACKNOWLEDGE_TIMEOUT_MS as u32;
+
+    // initialize eCAL with custom config
+    Ecal::initialize(
+        Some("performance send rust"),
+        EcalComponents::DEFAULT,
+        Some(&cfg),
+    )
+    .expect("eCAL initialization failed");
+
+    // create payload buffer and publisher
+    let payload_vec: Vec<u8> = vec![0u8; payload_size];
+    let mut payload: Arc<[u8]> = Arc::from(payload_vec);
+    let publisher: TypedPublisher<BytesMessage> = TypedPublisher::new("Performance")?;
+
+    // benchmark loop
+    let mut msgs_sent  = 0u64;
+    let mut bytes_sent = 0u64;
+    let mut iterations = 0u64;
+    let mut last_log   = Instant::now();
+
+    // wait for at least one subscriber to be ready
+    while publisher.get_subscriber_count() == 0 {
+        println!("Waiting for performance receive to start ...");
+        std::thread::sleep(Duration::from_millis(1000));
+    }
+
+    // send loop
+    while Ecal::ok() {
+        // modify the payload for each message
+        {
+            let buf: &mut [u8] = Arc::make_mut(&mut payload);
+            let chr = (msgs_sent % 9 + 48) as u8;
+            buf[0..16].fill(chr);
+        }
+
+        let wrapped = BytesMessage { data: payload.clone() };
+        publisher.send(&wrapped);
+
+        msgs_sent += 1;
+        bytes_sent += payload_size as u64;
+        iterations += 1;
+
+        // every second, print statistics
+        if iterations % 2000 == 0 {
+            let elapsed = last_log.elapsed();
+            if elapsed >= Duration::from_secs(1) {
+                let secs       = elapsed.as_secs_f64();
+                let kbyte_s    = (bytes_sent as f64 / 1024.0) / secs;
+                let mbyte_s    = kbyte_s / 1024.0;
+                let gbyte_s    = mbyte_s / 1024.0;
+                let msg_s      = (msgs_sent as f64) / secs;
+                let latency_us = (secs * 1e6) / (msgs_sent as f64);
+
+                println!("Payload size      : {} kB", payload_size / 1024);
+                println!("Throughput (kB/s) : {:.0}", kbyte_s);
+                println!("Throughput (MB/s) : {:.2}", mbyte_s);
+                println!("Throughput (GB/s) : {:.3}", gbyte_s);
+                println!("Messages/s        : {:.0}", msg_s);
+                println!("Latency (µs)      : {:.2}", latency_us);
+                println!();
+
+                msgs_sent  = 0;
+                bytes_sent = 0;
+                last_log   = Instant::now();
+            }
+        }
+    }
+
+    // clean up and finalize eCAL
+    Ecal::finalize();
+    Ok(())
+}

rustecal-samples/monitoring/logging_receive/src/main.rs

@@ -3,18 +3,18 @@ use rustecal_core::log::Log;
 use std::{thread, time::Duration};
 
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    // Initialize eCAL (only the logging component)
+    // initialize eCAL (only the logging component)
     Ecal::initialize(Some("logging receive sample"), EcalComponents::LOGGING, None)?;
     println!("eCAL initialized. Entering logging loop…");
 
     while Ecal::ok() {
-        // Fetch whatever log entries are available
+        // fetch whatever log entries are available
         let entries = Log::get_logging()?;
 
         println!("=== Logging Snapshot ===\n");
         println!("Entries:\n{:#?}", entries);
 
-        // Sleep before next poll
+        // sleep before next poll
         thread::sleep(Duration::from_secs(1));
     }
 

rustecal-samples/monitoring/monitoring_receive/src/main.rs

@@ -3,7 +3,7 @@ use rustecal_core::monitoring::Monitoring;
 use std::{thread, time::Duration};
 
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    // Initialize eCAL (only the monitoring component)
+    // initialize eCAL (only the monitoring component)
     Ecal::initialize(Some("monitoring receive sample"), EcalComponents::MONITORING, None)?;
     println!("eCAL initialized. Entering monitoring loop…");
 
@@ -18,7 +18,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         println!("\nServers:\n{:#?}", snap.servers);
         println!("\nClients:\n{:#?}", snap.clients);
 
-        // Sleep before next poll
+        // sleep before next poll
         thread::sleep(Duration::from_secs(1));
     }