perf(NODE-4727): Improve performance of buffering and cursors (mongodb#3447)

Author: nbbeeken

package-lock.json

@@ -26,7 +26,6 @@
   },
   "dependencies": {
     "bson": "^4.7.0",
-    "denque": "^2.1.0",
     "mongodb-connection-string-url": "^2.5.4",
     "socks": "^2.7.1"
   },

package.json

@@ -1,4 +1,3 @@
-import Denque = require('denque');
 import { clearTimeout, setTimeout } from 'timers';
 
 import type { ObjectId } from '../bson';
@@ -26,7 +25,7 @@ import {
 import { Logger } from '../logger';
 import { CancellationToken, TypedEventEmitter } from '../mongo_types';
 import type { Server } from '../sdam/server';
-import { Callback, eachAsync, makeCounter } from '../utils';
+import { Callback, eachAsync, List, makeCounter } from '../utils';
 import { connect } from './connect';
 import { Connection, ConnectionEvents, ConnectionOptions } from './connection';
 import {
@@ -137,7 +136,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
   [kPoolState]: typeof PoolState[keyof typeof PoolState];
   [kServer]: Server;
   [kLogger]: Logger;
-  [kConnections]: Denque<Connection>;
+  [kConnections]: List<Connection>;
   [kPending]: number;
   [kCheckedOut]: Set<Connection>;
   [kMinPoolSizeTimer]?: NodeJS.Timeout;
@@ -151,7 +150,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
   [kServiceGenerations]: Map<string, number>;
   [kConnectionCounter]: Generator<number>;
   [kCancellationToken]: CancellationToken;
-  [kWaitQueue]: Denque<WaitQueueMember>;
+  [kWaitQueue]: List<WaitQueueMember>;
   [kMetrics]: ConnectionPoolMetrics;
   [kProcessingWaitQueue]: boolean;
 
@@ -236,7 +235,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
     this[kPoolState] = PoolState.paused;
     this[kServer] = server;
     this[kLogger] = new Logger('ConnectionPool');
-    this[kConnections] = new Denque();
+    this[kConnections] = new List();
     this[kPending] = 0;
     this[kCheckedOut] = new Set();
     this[kMinPoolSizeTimer] = undefined;
@@ -245,7 +244,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
     this[kConnectionCounter] = makeCounter(1);
     this[kCancellationToken] = new CancellationToken();
     this[kCancellationToken].setMaxListeners(Infinity);
-    this[kWaitQueue] = new Denque();
+    this[kWaitQueue] = new List();
     this[kMetrics] = new ConnectionPoolMetrics();
     this[kProcessingWaitQueue] = false;
 
@@ -659,12 +658,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
       return;
     }
 
-    for (let i = 0; i < this[kConnections].length; i++) {
-      const connection = this[kConnections].peekAt(i);
-      if (connection && this.connectionIsPerished(connection)) {
-        this[kConnections].removeOne(i);
-      }
-    }
+    this[kConnections].prune(connection => this.connectionIsPerished(connection));
 
     if (
       this.totalConnectionCount < minPoolSize &&
@@ -705,7 +699,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
     this[kProcessingWaitQueue] = true;
 
     while (this.waitQueueSize) {
-      const waitQueueMember = this[kWaitQueue].peekFront();
+      const waitQueueMember = this[kWaitQueue].first();
       if (!waitQueueMember) {
         this[kWaitQueue].shift();
         continue;

src/cmap/connection_pool.ts

@@ -132,31 +132,28 @@ function canCompress(command: WriteProtocolMessageType) {
   return !uncompressibleCommands.has(commandName);
 }
 
-function processIncomingData(stream: MessageStream, callback: Callback<Buffer>) {
+function processIncomingData(stream: MessageStream, callback: Callback<Buffer>): void {
   const buffer = stream[kBuffer];
-  if (buffer.length < 4) {
-    callback();
-    return;
+  const sizeOfMessage = buffer.getInt32();
+
+  if (sizeOfMessage == null) {
+    return callback();
   }
 
-  const sizeOfMessage = buffer.peek(4).readInt32LE();
   if (sizeOfMessage < 0) {
-    callback(new MongoParseError(`Invalid message size: ${sizeOfMessage}`));
-    return;
+    return callback(new MongoParseError(`Invalid message size: ${sizeOfMessage}`));
   }
 
   if (sizeOfMessage > stream.maxBsonMessageSize) {
-    callback(
+    return callback(
       new MongoParseError(
         `Invalid message size: ${sizeOfMessage}, max allowed: ${stream.maxBsonMessageSize}`
       )
     );
-    return;
   }
 
   if (sizeOfMessage > buffer.length) {
-    callback();
-    return;
+    return callback();
   }
 
   const message = buffer.read(sizeOfMessage);
@@ -170,36 +167,31 @@ function processIncomingData(stream: MessageStream, callback: Callback<Buffer>)
   const monitorHasAnotherHello = () => {
     if (stream.isMonitoringConnection) {
       // Can we read the next message size?
-      if (buffer.length >= 4) {
-        const sizeOfMessage = buffer.peek(4).readInt32LE();
-        if (sizeOfMessage <= buffer.length) {
-          return true;
-        }
+      const sizeOfMessage = buffer.getInt32();
+      if (sizeOfMessage != null && sizeOfMessage <= buffer.length) {
+        return true;
       }
     }
     return false;
   };
 
   let ResponseType = messageHeader.opCode === OP_MSG ? BinMsg : Response;
   if (messageHeader.opCode !== OP_COMPRESSED) {
-    const messageBody = message.slice(MESSAGE_HEADER_SIZE);
+    const messageBody = message.subarray(MESSAGE_HEADER_SIZE);
 
     // If we are a monitoring connection message stream and
     // there is more in the buffer that can be read, skip processing since we
     // want the last hello command response that is in the buffer.
     if (monitorHasAnotherHello()) {
-      processIncomingData(stream, callback);
-    } else {
-      stream.emit('message', new ResponseType(message, messageHeader, messageBody));
-
-      if (buffer.length >= 4) {
-        processIncomingData(stream, callback);
-      } else {
-        callback();
-      }
+      return processIncomingData(stream, callback);
     }
 
-    return;
+    stream.emit('message', new ResponseType(message, messageHeader, messageBody));
+
+    if (buffer.length >= 4) {
+      return processIncomingData(stream, callback);
+    }
+    return callback();
   }
 
   messageHeader.fromCompressed = true;
@@ -210,33 +202,28 @@ function processIncomingData(stream: MessageStream, callback: Callback<Buffer>)
 
   // recalculate based on wrapped opcode
   ResponseType = messageHeader.opCode === OP_MSG ? BinMsg : Response;
-  decompress(compressorID, compressedBuffer, (err, messageBody) => {
+  return decompress(compressorID, compressedBuffer, (err, messageBody) => {
     if (err || !messageBody) {
-      callback(err);
-      return;
+      return callback(err);
     }
 
     if (messageBody.length !== messageHeader.length) {
-      callback(
+      return callback(
         new MongoDecompressionError('Message body and message header must be the same length')
       );
-
-      return;
     }
 
     // If we are a monitoring connection message stream and
     // there is more in the buffer that can be read, skip processing since we
     // want the last hello command response that is in the buffer.
     if (monitorHasAnotherHello()) {
-      processIncomingData(stream, callback);
-    } else {
-      stream.emit('message', new ResponseType(message, messageHeader, messageBody));
-
-      if (buffer.length >= 4) {
-        processIncomingData(stream, callback);
-      } else {
-        callback();
-      }
+      return processIncomingData(stream, callback);
+    }
+    stream.emit('message', new ResponseType(message, messageHeader, messageBody));
+
+    if (buffer.length >= 4) {
+      return processIncomingData(stream, callback);
     }
+    return callback();
   });
 }

src/cmap/message_stream.ts

@@ -21,7 +21,7 @@ import { ReadConcern, ReadConcernLike } from '../read_concern';
 import { ReadPreference, ReadPreferenceLike } from '../read_preference';
 import type { Server } from '../sdam/server';
 import { ClientSession, maybeClearPinnedConnection } from '../sessions';
-import { Callback, maybeCallback, MongoDBNamespace, ns } from '../utils';
+import { Callback, List, maybeCallback, MongoDBNamespace, ns } from '../utils';
 
 /** @internal */
 const kId = Symbol('id');
@@ -143,15 +143,15 @@ export abstract class AbstractCursor<
   CursorEvents extends AbstractCursorEvents = AbstractCursorEvents
 > extends TypedEventEmitter<CursorEvents> {
   /** @internal */
-  [kId]?: Long;
+  [kId]: Long | null;
   /** @internal */
   [kSession]: ClientSession;
   /** @internal */
   [kServer]?: Server;
   /** @internal */
   [kNamespace]: MongoDBNamespace;
   /** @internal */
-  [kDocuments]: TSchema[];
+  [kDocuments]: List<TSchema>;
   /** @internal */
   [kClient]: MongoClient;
   /** @internal */
@@ -181,7 +181,8 @@ export abstract class AbstractCursor<
     }
     this[kClient] = client;
     this[kNamespace] = namespace;
-    this[kDocuments] = [];
+    this[kId] = null;
+    this[kDocuments] = new List();
     this[kInitialized] = false;
     this[kClosed] = false;
     this[kKilled] = false;
@@ -224,7 +225,7 @@ export abstract class AbstractCursor<
   }
 
   get id(): Long | undefined {
-    return this[kId];
+    return this[kId] ?? undefined;
   }
 
   /** @internal */
@@ -282,7 +283,17 @@ export abstract class AbstractCursor<
 
   /** Returns current buffered documents */
   readBufferedDocuments(number?: number): TSchema[] {
-    return this[kDocuments].splice(0, number ?? this[kDocuments].length);
+    const bufferedDocs: TSchema[] = [];
+    const documentsToRead = Math.min(number ?? this[kDocuments].length, this[kDocuments].length);
+
+    for (let count = 0; count < documentsToRead; count++) {
+      const document = this[kDocuments].shift();
+      if (document != null) {
+        bufferedDocs.push(document);
+      }
+    }
+
+    return bufferedDocs;
   }
 
   [Symbol.asyncIterator](): AsyncIterator<TSchema, void> {
@@ -350,7 +361,7 @@ export abstract class AbstractCursor<
         return false;
       }
 
-      if (this[kDocuments].length) {
+      if (this[kDocuments].length !== 0) {
         return true;
       }
 
@@ -597,8 +608,8 @@ export abstract class AbstractCursor<
       return;
     }
 
-    this[kId] = undefined;
-    this[kDocuments] = [];
+    this[kId] = null;
+    this[kDocuments].clear();
     this[kClosed] = false;
     this[kKilled] = false;
     this[kInitialized] = false;
@@ -662,7 +673,7 @@ export abstract class AbstractCursor<
             this[kNamespace] = ns(response.cursor.ns);
           }
 
-          this[kDocuments] = response.cursor.firstBatch;
+          this[kDocuments].pushMany(response.cursor.firstBatch);
         }
 
         // When server responses return without a cursor document, we close this cursor
@@ -671,7 +682,7 @@ export abstract class AbstractCursor<
         if (this[kId] == null) {
           this[kId] = Long.ZERO;
           // TODO(NODE-3286): ExecutionResult needs to accept a generic parameter
-          this[kDocuments] = [state.response as TODO_NODE_3286];
+          this[kDocuments].push(state.response as TODO_NODE_3286);
         }
       }
 
@@ -687,22 +698,14 @@ export abstract class AbstractCursor<
   }
 }
 
-function nextDocument<T>(cursor: AbstractCursor): T | null {
-  if (cursor[kDocuments] == null || !cursor[kDocuments].length) {
-    return null;
-  }
-
+function nextDocument<T>(cursor: AbstractCursor<T>): T | null {
   const doc = cursor[kDocuments].shift();
-  if (doc) {
-    const transform = cursor[kTransform];
-    if (transform) {
-      return transform(doc) as T;
-    }
 
-    return doc;
+  if (doc && cursor[kTransform]) {
+    return cursor[kTransform](doc) as T;
   }
 
-  return null;
+  return doc;
 }
 
 const nextAsync = promisify(
@@ -733,7 +736,7 @@ export function next<T>(
     return callback(undefined, null);
   }
 
-  if (cursor[kDocuments] && cursor[kDocuments].length) {
+  if (cursor[kDocuments].length !== 0) {
     callback(undefined, nextDocument<T>(cursor));
     return;
   }
@@ -757,19 +760,19 @@ export function next<T>(
 
   // otherwise need to call getMore
   const batchSize = cursor[kOptions].batchSize || 1000;
-  cursor._getMore(batchSize, (err, response) => {
+  cursor._getMore(batchSize, (error, response) => {
     if (response) {
       const cursorId =
         typeof response.cursor.id === 'number'
           ? Long.fromNumber(response.cursor.id)
           : response.cursor.id;
 
-      cursor[kDocuments] = response.cursor.nextBatch;
+      cursor[kDocuments].pushMany(response.cursor.nextBatch);
       cursor[kId] = cursorId;
     }
 
-    if (err || cursorIsDead(cursor)) {
-      return cleanupCursor(cursor, { error: err }, () => callback(err, nextDocument<T>(cursor)));
+    if (error || cursorIsDead(cursor)) {
+      return cleanupCursor(cursor, { error }, () => callback(error, nextDocument<T>(cursor)));
     }
 
     if (cursor[kDocuments].length === 0 && blocking === false) {

src/cursor/abstract_cursor.ts

@@ -477,6 +477,7 @@ export type {
   ClientMetadataOptions,
   EventEmitterWithState,
   HostAddress,
+  List,
   MongoDBNamespace
 } from './utils';
 export type { W, WriteConcernOptions, WriteConcernSettings } from './write_concern';