Estructura inicial del proyecto
This commit is contained in:
771
backend/node_modules/mongodb/src/sdam/monitor.ts
generated
vendored
Normal file
771
backend/node_modules/mongodb/src/sdam/monitor.ts
generated
vendored
Normal file
@ -0,0 +1,771 @@
|
||||
import { clearTimeout, setTimeout } from 'timers';
|
||||
|
||||
import { type Document, Long } from '../bson';
|
||||
import { connect, makeConnection, makeSocket, performInitialHandshake } from '../cmap/connect';
|
||||
import type { Connection, ConnectionOptions } from '../cmap/connection';
|
||||
import { getFAASEnv } from '../cmap/handshake/client_metadata';
|
||||
import { LEGACY_HELLO_COMMAND } from '../constants';
|
||||
import { MongoError, MongoErrorLabel, MongoNetworkTimeoutError } from '../error';
|
||||
import { MongoLoggableComponent } from '../mongo_logger';
|
||||
import { CancellationToken, TypedEventEmitter } from '../mongo_types';
|
||||
import {
|
||||
calculateDurationInMs,
|
||||
type Callback,
|
||||
type EventEmitterWithState,
|
||||
makeStateMachine,
|
||||
noop,
|
||||
ns,
|
||||
processTimeMS
|
||||
} from '../utils';
|
||||
import { ServerType, STATE_CLOSED, STATE_CLOSING } from './common';
|
||||
import {
|
||||
ServerHeartbeatFailedEvent,
|
||||
ServerHeartbeatStartedEvent,
|
||||
ServerHeartbeatSucceededEvent
|
||||
} from './events';
|
||||
import { Server } from './server';
|
||||
import type { TopologyVersion } from './server_description';
|
||||
|
||||
const STATE_IDLE = 'idle';
|
||||
const STATE_MONITORING = 'monitoring';
|
||||
const stateTransition = makeStateMachine({
|
||||
[STATE_CLOSING]: [STATE_CLOSING, STATE_IDLE, STATE_CLOSED],
|
||||
[STATE_CLOSED]: [STATE_CLOSED, STATE_MONITORING],
|
||||
[STATE_IDLE]: [STATE_IDLE, STATE_MONITORING, STATE_CLOSING],
|
||||
[STATE_MONITORING]: [STATE_MONITORING, STATE_IDLE, STATE_CLOSING]
|
||||
});
|
||||
|
||||
const INVALID_REQUEST_CHECK_STATES = new Set([STATE_CLOSING, STATE_CLOSED, STATE_MONITORING]);
|
||||
function isInCloseState(monitor: Monitor) {
|
||||
return monitor.s.state === STATE_CLOSED || monitor.s.state === STATE_CLOSING;
|
||||
}
|
||||
|
||||
/** @public */
|
||||
export const ServerMonitoringMode = Object.freeze({
|
||||
auto: 'auto',
|
||||
poll: 'poll',
|
||||
stream: 'stream'
|
||||
} as const);
|
||||
|
||||
/** @public */
|
||||
export type ServerMonitoringMode = (typeof ServerMonitoringMode)[keyof typeof ServerMonitoringMode];
|
||||
|
||||
/** @internal */
|
||||
export interface MonitorPrivate {
|
||||
state: string;
|
||||
}
|
||||
|
||||
/** @public */
|
||||
export interface MonitorOptions
|
||||
extends Omit<ConnectionOptions, 'id' | 'generation' | 'hostAddress'> {
|
||||
connectTimeoutMS: number;
|
||||
heartbeatFrequencyMS: number;
|
||||
minHeartbeatFrequencyMS: number;
|
||||
serverMonitoringMode: ServerMonitoringMode;
|
||||
}
|
||||
|
||||
/** @public */
|
||||
export type MonitorEvents = {
|
||||
serverHeartbeatStarted(event: ServerHeartbeatStartedEvent): void;
|
||||
serverHeartbeatSucceeded(event: ServerHeartbeatSucceededEvent): void;
|
||||
serverHeartbeatFailed(event: ServerHeartbeatFailedEvent): void;
|
||||
resetServer(error?: MongoError): void;
|
||||
resetConnectionPool(): void;
|
||||
close(): void;
|
||||
} & EventEmitterWithState;
|
||||
|
||||
/** @internal */
|
||||
export class Monitor extends TypedEventEmitter<MonitorEvents> {
|
||||
/** @internal */
|
||||
s: MonitorPrivate;
|
||||
address: string;
|
||||
options: Readonly<
|
||||
Pick<
|
||||
MonitorOptions,
|
||||
| 'connectTimeoutMS'
|
||||
| 'heartbeatFrequencyMS'
|
||||
| 'minHeartbeatFrequencyMS'
|
||||
| 'serverMonitoringMode'
|
||||
>
|
||||
>;
|
||||
connectOptions: ConnectionOptions;
|
||||
isRunningInFaasEnv: boolean;
|
||||
server: Server;
|
||||
connection: Connection | null;
|
||||
cancellationToken: CancellationToken;
|
||||
/** @internal */
|
||||
monitorId?: MonitorInterval;
|
||||
rttPinger?: RTTPinger;
|
||||
/** @internal */
|
||||
override component = MongoLoggableComponent.TOPOLOGY;
|
||||
/** @internal */
|
||||
private rttSampler: RTTSampler;
|
||||
|
||||
constructor(server: Server, options: MonitorOptions) {
|
||||
super();
|
||||
this.on('error', noop);
|
||||
|
||||
this.server = server;
|
||||
this.connection = null;
|
||||
this.cancellationToken = new CancellationToken();
|
||||
this.cancellationToken.setMaxListeners(Infinity);
|
||||
this.monitorId = undefined;
|
||||
this.s = {
|
||||
state: STATE_CLOSED
|
||||
};
|
||||
this.address = server.description.address;
|
||||
this.options = Object.freeze({
|
||||
connectTimeoutMS: options.connectTimeoutMS ?? 10000,
|
||||
heartbeatFrequencyMS: options.heartbeatFrequencyMS ?? 10000,
|
||||
minHeartbeatFrequencyMS: options.minHeartbeatFrequencyMS ?? 500,
|
||||
serverMonitoringMode: options.serverMonitoringMode
|
||||
});
|
||||
this.isRunningInFaasEnv = getFAASEnv() != null;
|
||||
this.mongoLogger = this.server.topology.client?.mongoLogger;
|
||||
this.rttSampler = new RTTSampler(10);
|
||||
|
||||
const cancellationToken = this.cancellationToken;
|
||||
// TODO: refactor this to pull it directly from the pool, requires new ConnectionPool integration
|
||||
const connectOptions = {
|
||||
id: '<monitor>' as const,
|
||||
generation: server.pool.generation,
|
||||
cancellationToken,
|
||||
hostAddress: server.description.hostAddress,
|
||||
...options,
|
||||
// force BSON serialization options
|
||||
raw: false,
|
||||
useBigInt64: false,
|
||||
promoteLongs: true,
|
||||
promoteValues: true,
|
||||
promoteBuffers: true
|
||||
};
|
||||
|
||||
// ensure no authentication is used for monitoring
|
||||
delete connectOptions.credentials;
|
||||
if (connectOptions.autoEncrypter) {
|
||||
delete connectOptions.autoEncrypter;
|
||||
}
|
||||
|
||||
this.connectOptions = Object.freeze(connectOptions);
|
||||
}
|
||||
|
||||
connect(): void {
|
||||
if (this.s.state !== STATE_CLOSED) {
|
||||
return;
|
||||
}
|
||||
|
||||
// start
|
||||
const heartbeatFrequencyMS = this.options.heartbeatFrequencyMS;
|
||||
const minHeartbeatFrequencyMS = this.options.minHeartbeatFrequencyMS;
|
||||
this.monitorId = new MonitorInterval(monitorServer(this), {
|
||||
heartbeatFrequencyMS: heartbeatFrequencyMS,
|
||||
minHeartbeatFrequencyMS: minHeartbeatFrequencyMS,
|
||||
immediate: true
|
||||
});
|
||||
}
|
||||
|
||||
requestCheck(): void {
|
||||
if (INVALID_REQUEST_CHECK_STATES.has(this.s.state)) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.monitorId?.wake();
|
||||
}
|
||||
|
||||
reset(): void {
|
||||
const topologyVersion = this.server.description.topologyVersion;
|
||||
if (isInCloseState(this) || topologyVersion == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
stateTransition(this, STATE_CLOSING);
|
||||
resetMonitorState(this);
|
||||
|
||||
// restart monitor
|
||||
stateTransition(this, STATE_IDLE);
|
||||
|
||||
// restart monitoring
|
||||
const heartbeatFrequencyMS = this.options.heartbeatFrequencyMS;
|
||||
const minHeartbeatFrequencyMS = this.options.minHeartbeatFrequencyMS;
|
||||
this.monitorId = new MonitorInterval(monitorServer(this), {
|
||||
heartbeatFrequencyMS: heartbeatFrequencyMS,
|
||||
minHeartbeatFrequencyMS: minHeartbeatFrequencyMS
|
||||
});
|
||||
}
|
||||
|
||||
close(): void {
|
||||
if (isInCloseState(this)) {
|
||||
return;
|
||||
}
|
||||
|
||||
stateTransition(this, STATE_CLOSING);
|
||||
resetMonitorState(this);
|
||||
|
||||
// close monitor
|
||||
this.emit('close');
|
||||
stateTransition(this, STATE_CLOSED);
|
||||
}
|
||||
|
||||
get roundTripTime(): number {
|
||||
return this.rttSampler.average();
|
||||
}
|
||||
|
||||
get minRoundTripTime(): number {
|
||||
return this.rttSampler.min();
|
||||
}
|
||||
|
||||
get latestRtt(): number | null {
|
||||
return this.rttSampler.last;
|
||||
}
|
||||
|
||||
addRttSample(rtt: number) {
|
||||
this.rttSampler.addSample(rtt);
|
||||
}
|
||||
|
||||
clearRttSamples() {
|
||||
this.rttSampler.clear();
|
||||
}
|
||||
}
|
||||
|
||||
function resetMonitorState(monitor: Monitor) {
|
||||
monitor.monitorId?.stop();
|
||||
monitor.monitorId = undefined;
|
||||
|
||||
monitor.rttPinger?.close();
|
||||
monitor.rttPinger = undefined;
|
||||
|
||||
monitor.cancellationToken.emit('cancel');
|
||||
|
||||
monitor.connection?.destroy();
|
||||
monitor.connection = null;
|
||||
|
||||
monitor.clearRttSamples();
|
||||
}
|
||||
|
||||
function useStreamingProtocol(monitor: Monitor, topologyVersion: TopologyVersion | null): boolean {
|
||||
// If we have no topology version we always poll no matter
|
||||
// what the user provided, since the server does not support
|
||||
// the streaming protocol.
|
||||
if (topologyVersion == null) return false;
|
||||
|
||||
const serverMonitoringMode = monitor.options.serverMonitoringMode;
|
||||
if (serverMonitoringMode === ServerMonitoringMode.poll) return false;
|
||||
if (serverMonitoringMode === ServerMonitoringMode.stream) return true;
|
||||
|
||||
// If we are in auto mode, we need to figure out if we're in a FaaS
|
||||
// environment or not and choose the appropriate mode.
|
||||
if (monitor.isRunningInFaasEnv) return false;
|
||||
return true;
|
||||
}
|
||||
|
||||
function checkServer(monitor: Monitor, callback: Callback<Document | null>) {
|
||||
let start: number;
|
||||
let awaited: boolean;
|
||||
const topologyVersion = monitor.server.description.topologyVersion;
|
||||
const isAwaitable = useStreamingProtocol(monitor, topologyVersion);
|
||||
monitor.emitAndLogHeartbeat(
|
||||
Server.SERVER_HEARTBEAT_STARTED,
|
||||
monitor.server.topology.s.id,
|
||||
undefined,
|
||||
new ServerHeartbeatStartedEvent(monitor.address, isAwaitable)
|
||||
);
|
||||
|
||||
function onHeartbeatFailed(err: Error) {
|
||||
monitor.connection?.destroy();
|
||||
monitor.connection = null;
|
||||
monitor.emitAndLogHeartbeat(
|
||||
Server.SERVER_HEARTBEAT_FAILED,
|
||||
monitor.server.topology.s.id,
|
||||
undefined,
|
||||
new ServerHeartbeatFailedEvent(monitor.address, calculateDurationInMs(start), err, awaited)
|
||||
);
|
||||
|
||||
const error = !(err instanceof MongoError)
|
||||
? new MongoError(MongoError.buildErrorMessage(err), { cause: err })
|
||||
: err;
|
||||
error.addErrorLabel(MongoErrorLabel.ResetPool);
|
||||
if (error instanceof MongoNetworkTimeoutError) {
|
||||
error.addErrorLabel(MongoErrorLabel.InterruptInUseConnections);
|
||||
}
|
||||
|
||||
monitor.emit('resetServer', error);
|
||||
callback(err);
|
||||
}
|
||||
|
||||
function onHeartbeatSucceeded(hello: Document) {
|
||||
if (!('isWritablePrimary' in hello)) {
|
||||
// Provide hello-style response document.
|
||||
hello.isWritablePrimary = hello[LEGACY_HELLO_COMMAND];
|
||||
}
|
||||
|
||||
// NOTE: here we use the latestRtt as this measurement corresponds with the value
|
||||
// obtained for this successful heartbeat, if there is no latestRtt, then we calculate the
|
||||
// duration
|
||||
const duration =
|
||||
isAwaitable && monitor.rttPinger
|
||||
? (monitor.rttPinger.latestRtt ?? calculateDurationInMs(start))
|
||||
: calculateDurationInMs(start);
|
||||
|
||||
monitor.addRttSample(duration);
|
||||
|
||||
monitor.emitAndLogHeartbeat(
|
||||
Server.SERVER_HEARTBEAT_SUCCEEDED,
|
||||
monitor.server.topology.s.id,
|
||||
hello.connectionId,
|
||||
new ServerHeartbeatSucceededEvent(monitor.address, duration, hello, isAwaitable)
|
||||
);
|
||||
|
||||
if (isAwaitable) {
|
||||
// If we are using the streaming protocol then we immediately issue another 'started'
|
||||
// event, otherwise the "check" is complete and return to the main monitor loop
|
||||
monitor.emitAndLogHeartbeat(
|
||||
Server.SERVER_HEARTBEAT_STARTED,
|
||||
monitor.server.topology.s.id,
|
||||
undefined,
|
||||
new ServerHeartbeatStartedEvent(monitor.address, true)
|
||||
);
|
||||
// We have not actually sent an outgoing handshake, but when we get the next response we
|
||||
// want the duration to reflect the time since we last heard from the server
|
||||
start = processTimeMS();
|
||||
} else {
|
||||
monitor.rttPinger?.close();
|
||||
monitor.rttPinger = undefined;
|
||||
|
||||
callback(undefined, hello);
|
||||
}
|
||||
}
|
||||
|
||||
const { connection } = monitor;
|
||||
if (connection && !connection.closed) {
|
||||
const { serverApi, helloOk } = connection;
|
||||
const connectTimeoutMS = monitor.options.connectTimeoutMS;
|
||||
const maxAwaitTimeMS = monitor.options.heartbeatFrequencyMS;
|
||||
|
||||
const cmd = {
|
||||
[serverApi?.version || helloOk ? 'hello' : LEGACY_HELLO_COMMAND]: 1,
|
||||
...(isAwaitable && topologyVersion
|
||||
? { maxAwaitTimeMS, topologyVersion: makeTopologyVersion(topologyVersion) }
|
||||
: {})
|
||||
};
|
||||
|
||||
const options = isAwaitable
|
||||
? {
|
||||
socketTimeoutMS: connectTimeoutMS ? connectTimeoutMS + maxAwaitTimeMS : 0,
|
||||
exhaustAllowed: true
|
||||
}
|
||||
: { socketTimeoutMS: connectTimeoutMS };
|
||||
|
||||
if (isAwaitable && monitor.rttPinger == null) {
|
||||
monitor.rttPinger = new RTTPinger(monitor);
|
||||
}
|
||||
|
||||
// Record new start time before sending handshake
|
||||
start = processTimeMS();
|
||||
|
||||
if (isAwaitable) {
|
||||
awaited = true;
|
||||
return connection.exhaustCommand(ns('admin.$cmd'), cmd, options, (error, hello) => {
|
||||
if (error) return onHeartbeatFailed(error);
|
||||
return onHeartbeatSucceeded(hello);
|
||||
});
|
||||
}
|
||||
|
||||
awaited = false;
|
||||
connection
|
||||
.command(ns('admin.$cmd'), cmd, options)
|
||||
.then(onHeartbeatSucceeded, onHeartbeatFailed);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
// connecting does an implicit `hello`
|
||||
(async () => {
|
||||
const socket = await makeSocket(monitor.connectOptions);
|
||||
const connection = makeConnection(monitor.connectOptions, socket);
|
||||
// The start time is after socket creation but before the handshake
|
||||
start = processTimeMS();
|
||||
try {
|
||||
await performInitialHandshake(connection, monitor.connectOptions);
|
||||
return connection;
|
||||
} catch (error) {
|
||||
connection.destroy();
|
||||
throw error;
|
||||
}
|
||||
})().then(
|
||||
connection => {
|
||||
if (isInCloseState(monitor)) {
|
||||
connection.destroy();
|
||||
return;
|
||||
}
|
||||
const duration = calculateDurationInMs(start);
|
||||
monitor.addRttSample(duration);
|
||||
|
||||
monitor.connection = connection;
|
||||
monitor.emitAndLogHeartbeat(
|
||||
Server.SERVER_HEARTBEAT_SUCCEEDED,
|
||||
monitor.server.topology.s.id,
|
||||
connection.hello?.connectionId,
|
||||
new ServerHeartbeatSucceededEvent(
|
||||
monitor.address,
|
||||
duration,
|
||||
connection.hello,
|
||||
useStreamingProtocol(monitor, connection.hello?.topologyVersion)
|
||||
)
|
||||
);
|
||||
|
||||
callback(undefined, connection.hello);
|
||||
},
|
||||
error => {
|
||||
monitor.connection = null;
|
||||
awaited = false;
|
||||
onHeartbeatFailed(error);
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
function monitorServer(monitor: Monitor) {
|
||||
return (callback: Callback) => {
|
||||
if (monitor.s.state === STATE_MONITORING) {
|
||||
queueMicrotask(callback);
|
||||
return;
|
||||
}
|
||||
stateTransition(monitor, STATE_MONITORING);
|
||||
function done() {
|
||||
if (!isInCloseState(monitor)) {
|
||||
stateTransition(monitor, STATE_IDLE);
|
||||
}
|
||||
|
||||
callback();
|
||||
}
|
||||
|
||||
checkServer(monitor, (err, hello) => {
|
||||
if (err) {
|
||||
// otherwise an error occurred on initial discovery, also bail
|
||||
if (monitor.server.description.type === ServerType.Unknown) {
|
||||
return done();
|
||||
}
|
||||
}
|
||||
|
||||
// if the check indicates streaming is supported, immediately reschedule monitoring
|
||||
if (useStreamingProtocol(monitor, hello?.topologyVersion)) {
|
||||
setTimeout(() => {
|
||||
if (!isInCloseState(monitor)) {
|
||||
monitor.monitorId?.wake();
|
||||
}
|
||||
}, 0);
|
||||
}
|
||||
|
||||
done();
|
||||
});
|
||||
};
|
||||
}
|
||||
|
||||
function makeTopologyVersion(tv: TopologyVersion) {
|
||||
return {
|
||||
processId: tv.processId,
|
||||
// tests mock counter as just number, but in a real situation counter should always be a Long
|
||||
// TODO(NODE-2674): Preserve int64 sent from MongoDB
|
||||
counter: Long.isLong(tv.counter) ? tv.counter : Long.fromNumber(tv.counter)
|
||||
};
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
export interface RTTPingerOptions extends ConnectionOptions {
|
||||
heartbeatFrequencyMS: number;
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
export class RTTPinger {
|
||||
connection?: Connection;
|
||||
/** @internal */
|
||||
cancellationToken: CancellationToken;
|
||||
/** @internal */
|
||||
monitorId: NodeJS.Timeout;
|
||||
/** @internal */
|
||||
monitor: Monitor;
|
||||
closed: boolean;
|
||||
/** @internal */
|
||||
latestRtt?: number;
|
||||
|
||||
constructor(monitor: Monitor) {
|
||||
this.connection = undefined;
|
||||
this.cancellationToken = monitor.cancellationToken;
|
||||
this.closed = false;
|
||||
this.monitor = monitor;
|
||||
this.latestRtt = monitor.latestRtt ?? undefined;
|
||||
|
||||
const heartbeatFrequencyMS = monitor.options.heartbeatFrequencyMS;
|
||||
this.monitorId = setTimeout(() => this.measureRoundTripTime(), heartbeatFrequencyMS);
|
||||
}
|
||||
|
||||
get roundTripTime(): number {
|
||||
return this.monitor.roundTripTime;
|
||||
}
|
||||
|
||||
get minRoundTripTime(): number {
|
||||
return this.monitor.minRoundTripTime;
|
||||
}
|
||||
|
||||
close(): void {
|
||||
this.closed = true;
|
||||
clearTimeout(this.monitorId);
|
||||
|
||||
this.connection?.destroy();
|
||||
this.connection = undefined;
|
||||
}
|
||||
|
||||
private measureAndReschedule(start: number, conn?: Connection) {
|
||||
if (this.closed) {
|
||||
conn?.destroy();
|
||||
return;
|
||||
}
|
||||
|
||||
if (this.connection == null) {
|
||||
this.connection = conn;
|
||||
}
|
||||
|
||||
this.latestRtt = calculateDurationInMs(start);
|
||||
this.monitorId = setTimeout(
|
||||
() => this.measureRoundTripTime(),
|
||||
this.monitor.options.heartbeatFrequencyMS
|
||||
);
|
||||
}
|
||||
|
||||
private measureRoundTripTime() {
|
||||
const start = processTimeMS();
|
||||
|
||||
if (this.closed) {
|
||||
return;
|
||||
}
|
||||
|
||||
const connection = this.connection;
|
||||
if (connection == null) {
|
||||
connect(this.monitor.connectOptions).then(
|
||||
connection => {
|
||||
this.measureAndReschedule(start, connection);
|
||||
},
|
||||
() => {
|
||||
this.connection = undefined;
|
||||
}
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const commandName =
|
||||
connection.serverApi?.version || connection.helloOk ? 'hello' : LEGACY_HELLO_COMMAND;
|
||||
|
||||
connection.command(ns('admin.$cmd'), { [commandName]: 1 }, undefined).then(
|
||||
() => this.measureAndReschedule(start),
|
||||
() => {
|
||||
this.connection?.destroy();
|
||||
this.connection = undefined;
|
||||
return;
|
||||
}
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @internal
|
||||
*/
|
||||
export interface MonitorIntervalOptions {
|
||||
/** The interval to execute a method on */
|
||||
heartbeatFrequencyMS: number;
|
||||
/** A minimum interval that must elapse before the method is called */
|
||||
minHeartbeatFrequencyMS: number;
|
||||
/** Whether the method should be called immediately when the interval is started */
|
||||
immediate: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* @internal
|
||||
*/
|
||||
export class MonitorInterval {
|
||||
fn: (callback: Callback) => void;
|
||||
timerId: NodeJS.Timeout | undefined;
|
||||
lastExecutionEnded: number;
|
||||
isExpeditedCallToFnScheduled = false;
|
||||
stopped = false;
|
||||
isExecutionInProgress = false;
|
||||
hasExecutedOnce = false;
|
||||
|
||||
heartbeatFrequencyMS: number;
|
||||
minHeartbeatFrequencyMS: number;
|
||||
|
||||
constructor(fn: (callback: Callback) => void, options: Partial<MonitorIntervalOptions> = {}) {
|
||||
this.fn = fn;
|
||||
this.lastExecutionEnded = -Infinity;
|
||||
|
||||
this.heartbeatFrequencyMS = options.heartbeatFrequencyMS ?? 1000;
|
||||
this.minHeartbeatFrequencyMS = options.minHeartbeatFrequencyMS ?? 500;
|
||||
|
||||
if (options.immediate) {
|
||||
this._executeAndReschedule();
|
||||
} else {
|
||||
this._reschedule(undefined);
|
||||
}
|
||||
}
|
||||
|
||||
wake() {
|
||||
const currentTime = processTimeMS();
|
||||
const timeSinceLastCall = currentTime - this.lastExecutionEnded;
|
||||
|
||||
// TODO(NODE-4674): Add error handling and logging to the monitor
|
||||
if (timeSinceLastCall < 0) {
|
||||
return this._executeAndReschedule();
|
||||
}
|
||||
|
||||
if (this.isExecutionInProgress) {
|
||||
return;
|
||||
}
|
||||
|
||||
// debounce multiple calls to wake within the `minInterval`
|
||||
if (this.isExpeditedCallToFnScheduled) {
|
||||
return;
|
||||
}
|
||||
|
||||
// reschedule a call as soon as possible, ensuring the call never happens
|
||||
// faster than the `minInterval`
|
||||
if (timeSinceLastCall < this.minHeartbeatFrequencyMS) {
|
||||
this.isExpeditedCallToFnScheduled = true;
|
||||
this._reschedule(this.minHeartbeatFrequencyMS - timeSinceLastCall);
|
||||
return;
|
||||
}
|
||||
|
||||
this._executeAndReschedule();
|
||||
}
|
||||
|
||||
stop() {
|
||||
this.stopped = true;
|
||||
if (this.timerId) {
|
||||
clearTimeout(this.timerId);
|
||||
this.timerId = undefined;
|
||||
}
|
||||
|
||||
this.lastExecutionEnded = -Infinity;
|
||||
this.isExpeditedCallToFnScheduled = false;
|
||||
}
|
||||
|
||||
toString() {
|
||||
return JSON.stringify(this);
|
||||
}
|
||||
|
||||
toJSON() {
|
||||
const currentTime = processTimeMS();
|
||||
const timeSinceLastCall = currentTime - this.lastExecutionEnded;
|
||||
return {
|
||||
timerId: this.timerId != null ? 'set' : 'cleared',
|
||||
lastCallTime: this.lastExecutionEnded,
|
||||
isExpeditedCheckScheduled: this.isExpeditedCallToFnScheduled,
|
||||
stopped: this.stopped,
|
||||
heartbeatFrequencyMS: this.heartbeatFrequencyMS,
|
||||
minHeartbeatFrequencyMS: this.minHeartbeatFrequencyMS,
|
||||
currentTime,
|
||||
timeSinceLastCall
|
||||
};
|
||||
}
|
||||
|
||||
private _reschedule(ms?: number) {
|
||||
if (this.stopped) return;
|
||||
if (this.timerId) {
|
||||
clearTimeout(this.timerId);
|
||||
}
|
||||
|
||||
this.timerId = setTimeout(this._executeAndReschedule, ms || this.heartbeatFrequencyMS);
|
||||
}
|
||||
|
||||
private _executeAndReschedule = () => {
|
||||
if (this.stopped) return;
|
||||
if (this.timerId) {
|
||||
clearTimeout(this.timerId);
|
||||
}
|
||||
|
||||
this.isExpeditedCallToFnScheduled = false;
|
||||
this.isExecutionInProgress = true;
|
||||
|
||||
this.fn(() => {
|
||||
this.lastExecutionEnded = processTimeMS();
|
||||
this.isExecutionInProgress = false;
|
||||
this._reschedule(this.heartbeatFrequencyMS);
|
||||
});
|
||||
};
|
||||
}
|
||||
|
||||
/** @internal
|
||||
* This class implements the RTT sampling logic specified for [CSOT](https://github.com/mongodb/specifications/blob/bbb335e60cd7ea1e0f7cd9a9443cb95fc9d3b64d/source/client-side-operations-timeout/client-side-operations-timeout.md#drivers-use-minimum-rtt-to-short-circuit-operations)
|
||||
*
|
||||
* This is implemented as a [circular buffer](https://en.wikipedia.org/wiki/Circular_buffer) keeping
|
||||
* the most recent `windowSize` samples
|
||||
* */
|
||||
export class RTTSampler {
|
||||
/** Index of the next slot to be overwritten */
|
||||
private writeIndex: number;
|
||||
private length: number;
|
||||
private rttSamples: Float64Array;
|
||||
|
||||
constructor(windowSize = 10) {
|
||||
this.rttSamples = new Float64Array(windowSize);
|
||||
this.length = 0;
|
||||
this.writeIndex = 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Adds an rtt sample to the end of the circular buffer
|
||||
* When `windowSize` samples have been collected, `addSample` overwrites the least recently added
|
||||
* sample
|
||||
*/
|
||||
addSample(sample: number) {
|
||||
this.rttSamples[this.writeIndex++] = sample;
|
||||
if (this.length < this.rttSamples.length) {
|
||||
this.length++;
|
||||
}
|
||||
|
||||
this.writeIndex %= this.rttSamples.length;
|
||||
}
|
||||
|
||||
/**
|
||||
* When \< 2 samples have been collected, returns 0
|
||||
* Otherwise computes the minimum value samples contained in the buffer
|
||||
*/
|
||||
min(): number {
|
||||
if (this.length < 2) return 0;
|
||||
let min = this.rttSamples[0];
|
||||
for (let i = 1; i < this.length; i++) {
|
||||
if (this.rttSamples[i] < min) min = this.rttSamples[i];
|
||||
}
|
||||
|
||||
return min;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns mean of samples contained in the buffer
|
||||
*/
|
||||
average(): number {
|
||||
if (this.length === 0) return 0;
|
||||
let sum = 0;
|
||||
for (let i = 0; i < this.length; i++) {
|
||||
sum += this.rttSamples[i];
|
||||
}
|
||||
|
||||
return sum / this.length;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns most recently inserted element in the buffer
|
||||
* Returns null if the buffer is empty
|
||||
* */
|
||||
get last(): number | null {
|
||||
if (this.length === 0) return null;
|
||||
return this.rttSamples[this.writeIndex === 0 ? this.length - 1 : this.writeIndex - 1];
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear the buffer
|
||||
* NOTE: this does not overwrite the data held in the internal array, just the pointers into
|
||||
* this array
|
||||
*/
|
||||
clear() {
|
||||
this.length = 0;
|
||||
this.writeIndex = 0;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user