update kqueue, use Duration directly

Author: drahnr

src/pyroscope.rs

@@ -5,6 +5,7 @@ use std::{
         Arc, Condvar, Mutex,
     },
     thread::JoinHandle,
+    time::Duration,
 };
 
 use crate::{
@@ -33,7 +34,7 @@ pub struct PyroscopeConfig {
     /// Sample rate used in Hz
     pub sample_rate: i32,
     /// How long to accumulate data for before sending it upstream
-    pub accumulation_cycle: std::time::Duration,
+    pub accumulation_cycle: Duration,
     // TODO
     // log_level
     // auth_token
@@ -54,7 +55,7 @@ impl PyroscopeConfig {
             application_name: application_name.as_ref().to_owned(),
             tags: HashMap::new(),
             sample_rate: 100i32,
-            accumulation_cycle: std::time::Duration::from_secs(10),
+            accumulation_cycle: Duration::from_secs(10),
         }
     }
 
@@ -86,7 +87,7 @@ impl PyroscopeConfig {
     /// # Ok(())
     /// # }
     /// ```
-    pub fn accumulation_cycle(self, accumulation_cycle: std::time::Duration) -> Self {
+    pub fn accumulation_cycle(self, accumulation_cycle: Duration) -> Self {
         Self {
             accumulation_cycle,
             ..self
@@ -198,7 +199,7 @@ impl PyroscopeAgentBuilder {
     /// # Ok(())
     /// # }
     /// ```
-    pub fn accumulation_cycle(self, acc_cycle: impl Into<std::time::Duration>) -> Self {
+    pub fn accumulation_cycle(self, acc_cycle: impl Into<Duration>) -> Self {
         Self {
             config: self.config.accumulation_cycle(acc_cycle.into()),
             ..self

src/session.rs

@@ -1,7 +1,7 @@
 use std::{
     sync::mpsc::{sync_channel, Receiver, SyncSender},
-    thread,
-    thread::JoinHandle,
+    thread::{self, JoinHandle},
+    time::Duration,
 };
 
 use crate::pyroscope::PyroscopeConfig;
@@ -161,7 +161,7 @@ impl Session {
                 ("spyName", "pyroscope-rs"),
             ])
             .body(self.report)
-            .timeout(std::time::Duration::from_secs(10))
+            .timeout(Duration::from_secs(10))
             .send()?;
 
         Ok(())

src/timer/epoll.rs

@@ -14,7 +14,10 @@ use std::sync::{
     mpsc::{channel, Sender},
     Arc, Mutex,
 };
-use std::{thread, thread::JoinHandle};
+use std::{
+    time::Duration,
+    thread::{self, JoinHandle},
+};
 
 /// A thread that sends a notification every 10th second
 ///
@@ -35,7 +38,7 @@ pub struct Timer {
 
 impl Timer {
     /// Initialize Timer and run a thread to send events to attached listeners
-    pub fn initialize(cycle: std::time::Duration) -> Result<Self> {
+    pub fn initialize(cycle: Duration) -> Result<Self> {
         let txs = Arc::new(Mutex::new(Vec::new()));
 
         // Add a dummy tx so the below thread does not terminate early
@@ -76,7 +79,7 @@ impl Timer {
     }
 
     /// create and set a timer file descriptor
-    fn set_timerfd(cycle: std::time::Duration) -> Result<libc::c_int> {
+    fn set_timerfd(cycle: Duration) -> Result<libc::c_int> {
         // Set the timer to use the system time.
         let clockid: libc::clockid_t = libc::CLOCK_REALTIME;
         // Non-blocking file descriptor

src/timer/kqueue.rs

@@ -12,7 +12,10 @@ use std::sync::{
     mpsc::{channel, Receiver, Sender},
     Arc, Mutex,
 };
-use std::{thread, thread::JoinHandle};
+use std::{
+    time::Duration,
+    thread::{self, JoinHandle},
+};
 
 /// A thread that sends a notification every 10th second
 ///
@@ -33,7 +36,7 @@ pub struct Timer {
 
 impl Timer {
     /// Initialize Timer and run a thread to send events to attached listeners
-    pub fn initialize(self) -> Result<Self> {
+    pub fn initialize(self, accumulation_cycle: Duration) -> Result<Self> {
         let txs = Arc::clone(&self.txs);
 
         // Add Default tx
@@ -44,11 +47,11 @@ impl Timer {
 
         let handle = Some(thread::spawn(move || {
             // Wait for initial expiration
-            let initial_event = Timer::register_initial_expiration(kqueue)?;
+            let initial_event = Timer::register_initial_expiration(kqueue, Duration::from_millis(0))?;
             Timer::wait_event(kqueue, [initial_event].as_mut_ptr())?;
 
             // Register loop event
-            let loop_event = Timer::register_loop_expiration(kqueue)?;
+            let loop_event = Timer::register_loop_expiration(kqueue, accumulation_cycle)?;
 
             // Loop 10s
             loop {
@@ -132,14 +135,14 @@ impl Timer {
     }
 
     /// Register a loop expiration event
-    fn register_loop_expiration(kqueue: i32) -> Result<libc::kevent> {
+    fn register_loop_expiration(kqueue: i32, duration: Duration) -> Result<libc::kevent> {
         let loop_event = libc::kevent {
             ident: 1,
             filter: libc::EVFILT_TIMER,
             flags: libc::EV_ADD | libc::EV_ENABLE,
             fflags: 0,
-            data: 10000,
-            udata: 0 as *mut libc::c_void,
+            data: duration.as_millis(),
+            udata: std::ptr::null(),
         };
 
         // add loop event

tests/timer.rs

@@ -1,10 +1,11 @@
 use assert_matches::assert_matches;
 use pyroscope::timer::{Timer, TimerSignal};
+use std::time::Duration;
 
 #[test]
 fn test_timer() {
     // Initialize Timer
-    let mut timer = Timer::initialize(std::time::Duration::from_secs(10)).unwrap();
+    let mut timer = Timer::initialize(Duration::from_secs(10)).unwrap();
 
     // Attach a listener
     let (tx, rx) = std::sync::mpsc::channel();