rzmq 0.5.21

High performance, CPU and memory efficient, fully asynchronous, safe pure-Rust implementation of ZeroMQ (ØMQ) messaging with io_uring and TCP Cork acceleration on Linux.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

#[cfg(feature = "diagnostics")]
mod active {
  use std::time::{Duration, Instant};

  const MAX_SEGMENTS: usize = 8;

  pub struct ProfileGuard<'a> {
    profiler: &'a mut LoopProfiler,
    index: usize,
    start_time: Instant,
  }

  impl<'a> Drop for ProfileGuard<'a> {
    #[inline(always)]
    fn drop(&mut self) {
      let elapsed = self.start_time.elapsed();
      if self.index < MAX_SEGMENTS {
        self.profiler.durations[self.index] = elapsed;
      }
    }
  }

  pub struct LoopProfiler {
    pub(super) loop_start_time: Instant,
    pub(super) durations: [Duration; MAX_SEGMENTS],
    pub(super) labels: [&'static str; MAX_SEGMENTS],
    pub(super) threshold: Duration,
    pub(super) log_counter: usize,
    pub(super) log_interval: usize,
  }

  impl LoopProfiler {
    pub fn new(threshold: Duration, log_interval: usize) -> Self {
      Self {
        loop_start_time: Instant::now(),
        durations: [Duration::ZERO; MAX_SEGMENTS],
        labels: [""; MAX_SEGMENTS],
        threshold,
        log_counter: 0,
        log_interval,
      }
    }

    #[inline(always)]
    pub fn loop_start(&mut self) {
      self.loop_start_time = Instant::now();
      self.durations = [Duration::ZERO; MAX_SEGMENTS];
    }

    #[inline(always)]
    pub fn profile(&mut self, index: usize, label: &'static str) -> ProfileGuard<'_> {
      if index < MAX_SEGMENTS {
        self.labels[index] = label;
      }
      ProfileGuard {
        profiler: self,
        index,
        start_time: Instant::now(),
      }
    }

    pub fn log_and_reset_for_next_loop(&mut self) {
      let total_loop_duration = self.loop_start_time.elapsed();
      self.log_counter = self.log_counter.wrapping_add(1);

      if total_loop_duration > self.threshold
        || (self.log_interval > 0 && self.log_counter % self.log_interval == 0)
      {
        tracing::trace!(
          target: "rzmq::worker_latency",
          total_us = total_loop_duration.as_micros(),
          s0_label = self.labels[0], s0_us = self.durations[0].as_micros(),
          s1_label = self.labels[1], s1_us = self.durations[1].as_micros(),
          s2_label = self.labels[2], s2_us = self.durations[2].as_micros(),
          s3_label = self.labels[3], s3_us = self.durations[3].as_micros(),
          s4_label = self.labels[4], s4_us = self.durations[4].as_micros(),
          s5_label = self.labels[5], s5_us = self.durations[5].as_micros(),
          s6_label = self.labels[6], s6_us = self.durations[6].as_micros(),
          s7_label = self.labels[7], s7_us = self.durations[7].as_micros(),
        );
      }
      self.loop_start();
    }
  }
}

#[cfg(not(feature = "diagnostics"))]
mod stub {
  use std::time::Duration;

  #[derive(Debug, Clone, Copy, Default)]
  pub struct ProfileGuard;

  #[derive(Debug, Clone, Copy, Default)]
  pub struct LoopProfiler;

  impl LoopProfiler {
    #[inline(always)]
    pub fn new(_threshold: Duration, _log_interval: usize) -> Self {
      LoopProfiler
    }

    #[inline(always)]
    pub fn loop_start(&mut self) {}

    #[inline(always)]
    pub fn profile(&mut self, _index: usize, _label: &'static str) -> ProfileGuard {
      ProfileGuard
    }

    #[inline(always)]
    pub fn log_and_reset_for_next_loop(&mut self) {}
  }
}

#[cfg(feature = "diagnostics")]
pub(crate) use active::{LoopProfiler, ProfileGuard};

#[cfg(not(feature = "diagnostics"))]
pub(crate) use stub::{LoopProfiler, ProfileGuard};