feat: add local/remote sub cache to publisher

examples/examples/z_local_pub_sub_thr.rs

@@ -0,0 +1,155 @@
+//
+// Copyright (c) 2023 ZettaScale Technology
+//
+// This program and the accompanying materials are made available under the
+// terms of the Eclipse Public License 2.0 which is available at
+// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+// which is available at https://www.apache.org/licenses/LICENSE-2.0.
+//
+// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+//
+// Contributors:
+//   ZettaScale Zenoh Team, <[email protected]>
+//
+
+use std::{convert::TryInto, time::Instant};
+
+use clap::Parser;
+use zenoh::{
+    bytes::ZBytes,
+    qos::{CongestionControl, Priority},
+    Wait,
+};
+use zenoh_examples::CommonArgs;
+
+struct Stats {
+    round_count: usize,
+    round_size: usize,
+    finished_rounds: usize,
+    round_start: Instant,
+    global_start: Option<Instant>,
+}
+impl Stats {
+    fn new(round_size: usize) -> Self {
+        Stats {
+            round_count: 0,
+            round_size,
+            finished_rounds: 0,
+            round_start: Instant::now(),
+            global_start: None,
+        }
+    }
+    fn increment(&mut self) {
+        if self.round_count == 0 {
+            self.round_start = Instant::now();
+            if self.global_start.is_none() {
+                self.global_start = Some(self.round_start)
+            }
+            self.round_count += 1;
+        } else if self.round_count < self.round_size {
+            self.round_count += 1;
+        } else {
+            self.print_round();
+            self.finished_rounds += 1;
+            self.round_count = 0;
+        }
+    }
+    fn print_round(&self) {
+        let elapsed = self.round_start.elapsed().as_secs_f64();
+        let throughput = (self.round_size as f64) / elapsed;
+        println!("{throughput} msg/s");
+    }
+}
+impl Drop for Stats {
+    fn drop(&mut self) {
+        let Some(global_start) = self.global_start else {
+            return;
+        };
+        let elapsed = global_start.elapsed().as_secs_f64();
+        let total = self.round_size * self.finished_rounds + self.round_count;
+        let throughput = total as f64 / elapsed;
+        println!("Received {total} messages over {elapsed:.2}s: {throughput}msg/s");
+    }
+}
+
+fn main() {
+    // initiate logging
+    zenoh::init_log_from_env_or("error");
+    let args = Args::parse();
+
+    let session = zenoh::open(args.common).wait().unwrap();
+
+    let key_expr = "test/thr";
+
+    let mut stats = Stats::new(args.number);
+    session
+        .declare_subscriber(key_expr)
+        .callback_mut(move |_sample| {
+            stats.increment();
+            if stats.finished_rounds >= args.samples {
+                std::process::exit(0)
+            }
+        })
+        .background()
+        .wait()
+        .unwrap();
+
+    let mut prio = Priority::DEFAULT;
+    if let Some(p) = args.priority {
+        prio = p.try_into().unwrap();
+    }
+
+    let publisher = session
+        .declare_publisher(key_expr)
+        .congestion_control(CongestionControl::Block)
+        .priority(prio)
+        .express(args.express)
+        .wait()
+        .unwrap();
+
+    println!("Press CTRL-C to quit...");
+    let payload_size = args.payload_size;
+    let data: ZBytes = (0..payload_size)
+        .map(|i| (i % 10) as u8)
+        .collect::<Vec<u8>>()
+        .into();
+    let mut count: usize = 0;
+    let mut start = std::time::Instant::now();
+    loop {
+        publisher.put(data.clone()).wait().unwrap();
+
+        if args.print {
+            if count < args.number {
+                count += 1;
+            } else {
+                let thpt = count as f64 / start.elapsed().as_secs_f64();
+                println!("{thpt} msg/s");
+                count = 0;
+                start = std::time::Instant::now();
+            }
+        }
+    }
+}
+
+#[derive(Parser, Clone, PartialEq, Eq, Hash, Debug)]
+struct Args {
+    #[arg(short, long, default_value = "10")]
+    /// Number of throughput measurements.
+    samples: usize,
+    /// express for sending data
+    #[arg(long, default_value = "false")]
+    express: bool,
+    /// Priority for sending data
+    #[arg(short, long)]
+    priority: Option<u8>,
+    /// Print the statistics
+    #[arg(short = 't', long)]
+    print: bool,
+    /// Number of messages in each throughput measurements
+    #[arg(short, long, default_value = "10000000")]
+    number: usize,
+    /// Sets the size of the payload to publish
+    payload_size: usize,
+    #[command(flatten)]
+    common: CommonArgs,
+}

