-
Notifications
You must be signed in to change notification settings - Fork 435
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fix: time out DHT network requests separately from query (#2524)
Apply a per-request timeout to each network request in a DHT query. To avoid having a "one size fits all" timeout, it is adaptive so will increase/decrease based on the average success/failure times during the previous (configurable, default 5s) time interval.
- Loading branch information
1 parent
d9366f9
commit bfa7660
Showing
8 changed files
with
395 additions
and
11 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| import { setMaxListeners } from '@libp2p/interface' | ||
| import { anySignal, type ClearableSignal } from 'any-signal' | ||
| import { MovingAverage } from './moving-average.js' | ||
| import type { MetricGroup, Metrics } from '@libp2p/interface' | ||
|
|
||
| export const DEFAULT_TIMEOUT_MULTIPLIER = 1.2 | ||
| export const DEFAULT_FAILURE_MULTIPLIER = 2 | ||
| export const DEFAULT_MIN_TIMEOUT = 2000 | ||
|
|
||
| export interface AdaptiveTimeoutSignal extends ClearableSignal { | ||
| start: number | ||
| timeout: number | ||
| } | ||
|
|
||
| export interface AdaptiveTimeoutInit { | ||
| metricName?: string | ||
| metrics?: Metrics | ||
| interval?: number | ||
| initialValue?: number | ||
| timeoutMultiplier?: number | ||
| failureMultiplier?: number | ||
| minTimeout?: number | ||
| } | ||
|
|
||
| export interface GetTimeoutSignalOptions { | ||
| timeoutFactor?: number | ||
| signal?: AbortSignal | ||
| } | ||
|
|
||
| export class AdaptiveTimeout { | ||
| private readonly success: MovingAverage | ||
| private readonly failure: MovingAverage | ||
| private readonly next: MovingAverage | ||
| private readonly metric?: MetricGroup | ||
| private readonly timeoutMultiplier: number | ||
| private readonly failureMultiplier: number | ||
| private readonly minTimeout: number | ||
|
|
||
| constructor (init: AdaptiveTimeoutInit = {}) { | ||
| this.success = new MovingAverage(init.interval ?? 5000) | ||
| this.failure = new MovingAverage(init.interval ?? 5000) | ||
| this.next = new MovingAverage(init.interval ?? 5000) | ||
| this.failureMultiplier = init.failureMultiplier ?? DEFAULT_FAILURE_MULTIPLIER | ||
| this.timeoutMultiplier = init.timeoutMultiplier ?? DEFAULT_TIMEOUT_MULTIPLIER | ||
| this.minTimeout = init.minTimeout ?? DEFAULT_MIN_TIMEOUT | ||
|
|
||
| if (init.metricName != null) { | ||
| this.metric = init.metrics?.registerMetricGroup(init.metricName) | ||
| } | ||
| } | ||
|
|
||
| getTimeoutSignal (options: GetTimeoutSignalOptions = {}): AdaptiveTimeoutSignal { | ||
| // calculate timeout for individual peers based on moving average of | ||
| // previous successful requests | ||
| const timeout = Math.max( | ||
| Math.round(this.next.movingAverage * (options.timeoutFactor ?? this.timeoutMultiplier)), | ||
| this.minTimeout | ||
| ) | ||
| const sendTimeout = AbortSignal.timeout(timeout) | ||
| const timeoutSignal = anySignal([options.signal, sendTimeout]) as AdaptiveTimeoutSignal | ||
| setMaxListeners(Infinity, timeoutSignal, sendTimeout) | ||
|
|
||
| timeoutSignal.start = Date.now() | ||
| timeoutSignal.timeout = timeout | ||
|
|
||
| return timeoutSignal | ||
| } | ||
|
|
||
| cleanUp (signal: AdaptiveTimeoutSignal): void { | ||
| const time = Date.now() - signal.start | ||
|
|
||
| if (signal.aborted) { | ||
| this.failure.push(time) | ||
| this.next.push(time * this.failureMultiplier) | ||
| this.metric?.update({ | ||
| failureMovingAverage: this.failure.movingAverage, | ||
| failureDeviation: this.failure.deviation, | ||
| failureForecast: this.failure.forecast, | ||
| failureVariance: this.failure.variance, | ||
| failure: time | ||
| }) | ||
| } else { | ||
| this.success.push(time) | ||
| this.next.push(time) | ||
| this.metric?.update({ | ||
| successMovingAverage: this.success.movingAverage, | ||
| successDeviation: this.success.deviation, | ||
| successForecast: this.success.forecast, | ||
| successVariance: this.success.variance, | ||
| success: time | ||
| }) | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| /** | ||
| * Implements exponential moving average. Ported from `moving-average`. | ||
| * | ||
| * @see https://en.wikipedia.org/wiki/Moving_average#Exponential_moving_average | ||
| * @see https://www.npmjs.com/package/moving-average | ||
| */ | ||
| export class MovingAverage { | ||
| public movingAverage: number | ||
| public variance: number | ||
| public deviation: number | ||
| public forecast: number | ||
| private readonly timespan: number | ||
| private previousTime?: number | ||
|
|
||
| constructor (timespan: number) { | ||
| this.timespan = timespan | ||
| this.movingAverage = 0 | ||
| this.variance = 0 | ||
| this.deviation = 0 | ||
| this.forecast = 0 | ||
| } | ||
|
|
||
| alpha (t: number, pt: number): number { | ||
| return 1 - (Math.exp(-(t - pt) / this.timespan)) | ||
| } | ||
|
|
||
| push (value: number, time: number = Date.now()): void { | ||
| if (this.previousTime != null) { | ||
| // calculate moving average | ||
| const a = this.alpha(time, this.previousTime) | ||
| const diff = value - this.movingAverage | ||
| const incr = a * diff | ||
| this.movingAverage = a * value + (1 - a) * this.movingAverage | ||
| // calculate variance & deviation | ||
| this.variance = (1 - a) * (this.variance + diff * incr) | ||
| this.deviation = Math.sqrt(this.variance) | ||
| // calculate forecast | ||
| this.forecast = this.movingAverage + a * diff | ||
| } else { | ||
| this.movingAverage = value | ||
| } | ||
|
|
||
| this.previousTime = time | ||
| } | ||
| } |
Oops, something went wrong.