mirror of
https://github.com/chanzuckerberg/cellxgene.git
synced 2026-10-06 17:58:11 +08:00
type promise limit
This commit is contained in:
@@ -28,37 +28,55 @@ Priority is a numeric value. Lower first. Stable ordering.
|
|||||||
|
|
||||||
import TinyQueue from "tinyqueue";
|
import TinyQueue from "tinyqueue";
|
||||||
|
|
||||||
// @ts-expect-error ts-migrate(7006) FIXME: Parameter 'a' implicitly has an 'any' type.
|
function compare<T>(
|
||||||
function compare(a, b) {
|
a: PromiseLimitQueueItem<T>,
|
||||||
|
b: PromiseLimitQueueItem<T>
|
||||||
|
): number {
|
||||||
const diff = a.priority - b.priority;
|
const diff = a.priority - b.priority;
|
||||||
if (diff) return diff;
|
if (diff) return diff;
|
||||||
return a.order - b.order;
|
return a.order - b.order;
|
||||||
}
|
}
|
||||||
|
|
||||||
export default class PromiseLimit {
|
interface PromiseLimitQueueItem<T> {
|
||||||
|
priority: number;
|
||||||
|
order: number;
|
||||||
|
resolve: (value: T | PromiseLike<T>) => void;
|
||||||
|
reject: (reason?: unknown) => void;
|
||||||
|
fn: (...args: Array<unknown>) => Promise<T>;
|
||||||
|
args: Array<unknown>;
|
||||||
|
}
|
||||||
|
|
||||||
|
export default class PromiseLimit<T> {
|
||||||
|
queue: TinyQueue<PromiseLimitQueueItem<T>>;
|
||||||
|
|
||||||
|
maxConcurrency: number;
|
||||||
|
|
||||||
|
pending: number;
|
||||||
|
|
||||||
|
insertCounter: number;
|
||||||
|
|
||||||
constructor(maxConcurrency = 5) {
|
constructor(maxConcurrency = 5) {
|
||||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any --- FIXME: disabled temporarily on migrate to TS.
|
this.queue = new TinyQueue<PromiseLimitQueueItem<T>>(
|
||||||
(this as any).queue = new TinyQueue([], compare);
|
new Array<PromiseLimitQueueItem<T>>(),
|
||||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any --- FIXME: disabled temporarily on migrate to TS.
|
compare
|
||||||
(this as any).maxConcurrency = maxConcurrency;
|
);
|
||||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any --- FIXME: disabled temporarily on migrate to TS.
|
this.maxConcurrency = maxConcurrency;
|
||||||
(this as any).pending = 0;
|
this.pending = 0;
|
||||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any --- FIXME: disabled temporarily on migrate to TS.
|
this.insertCounter = 0;
|
||||||
(this as any).insertCounter = 0;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// @ts-expect-error ts-migrate(7006) FIXME: Parameter 'p' implicitly has an 'any' type.
|
priorityAdd(
|
||||||
// eslint-disable-next-line @typescript-eslint/explicit-module-boundary-types --- FIXME: disabled temporarily on migrate to TS.
|
p: number,
|
||||||
priorityAdd(p, fn, ...args) {
|
fn: () => Promise<T>,
|
||||||
|
...args: Array<unknown>
|
||||||
|
): Promise<T> {
|
||||||
// p - numermic priority (lower first)
|
// p - numermic priority (lower first)
|
||||||
// fn - must return a promise
|
// fn - must return a promise
|
||||||
// args - will be passed to fn
|
// args - will be passed to fn
|
||||||
return this._push(p, fn, args);
|
return this._push(p, fn, args);
|
||||||
}
|
}
|
||||||
|
|
||||||
// @ts-expect-error ts-migrate(7006) FIXME: Parameter 'fn' implicitly has an 'any' type.
|
add(fn: () => Promise<T>, ...args: Array<unknown>): Promise<T> {
|
||||||
// eslint-disable-next-line @typescript-eslint/explicit-module-boundary-types --- FIXME: disabled temporarily on migrate to TS.
|
|
||||||
add(fn, ...args) {
|
|
||||||
// fn - must return a promise
|
// fn - must return a promise
|
||||||
// args - will be passed to fn
|
// args - will be passed to fn
|
||||||
return this._push(0, fn, args);
|
return this._push(0, fn, args);
|
||||||
@@ -68,35 +86,25 @@ export default class PromiseLimit {
|
|||||||
Private below
|
Private below
|
||||||
**/
|
**/
|
||||||
|
|
||||||
// @ts-expect-error ts-migrate(7006) FIXME: Parameter 'priority' implicitly has an 'any' type.
|
_push(
|
||||||
// eslint-disable-next-line @typescript-eslint/explicit-module-boundary-types --- FIXME: disabled temporarily on migrate to TS.
|
priority: number,
|
||||||
_push(priority, fn, args) {
|
fn: () => Promise<T>,
|
||||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any --- FIXME: disabled temporarily on migrate to TS.
|
args: Array<unknown>
|
||||||
const order = (this as any).insertCount;
|
): Promise<T> {
|
||||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any --- FIXME: disabled temporarily on migrate to TS.
|
const order = this.insertCounter;
|
||||||
(this as any).insertCount += 1;
|
this.insertCounter += 1;
|
||||||
return new Promise((resolve, reject) => {
|
return new Promise((resolve, reject) => {
|
||||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any --- FIXME: disabled temporarily on migrate to TS.
|
this.queue.push({ priority, order, fn, args, resolve, reject });
|
||||||
(this as any).queue.push({ priority, order, fn, args, resolve, reject });
|
|
||||||
this._resolveNext(false);
|
this._resolveNext(false);
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
// eslint-disable-next-line @typescript-eslint/explicit-module-boundary-types --- FIXME: disabled temporarily on migrate to TS.
|
_resolveNext = (completed = true): void => {
|
||||||
_resolveNext = (completed = true) => {
|
if (completed) this.pending -= 1;
|
||||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any --- FIXME: disabled temporarily on migrate to TS.
|
|
||||||
if (completed) (this as any).pending -= 1;
|
|
||||||
|
|
||||||
while (
|
while (this.queue.length > 0 && this.pending < this.maxConcurrency) {
|
||||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any --- FIXME: disabled temporarily on migrate to TS.
|
const task = this.queue.pop() as PromiseLimitQueueItem<T>; // order of insertion
|
||||||
(this as any).queue.length > 0 &&
|
this.pending += 1;
|
||||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any --- FIXME: disabled temporarily on migrate to TS.
|
|
||||||
(this as any).pending < (this as any).maxConcurrency
|
|
||||||
) {
|
|
||||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any --- FIXME: disabled temporarily on migrate to TS.
|
|
||||||
const task = (this as any).queue.pop(); // order of insertion
|
|
||||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any --- FIXME: disabled temporarily on migrate to TS.
|
|
||||||
(this as any).pending += 1;
|
|
||||||
const { resolve, reject, fn, args } = task;
|
const { resolve, reject, fn, args } = task;
|
||||||
|
|
||||||
try {
|
try {
|
||||||
|
|||||||
Reference in New Issue
Block a user