Merge pull request #9 from stjepang/timeout-test

Add timeout test

Cargo.toml

@@ -18,6 +18,7 @@ libc = "0.2.74"
 
 [dev-dependencies]
 doc-comment = "0.3"
+easy-parallel = "3.1.0"
 
 [target.'cfg(windows)'.dependencies]
 # Patched version of wepoll that can be notified by PostQueuedCompletionStatus.

src/epoll.rs

@@ -132,11 +132,18 @@ impl Poller {
 
     /// Waits for I/O events with an optional timeout.
     pub fn wait(&self, events: &mut Events, timeout: Option<Duration>) -> io::Result<()> {
-        // Convert the `Duration` to `libc::timespec`.
-        let timeout = timeout.map(|t| libc::timespec {
-            tv_sec: t.as_secs() as libc::time_t,
-            tv_nsec: t.subsec_nanos() as libc::c_long,
-        });
+        // Configure the timeout using timerfd.
+        let new_val = libc::itimerspec {
+            it_interval: TS_ZERO,
+            it_value: match timeout {
+                None => TS_ZERO,
+                Some(t) => libc::timespec {
+                    tv_sec: t.as_secs() as libc::time_t,
+                    tv_nsec: t.subsec_nanos() as libc::c_long,
+                },
+            },
+        };
+        syscall!(timerfd_settime(self.timer_fd, 0, &new_val, ptr::null_mut()))?;
 
         // Set interest in timerfd.
         self.interest(
@@ -148,19 +155,21 @@ impl Poller {
             },
         )?;
 
-        // Configure the timeout using timerfd.
-        let new_val = libc::itimerspec {
-            it_interval: TS_ZERO,
-            it_value: timeout.unwrap_or(TS_ZERO),
+        // Timeout in milliseconds for epoll.
+        let timeout_ms = if timeout == Some(Duration::from_secs(0)) {
+            // This is a non-blocking call - use zero as the timeout.
+            0
+        } else {
+            // This is a blocking call - rely on timerfd to trigger the timeout.
+            -1
         };
-        syscall!(timerfd_settime(self.timer_fd, 0, &new_val, ptr::null_mut()))?;
 
         // Wait for I/O events.
         let res = syscall!(epoll_wait(
             self.epoll_fd,
             events.list.as_mut_ptr() as *mut libc::epoll_event,
             events.list.len() as libc::c_int,
-            -1,
+            timeout_ms,
         ))?;
         events.len = res as usize;
 

src/wepoll.rs

@@ -117,12 +117,11 @@ impl Poller {
     /// If a notification occurs, this method will return but the notification event will not be
     /// included in the `events` list nor contribute to the returned count.
     pub fn wait(&self, events: &mut Events, timeout: Option<Duration>) -> io::Result<()> {
-        let mut now = Instant::now();
-        let deadline = timeout.map(|t| now + t);
+        let deadline = timeout.map(|t| Instant::now() + t);
 
         loop {
             // Convert the timeout to milliseconds.
-            let timeout_ms = match deadline.map(|d| d - now) {
+            let timeout_ms = match deadline.map(|d| d.saturating_duration_since(Instant::now())) {
                 None => -1,
                 Some(t) => {
                     // Round up to a whole millisecond.
@@ -142,24 +141,10 @@ impl Poller {
                 timeout_ms,
             ))? as usize;
 
-            // If there was a notification, break.
-            if self.notified.swap(false, Ordering::SeqCst) {
+            // Break if there was a notification or at least one event, or if deadline is reached.
+            if self.notified.swap(false, Ordering::SeqCst) || events.len > 0 || timeout_ms == 0 {
                 break;
             }
-
-            // If there are any events at all, break.
-            if events.len > 0 {
-                break;
-            }
-
-            now = Instant::now();
-
-            // Check for timeout.
-            if let Some(d) = deadline {
-                if now >= d {
-                    break;
-                }
-            }
         }
 
         Ok(())

tests/notify.rs

@@ -0,0 +1,36 @@
+use std::io;
+use std::thread;
+use std::time::Duration;
+
+use easy_parallel::Parallel;
+use polling::Poller;
+
+#[test]
+fn simple() -> io::Result<()> {
+    let poller = Poller::new()?;
+    let mut events = Vec::new();
+
+    for _ in 0..10 {
+        poller.notify()?;
+        poller.wait(&mut events, None)?;
+    }
+
+    Ok(())
+}
+
+#[test]
+fn concurrent() -> io::Result<()> {
+    let poller = Poller::new()?;
+    let mut events = Vec::new();
+
+    for _ in 0..2 {
+        Parallel::new()
+            .add(|| {
+                thread::sleep(Duration::from_secs(0));
+                poller.notify().unwrap();
+            })
+            .finish(|| poller.wait(&mut events, None).unwrap());
+    }
+
+    Ok(())
+}

tests/timeout.rs

@@ -0,0 +1,32 @@
+use std::io;
+use std::time::{Duration, Instant};
+
+use polling::Poller;
+
+#[test]
+fn twice() -> io::Result<()> {
+    let poller = Poller::new()?;
+    let mut events = Vec::new();
+
+    for _ in 0..2 {
+        let start = Instant::now();
+        poller.wait(&mut events, Some(Duration::from_secs(1)))?;
+        let elapsed = start.elapsed();
+
+        assert!(elapsed >= Duration::from_secs(1));
+    }
+
+    Ok(())
+}
+
+#[test]
+fn non_blocking() -> io::Result<()> {
+    let poller = Poller::new()?;
+    let mut events = Vec::new();
+
+    for _ in 0..100 {
+        poller.wait(&mut events, Some(Duration::from_secs(0)))?;
+    }
+
+    Ok(())
+}