Less truncating when casting timeout values for Poll::poll

src/lib.rs

@@ -186,7 +186,14 @@ mod convert {
     const NANOS_PER_MILLI: u32 = 1_000_000;
     const MILLIS_PER_SEC: u64 = 1_000;
 
+    /// Convert a `Duration` to milliseconds, rounding up and saturating at
+    /// `u64::MAX`.
+    ///
+    /// The saturating is fine because `u64::MAX` milliseconds are still many
+    /// million years.
     pub fn millis(duration: Duration) -> u64 {
-        duration.as_secs() * MILLIS_PER_SEC + ((duration.subsec_nanos() / NANOS_PER_MILLI) as u64)
+        // Round up.
+        let millis = (duration.subsec_nanos() + NANOS_PER_MILLI - 1) / NANOS_PER_MILLI;
+        duration.as_secs().saturating_mul(MILLIS_PER_SEC).saturating_add(millis as u64)
     }
 }

src/poll.rs

@@ -1,4 +1,4 @@
-use {convert, sys, Evented, Token};
+use {sys, Evented, Token};
 use event::{self, EventSet, Event, PollOpt};
 use std::{fmt, io, mem, ptr, usize};
 use std::cell::UnsafeCell;
@@ -229,11 +229,11 @@ impl Poll {
         let timeout = if !self.readiness_queue.is_empty() {
             trace!("custom readiness queue has pending events");
             // Never block if the readiness queue has pending events
-            Some(0)
+            Some(Duration::from_millis(0))
         } else if !self.readiness_queue.prepare_for_sleep() {
-            Some(0)
+            Some(Duration::from_millis(0))
         } else {
-            timeout.map(|to| convert::millis(to) as usize)
+            timeout
         };
 
         // First get selector events

src/sys/unix/epoll.rs

@@ -1,9 +1,10 @@
-use {io, EventSet, PollOpt, Token};
+use {convert, io, EventSet, PollOpt, Token};
 use event::Event;
 use nix::sys::epoll::*;
 use nix::unistd::close;
 use std::os::unix::io::RawFd;
 use std::sync::atomic::{AtomicUsize, Ordering, ATOMIC_USIZE_INIT};
+use std::time::Duration;
 
 /// Each Selector has a globally unique(ish) ID associated with it. This ID
 /// gets tracked by `TcpStream`, `TcpListener`, etc... when they are first
@@ -35,13 +36,12 @@ impl Selector {
     }
 
     /// Wait for events from the OS
-    pub fn select(&self, evts: &mut Events, awakener: Token, timeout_ms: Option<usize>) -> io::Result<bool> {
+    pub fn select(&self, evts: &mut Events, awakener: Token, timeout: Option<Duration>) -> io::Result<bool> {
         use std::{cmp, i32, slice};
 
-        let timeout_ms = match timeout_ms {
-            None => -1 as i32,
-            Some(x) => cmp::min(i32::MAX as usize, x) as i32,
-        };
+        let timeout_ms = timeout
+            .map(|to| cmp::min(convert::millis(to), i32::MAX as u64) as i32)
+            .unwrap_or(-1);
 
         let dst = unsafe {
             slice::from_raw_parts_mut(

src/sys/unix/kqueue.rs

@@ -3,12 +3,13 @@ use event::{self, Event};
 use nix::unistd::close;
 use nix::sys::event::{EventFilter, EventFlag, FilterFlag, KEvent, kqueue, kevent, kevent_ts};
 use nix::sys::event::{EV_ADD, EV_CLEAR, EV_DELETE, EV_DISABLE, EV_ENABLE, EV_EOF, EV_ERROR, EV_ONESHOT};
-use libc::{timespec, time_t, c_long};
-use std::{fmt, slice};
+use libc::{timespec, time_t};
+use std::{cmp, fmt, slice};
 use std::cell::UnsafeCell;
 use std::os::unix::io::RawFd;
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicUsize, Ordering, ATOMIC_USIZE_INIT};
+use std::time::Duration;
 
 /// Each Selector has a globally unique(ish) ID associated with it. This ID
 /// gets tracked by `TcpStream`, `TcpListener`, etc... when they are first
@@ -46,10 +47,10 @@ impl Selector {
         self.id
     }
 
-    pub fn select(&self, evts: &mut Events, awakener: Token, timeout_ms: Option<usize>) -> io::Result<bool> {
-        let timeout = timeout_ms.map(|x| timespec {
-            tv_sec: (x / 1000) as time_t,
-            tv_nsec: ((x % 1000) * 1_000_000) as c_long
+    pub fn select(&self, evts: &mut Events, awakener: Token, timeout: Option<Duration>) -> io::Result<bool> {
+        let timeout = timeout.map(|to| timespec {
+            tv_sec: cmp::min(to.as_secs(), time_t::max_value() as u64) as time_t,
+            tv_nsec: to.subsec_nanos() as i64,
         });
 
         let cnt = try!(kevent_ts(self.kq, &[], evts.as_mut_slice(), timeout)

src/sys/windows/selector.rs

@@ -1,8 +1,11 @@
-use std::{fmt, io, mem};
 use std::cell::UnsafeCell;
+use std::u32;
 use std::os::windows::prelude::*;
 use std::sync::{Arc, Mutex};
+use std::time::Duration;
+use std::{cmp, fmt, io, mem};
 
+use convert;
 use winapi::*;
 use miow;
 use miow::iocp::{CompletionPort, CompletionStatus};
@@ -54,15 +57,15 @@ impl Selector {
         })
     }
 
-    pub fn select(&self, events: &mut Events, awakener: Token, timeout_ms: Option<usize>) -> io::Result<bool> {
+    pub fn select(&self, events: &mut Events, awakener: Token, timeout: Option<Duration>) -> io::Result<bool> {
         let mut ret = false;
 
         // If we have some deferred events then we only want to poll for I/O
         // events, so clamp the timeout to 0 in that case.
         let timeout = if !self.should_block() {
             Some(0)
         } else {
-            timeout_ms.map(|ms| ms as u32)
+            timeout.map(|to| cmp::min(convert::millis(to), u32::MAX as u64) as u32)
         };
 
         trace!("select; timeout={:?}", timeout);