zenoh/src/api/builders/publisher.rs

@@ -31,8 +31,8 @@ use crate::{
         bytes::{OptionZBytes, ZBytes},
         encoding::Encoding,
         key_expr::KeyExpr,
-        publisher::{Priority, Publisher},
-        sample::{Locality, SampleKind},
+        publisher::{Priority, Publisher, PublisherCache, PublisherCacheValue},
+        sample::Locality,
     },
     Session,
 };
@@ -209,11 +209,10 @@ impl Wait for PublicationBuilder<PublisherBuilder<'_, '_>, PublicationBuilderPut
     #[inline]
     fn wait(mut self) -> <Self as Resolvable>::To {
         self.publisher = self.publisher.apply_qos_overwrites();
-        self.publisher.session.0.resolve_put(
+        self.publisher.session.0.resolve_push(
+            &mut PublisherCacheValue::default(),
             &self.publisher.key_expr?,
-            self.kind.payload,
-            SampleKind::Put,
-            self.kind.encoding,
+            Some(self.kind),
             self.publisher.congestion_control,
             self.publisher.priority,
             self.publisher.is_express,
@@ -232,11 +231,10 @@ impl Wait for PublicationBuilder<PublisherBuilder<'_, '_>, PublicationBuilderDel
     #[inline]
     fn wait(mut self) -> <Self as Resolvable>::To {
         self.publisher = self.publisher.apply_qos_overwrites();
-        self.publisher.session.0.resolve_put(
+        self.publisher.session.0.resolve_push(
+            &mut PublisherCacheValue::default(),
             &self.publisher.key_expr?,
-            ZBytes::new(),
-            SampleKind::Delete,
-            Encoding::ZENOH_BYTES,
+            None,
             self.publisher.congestion_control,
             self.publisher.priority,
             self.publisher.is_express,
@@ -468,6 +466,7 @@ impl Wait for PublisherBuilder<'_, '_> {
             .declare_publisher_inner(key_expr.clone(), self.destination)?;
         Ok(Publisher {
             session: self.session.downgrade(),
+            cache: PublisherCache::default(),
             id,
             key_expr,
             encoding: self.encoding,
@@ -495,43 +494,45 @@ impl IntoFuture for PublisherBuilder<'_, '_> {
 
 impl Wait for PublicationBuilder<&Publisher<'_>, PublicationBuilderPut> {
     fn wait(self) -> <Self as Resolvable>::To {
-        self.publisher.session.resolve_put(
-            &self.publisher.key_expr,
-            self.kind.payload,
-            SampleKind::Put,
-            self.kind.encoding,
-            self.publisher.congestion_control,
-            self.publisher.priority,
-            self.publisher.is_express,
-            self.publisher.destination,
-            #[cfg(feature = "unstable")]
-            self.publisher.reliability,
-            self.timestamp,
-            #[cfg(feature = "unstable")]
-            self.source_info,
-            self.attachment,
-        )
+        self.publisher.cache.with_cache(|cached| {
+            self.publisher.session.resolve_push(
+                cached,
+                &self.publisher.key_expr,
+                Some(self.kind),
+                self.publisher.congestion_control,
+                self.publisher.priority,
+                self.publisher.is_express,
+                self.publisher.destination,
+                #[cfg(feature = "unstable")]
+                self.publisher.reliability,
+                self.timestamp,
+                #[cfg(feature = "unstable")]
+                self.source_info,
+                self.attachment,
+            )
+        })
     }
 }
 
 impl Wait for PublicationBuilder<&Publisher<'_>, PublicationBuilderDelete> {
     fn wait(self) -> <Self as Resolvable>::To {
-        self.publisher.session.resolve_put(
-            &self.publisher.key_expr,
-            ZBytes::new(),
-            SampleKind::Delete,
-            Encoding::ZENOH_BYTES,
-            self.publisher.congestion_control,
-            self.publisher.priority,
-            self.publisher.is_express,
-            self.publisher.destination,
-            #[cfg(feature = "unstable")]
-            self.publisher.reliability,
-            self.timestamp,
-            #[cfg(feature = "unstable")]
-            self.source_info,
-            self.attachment,
-        )
+        self.publisher.cache.with_cache(|cached| {
+            self.publisher.session.resolve_push(
+                cached,
+                &self.publisher.key_expr,
+                None,
+                self.publisher.congestion_control,
+                self.publisher.priority,
+                self.publisher.is_express,
+                self.publisher.destination,
+                #[cfg(feature = "unstable")]
+                self.publisher.reliability,
+                self.timestamp,
+                #[cfg(feature = "unstable")]
+                self.source_info,
+                self.attachment,
+            )
+        })
     }
 }