refactor: use common methods for duration calculations

We use `process.hrtime` a lot throughout the codebase to track
durations for certain operations, often storing the result of it
on an object and using it later for duration calculation. This
patch changes that to depend on numbers rather than hrtime, which
would make it easy in the future to use something like
`performance.now` or even degrade into `Date.now` in environments
which don't support hi-res timers.

NODE-2651

Author: mbroadst

lib/change_stream.js

@@ -8,6 +8,8 @@ const Cursor = require('./cursor');
 const relayEvents = require('./core/utils').relayEvents;
 const maxWireVersion = require('./core/utils').maxWireVersion;
 const maybePromise = require('./utils').maybePromise;
+const now = require('./utils').now;
+const calculateDurationInMs = require('./utils').calculateDurationInMs;
 const AggregateOperation = require('./operations/aggregate');
 
 const kResumeQueue = Symbol('resumeQueue');
@@ -459,15 +461,21 @@ function applyKnownOptions(target, source, optionNames) {
 const SELECTION_TIMEOUT = 30000;
 function waitForTopologyConnected(topology, options, callback) {
   setTimeout(() => {
-    if (options && options.start == null) options.start = process.hrtime();
-    const start = options.start || process.hrtime();
+    if (options && options.start == null) {
+      options.start = now();
+    }
+
+    const start = options.start || now();
     const timeout = options.timeout || SELECTION_TIMEOUT;
     const readPreference = options.readPreference;
+    if (topology.isConnected({ readPreference })) {
+      return callback();
+    }
+
+    if (calculateDurationInMs(start) > timeout) {
+      return callback(new MongoError('Timed out waiting for connection'));
+    }
 
-    if (topology.isConnected({ readPreference })) return callback();
-    const hrElapsed = process.hrtime(start);
-    const elapsed = (hrElapsed[0] * 1e9 + hrElapsed[1]) / 1e6;
-    if (elapsed > timeout) return callback(new MongoError('Timed out waiting for connection'));
     waitForTopologyConnected(topology, options, callback);
   }, 500); // this is an arbitrary wait time to allow SDAM to transition
 }

lib/cmap/connection.js

@@ -11,6 +11,8 @@ const wp = require('../core/wireprotocol');
 const apm = require('../core/connection/apm');
 const updateSessionFromResponse = require('../core/sessions').updateSessionFromResponse;
 const uuidV4 = require('../core/utils').uuidV4;
+const now = require('../utils').now;
+const calculateDurationInMs = require('../utils').calculateDurationInMs;
 
 const kStream = Symbol('stream');
 const kQueue = Symbol('queue');
@@ -37,7 +39,7 @@ class Connection extends EventEmitter {
 
     this[kDescription] = new StreamDescription(this.address, options);
     this[kGeneration] = options.generation;
-    this[kLastUseTime] = Date.now();
+    this[kLastUseTime] = now();
 
     // retain a reference to an `AutoEncrypter` if present
     if (options.autoEncrypter) {
@@ -108,7 +110,7 @@ class Connection extends EventEmitter {
   }
 
   get idleTime() {
-    return Date.now() - this[kLastUseTime];
+    return calculateDurationInMs(this[kLastUseTime]);
   }
 
   get clusterTime() {
@@ -120,7 +122,7 @@ class Connection extends EventEmitter {
   }
 
   markAvailable() {
-    this[kLastUseTime] = Date.now();
+    this[kLastUseTime] = now();
   }
 
   destroy(options, callback) {
@@ -326,7 +328,7 @@ function write(command, options, callback) {
   if (this.monitorCommands) {
     this.emit('commandStarted', new apm.CommandStartedEvent(this, command));
 
-    operationDescription.started = process.hrtime();
+    operationDescription.started = now();
     operationDescription.cb = (err, reply) => {
       if (err) {
         this.emit(

lib/core/connection/apm.js

@@ -2,7 +2,7 @@
 const Msg = require('../connection/msg').Msg;
 const KillCursor = require('../connection/commands').KillCursor;
 const GetMore = require('../connection/commands').GetMore;
-const calculateDurationInMs = require('../utils').calculateDurationInMs;
+const calculateDurationInMs = require('../../utils').calculateDurationInMs;
 
 /** Commands that we want to redact because of the sensitive nature of their contents */
 const SENSITIVE_COMMANDS = new Set([

lib/core/connection/pool.js

@@ -21,6 +21,7 @@ const connect = require('./connect');
 const updateSessionFromResponse = require('../sessions').updateSessionFromResponse;
 const eachAsync = require('../utils').eachAsync;
 const makeStateMachine = require('../utils').makeStateMachine;
+const now = require('../../utils').now;
 
 const DISCONNECTED = 'disconnected';
 const CONNECTING = 'connecting';
@@ -898,7 +899,7 @@ Pool.prototype.write = function(command, options, cb) {
   if (self.options.monitorCommands) {
     this.emit('commandStarted', new apm.CommandStartedEvent(this, command));
 
-    operation.started = process.hrtime();
+    operation.started = now();
     operation.cb = (err, reply) => {
       if (err) {
         self.emit(

lib/core/sdam/monitor.js

@@ -1,14 +1,15 @@
 'use strict';
 
 const ServerType = require('./common').ServerType;
-const calculateDurationInMs = require('../utils').calculateDurationInMs;
 const EventEmitter = require('events');
 const connect = require('../connection/connect');
 const Connection = require('../../cmap/connection').Connection;
 const common = require('./common');
 const makeStateMachine = require('../utils').makeStateMachine;
 const MongoError = require('../error').MongoError;
 const makeInterruptableAsyncInterval = require('../../utils').makeInterruptableAsyncInterval;
+const calculateDurationInMs = require('../../utils').calculateDurationInMs;
+const now = require('../../utils').now;
 
 const sdamEvents = require('./events');
 const ServerHeartbeatStartedEvent = sdamEvents.ServerHeartbeatStartedEvent;
@@ -138,7 +139,7 @@ function checkServer(monitor, callback) {
     monitor[kConnection] = undefined;
   }
 
-  const start = process.hrtime();
+  const start = now();
   monitor.emit('serverHeartbeatStarted', new ServerHeartbeatStartedEvent(monitor.address));
 
   function failureHandler(err) {

lib/core/sdam/server_description.js

@@ -4,6 +4,7 @@ const arrayStrictEqual = require('../utils').arrayStrictEqual;
 const tagsStrictEqual = require('../utils').tagsStrictEqual;
 const errorStrictEqual = require('../utils').errorStrictEqual;
 const ServerType = require('./common').ServerType;
+const now = require('../../utils').now;
 
 const WRITABLE_SERVER_TYPES = new Set([
   ServerType.RSPrimary,
@@ -70,7 +71,7 @@ class ServerDescription {
     this.address = address;
     this.error = options.error;
     this.roundTripTime = options.roundTripTime || -1;
-    this.lastUpdateTime = Date.now();
+    this.lastUpdateTime = now();
     this.lastWriteDate = ismaster.lastWrite ? ismaster.lastWrite.lastWriteDate : null;
     this.opTime = ismaster.lastWrite ? ismaster.lastWrite.opTime : null;
     this.type = parseServerType(ismaster);

lib/core/sessions.js

@@ -17,7 +17,8 @@ const isTransactionCommand = require('./transactions').isTransactionCommand;
 const resolveClusterTime = require('./topologies/shared').resolveClusterTime;
 const isSharded = require('./wireprotocol/shared').isSharded;
 const maxWireVersion = require('./utils').maxWireVersion;
-
+const now = require('./../utils').now;
+const calculateDurationInMs = require('./../utils').calculateDurationInMs;
 const minWireVersionForShardedTransactions = 8;
 
 function assertAlive(session, callback) {
@@ -285,7 +286,7 @@ class ClientSession extends EventEmitter {
    * @param {TransactionOptions} [options] Optional settings for the transaction
    */
   withTransaction(fn, options) {
-    const startTime = Date.now();
+    const startTime = now();
     return attemptTransaction(this, startTime, fn, options);
   }
 }
@@ -301,7 +302,7 @@ const NON_DETERMINISTIC_WRITE_CONCERN_ERRORS = new Set([
 ]);
 
 function hasNotTimedOut(startTime, max) {
-  return Date.now() - startTime < max;
+  return calculateDurationInMs(startTime) < max;
 }
 
 function isUnknownTransactionCommitResult(err) {
@@ -558,7 +559,7 @@ function supportsRecoveryToken(session) {
 class ServerSession {
   constructor() {
     this.id = { id: new Binary(uuidV4(), Binary.SUBTYPE_UUID) };
-    this.lastUse = Date.now();
+    this.lastUse = now();
     this.txnNumber = 0;
     this.isDirty = false;
   }
@@ -573,7 +574,7 @@ class ServerSession {
     // Take the difference of the lastUse timestamp and now, which will result in a value in
     // milliseconds, and then convert milliseconds to minutes to compare to `sessionTimeoutMinutes`
     const idleTimeMinutes = Math.round(
-      (((Date.now() - this.lastUse) % 86400000) % 3600000) / 60000
+      ((calculateDurationInMs(this.lastUse) % 86400000) % 3600000) / 60000
     );
 
     return idleTimeMinutes > sessionTimeoutMinutes - 1;
@@ -708,7 +709,7 @@ function applySession(session, command, options) {
   }
 
   const serverSession = session.serverSession;
-  serverSession.lastUse = Date.now();
+  serverSession.lastUse = now();
   command.lsid = serverSession.id;
 
   // first apply non-transaction-specific sessions data

lib/core/topologies/replset.js

@@ -17,9 +17,10 @@ const isRetryableWritesSupported = require('./shared').isRetryableWritesSupporte
 const relayEvents = require('../utils').relayEvents;
 const isRetryableError = require('../error').isRetryableError;
 const BSON = retrieveBSON();
-const calculateDurationInMs = require('../utils').calculateDurationInMs;
 const getMMAPError = require('./shared').getMMAPError;
 const makeClientMetadata = require('../utils').makeClientMetadata;
+const now = require('../../utils').now;
+const calculateDurationInMs = require('../../utils').calculateDurationInMs;
 
 //
 // States
@@ -426,8 +427,7 @@ var pingServer = function(self, server, cb) {
       var latencyMS = new Date().getTime() - start;
 
       // Set the last updatedTime
-      var hrtime = process.hrtime();
-      server.lastUpdateTime = (hrtime[0] * 1e9 + hrtime[1]) / 1e6;
+      server.lastUpdateTime = now();
 
       // We had an error, remove it from the state
       if (err) {
@@ -1127,7 +1127,7 @@ ReplSet.prototype.selectServer = function(selector, options, callback) {
   }
 
   let lastError;
-  const start = process.hrtime();
+  const start = now();
   const _selectServer = () => {
     if (calculateDurationInMs(start) >= SERVER_SELECTION_TIMEOUT_MS) {
       if (lastError != null) {

lib/core/utils.js

@@ -13,17 +13,6 @@ const uuidV4 = () => {
   return result;
 };
 
-/**
- * Returns the duration calculated from two high resolution timers in milliseconds
- *
- * @param {Object} started A high resolution timestamp created from `process.hrtime()`
- * @returns {Number} The duration in milliseconds
- */
-const calculateDurationInMs = started => {
-  const hrtime = process.hrtime(started);
-  return (hrtime[0] * 1e9 + hrtime[1]) / 1e6;
-};
-
 /**
  * Relays events for a given listener and emitter
  *
@@ -261,7 +250,6 @@ const noop = () => {};
 
 module.exports = {
   uuidV4,
-  calculateDurationInMs,
   relayEvents,
   collationNotSupported,
   retrieveEJSON,

lib/utils.js

@@ -739,6 +739,16 @@ function now() {
   return Math.floor(hrtime[0] * 1000 + hrtime[1] / 1000000);
 }
 
+function calculateDurationInMs(started) {
+  if (typeof started !== 'number') {
+    console.trace('HERE');
+    throw TypeError('numeric value required to calculate duration');
+  }
+
+  const elapsed = now() - started;
+  return elapsed < 0 ? 0 : elapsed;
+}
+
 /**
  * Creates an interval timer which is able to be woken up sooner than
  * the interval. The timer will also debounce multiple calls to wake
@@ -849,5 +859,6 @@ module.exports = {
   makeCounter,
   maybePromise,
   now,
+  calculateDurationInMs,
   makeInterruptableAsyncInterval
 };