RUST-254 Implement minHeartbeatFrequencyMS (#102)

Author: saghm

src/client/options/mod.rs

@@ -32,6 +32,7 @@ use crate::{
     concern::{Acknowledgment, ReadConcern, WriteConcern},
     error::{ErrorKind, Result},
     event::{cmap::CmapEventHandler, command::CommandEventHandler},
+    sdam::MIN_HEARTBEAT_FREQUENCY,
     selection_criteria::{ReadPreference, SelectionCriteria, TagSet},
     srv::SrvResolver,
 };
@@ -1092,7 +1093,7 @@ impl ClientOptionsParser {
             k @ "heartbeatfrequencyms" => {
                 let duration = get_duration!(value, k);
 
-                if duration < 500 {
+                if duration < MIN_HEARTBEAT_FREQUENCY.num_milliseconds() as u64 {
                     return Err(ErrorKind::ArgumentError {
                         message: format!(
                             "'heartbeatFrequencyMS' must be at least 500, but {} was given",

src/sdam/mod.rs

@@ -7,5 +7,6 @@ pub use self::public::{ServerInfo, ServerType};
 
 pub(crate) use self::{
     description::{server::ServerDescription, topology::TopologyDescription},
+    monitor::MIN_HEARTBEAT_FREQUENCY,
     state::{server::Server, update_topology, Topology, TopologyUpdateCondvar},
 };

src/sdam/monitor.rs

@@ -4,6 +4,7 @@ use std::{
 };
 
 use bson::{bson, doc};
+use lazy_static::lazy_static;
 use time::PreciseTime;
 
 use super::{
@@ -19,6 +20,11 @@ use crate::{
 
 const DEFAULT_HEARTBEAT_FREQUENCY: Duration = Duration::from_secs(10);
 
+lazy_static! {
+    // Unfortunately, the `time` crate has not yet updated to make the `Duration` constructors `const`, so we have to use lazy_static.
+    pub(crate) static ref MIN_HEARTBEAT_FREQUENCY: time::Duration = time::Duration::milliseconds(500);
+}
+
 /// Starts a monitoring thread associated with a given Server. A weak reference is used to ensure
 /// that the monitoring thread doesn't keep the server alive after it's been removed from the
 /// topology or the client has been dropped.
@@ -37,9 +43,28 @@ pub(super) fn monitor_server(
                 None => return,
             };
 
-            match server.upgrade() {
+            let last_check = PreciseTime::now();
+
+            let timed_out = match server.upgrade() {
                 Some(server) => server.wait_timeout(heartbeat_frequency),
                 None => return,
+            };
+
+            if !timed_out {
+                let duration_since_last_check = last_check.to(PreciseTime::now());
+
+                if duration_since_last_check < *MIN_HEARTBEAT_FREQUENCY {
+                    let remaining_time = *MIN_HEARTBEAT_FREQUENCY - duration_since_last_check;
+
+                    // Since MIN_HEARTBEAT_FREQUENCY is 500 and `duration_since_last_check` is less
+                    // than it but still positive, we can be sure that the time::Duration can be
+                    // successfully converted to a std::time::Duration. However, in the case of some
+                    // bug causing this not to be true, rather than panicking the monitoring thread,
+                    // we instead just don't sleep and proceed to checking the server a bit early.
+                    if let Ok(remaining_time) = remaining_time.to_std() {
+                        std::thread::sleep(remaining_time);
+                    }
+                }
             }
         }
     });

src/sdam/state/server.rs

@@ -61,11 +61,14 @@ impl Server {
         self.topology.upgrade()
     }
 
-    pub(crate) fn wait_timeout(&self, duration: Duration) {
-        let _ = self
-            .condvar
+    /// Waits until either the server is requested to do a topology check or until `duration` has
+    /// elapsed. Returns `true` if `duration` has elapsed and `false` otherwise.
+    pub(crate) fn wait_timeout(&self, duration: Duration) -> bool {
+        self.condvar
             .wait_timeout(self.condvar_mutex.lock().unwrap(), duration)
-            .unwrap();
+            .unwrap()
+            .1
+            .timed_out()
     }
 
     pub(crate) fn request_topology_check(&self) {