From b91fe07309954fa6a18fa1934e3a5e9d9dffd233 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Wed, 27 Oct 2021 21:00:07 +0200 Subject: [PATCH] use a proper AbortController instead of the home baked one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- packages/backend-tasks/package.json | 1 + .../backend-tasks/src/tasks/CancelToken.ts | 49 ------------------- .../src/tasks/PluginTaskManagerJanitor.ts | 26 ++++------ .../backend-tasks/src/tasks/TaskWorker.ts | 29 ++++------- packages/backend-tasks/src/tasks/util.ts | 32 ++++++++++++ 5 files changed, 51 insertions(+), 86 deletions(-) delete mode 100644 packages/backend-tasks/src/tasks/CancelToken.ts diff --git a/packages/backend-tasks/package.json b/packages/backend-tasks/package.json index a5374d7b6e..dcb5922cf1 100644 --- a/packages/backend-tasks/package.json +++ b/packages/backend-tasks/package.json @@ -37,6 +37,7 @@ "knex": "^0.95.1", "lodash": "^4.17.21", "luxon": "^2.0.2", + "node-abort-controller": "^3.0.1", "uuid": "^8.0.0", "winston": "^3.2.1", "zod": "^3.9.5" diff --git a/packages/backend-tasks/src/tasks/CancelToken.ts b/packages/backend-tasks/src/tasks/CancelToken.ts deleted file mode 100644 index 190a5d6562..0000000000 --- a/packages/backend-tasks/src/tasks/CancelToken.ts +++ /dev/null @@ -1,49 +0,0 @@ -/* - * Copyright 2021 The Backstage Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -export class CancelToken { - #cancel: () => void; - #isCancelled: boolean; - #cancelPromise: Promise; - - static create(): CancelToken { - return new CancelToken(); - } - - private constructor() { - this.#isCancelled = false; - - this.#cancel = () => {}; // Avoids a TS warning - this.#cancelPromise = new Promise(resolve => { - this.#cancel = () => { - this.#isCancelled = true; - resolve(); - }; - }); - } - - cancel(): void { - this.#cancel(); - } - - get isCancelled(): boolean { - return this.#isCancelled; - } - - get promise(): Promise { - return this.#cancelPromise; - } -} diff --git a/packages/backend-tasks/src/tasks/PluginTaskManagerJanitor.ts b/packages/backend-tasks/src/tasks/PluginTaskManagerJanitor.ts index 2912041b3f..ec65eadf1e 100644 --- a/packages/backend-tasks/src/tasks/PluginTaskManagerJanitor.ts +++ b/packages/backend-tasks/src/tasks/PluginTaskManagerJanitor.ts @@ -16,9 +16,10 @@ import { Knex } from 'knex'; import { Duration } from 'luxon'; +import { AbortController, AbortSignal } from 'node-abort-controller'; import { Logger } from 'winston'; import { DbTasksRow, DB_TASKS_TABLE } from '../database/tables'; -import { CancelToken } from './CancelToken'; +import { sleep } from './util'; /** * Makes sure to auto-expire and clean up things that time out or for other @@ -28,7 +29,8 @@ export class PluginTaskManagerJanitor { private readonly knex: Knex; private readonly waitBetweenRuns: Duration; private readonly logger: Logger; - private readonly cancelToken: CancelToken; + private readonly abortController: AbortController; + private readonly abortSignal: AbortSignal; constructor(options: { knex: Knex; @@ -38,23 +40,24 @@ export class PluginTaskManagerJanitor { this.knex = options.knex; this.waitBetweenRuns = options.waitBetweenRuns; this.logger = options.logger; - this.cancelToken = CancelToken.create(); + this.abortController = new AbortController(); + this.abortSignal = this.abortController.signal; } async start() { - while (!this.cancelToken.isCancelled) { + while (!this.abortSignal.aborted) { try { await this.runOnce(); } catch (e) { this.logger.warn(`Error while performing janitorial tasks, ${e}`); } - await this.sleep(this.waitBetweenRuns); + await sleep(this.waitBetweenRuns, this.abortSignal); } } async stop() { - this.cancelToken.cancel(); + this.abortController.abort(); } private async runOnce() { @@ -79,15 +82,4 @@ export class PluginTaskManagerJanitor { } } } - - /** - * Sleeps for the given duration, but aborts sooner if the cancel token - * triggers. - */ - private async sleep(duration: Duration) { - await Promise.race([ - new Promise(resolve => setTimeout(resolve, duration.as('milliseconds'))), - this.cancelToken.promise, - ]); - } } diff --git a/packages/backend-tasks/src/tasks/TaskWorker.ts b/packages/backend-tasks/src/tasks/TaskWorker.ts index 881a013360..4b190c69c6 100644 --- a/packages/backend-tasks/src/tasks/TaskWorker.ts +++ b/packages/backend-tasks/src/tasks/TaskWorker.ts @@ -16,12 +16,12 @@ import { Knex } from 'knex'; import { Duration } from 'luxon'; +import { AbortController, AbortSignal } from 'node-abort-controller'; import { v4 as uuid } from 'uuid'; import { Logger } from 'winston'; import { DbTasksRow, DB_TASKS_TABLE } from '../database/tables'; -import { CancelToken } from './CancelToken'; import { TaskSettingsV1, taskSettingsV1Schema } from './types'; -import { nowPlus } from './util'; +import { nowPlus, sleep } from './util'; const WORK_CHECK_FREQUENCY = Duration.fromObject({ seconds: 5 }); @@ -35,7 +35,8 @@ export class TaskWorker { private readonly fn: () => void | Promise; private readonly knex: Knex; private readonly logger: Logger; - private readonly cancelToken: CancelToken; + private readonly abortController: AbortController; + private readonly abortSignal: AbortSignal; constructor( taskId: string, @@ -47,7 +48,8 @@ export class TaskWorker { this.fn = fn; this.knex = knex; this.logger = logger; - this.cancelToken = CancelToken.create(); + this.abortController = new AbortController(); + this.abortSignal = this.abortController.signal; } async start(settings: TaskSettingsV1) { @@ -63,13 +65,13 @@ export class TaskWorker { (async () => { try { - while (!this.cancelToken.isCancelled) { + while (!this.abortSignal.aborted) { const runResult = await this.runOnce(); if (runResult.result === 'abort') { break; } - await this.sleep(WORK_CHECK_FREQUENCY); + await sleep(WORK_CHECK_FREQUENCY, this.abortSignal); } this.logger.info(`Task worker finished: ${this.taskId}`); } catch (e) { @@ -79,7 +81,7 @@ export class TaskWorker { } stop() { - this.cancelToken.cancel(); + this.abortController.abort(); } /** @@ -120,19 +122,6 @@ export class TaskWorker { return { result: 'completed' }; } - /** - * Sleep for the given duration, but abort sooner if the cancel token - * triggers. - * - * @param duration - The amount of time to sleep, at most - */ - private async sleep(duration: Duration): Promise { - await Promise.race([ - new Promise(resolve => setTimeout(resolve, duration.as('milliseconds'))), - this.cancelToken.promise, - ]); - } - /** * Perform the initial store of the task info */ diff --git a/packages/backend-tasks/src/tasks/util.ts b/packages/backend-tasks/src/tasks/util.ts index dd5c2d4b7c..72753a46b2 100644 --- a/packages/backend-tasks/src/tasks/util.ts +++ b/packages/backend-tasks/src/tasks/util.ts @@ -17,6 +17,7 @@ import { InputError } from '@backstage/errors'; import { Knex } from 'knex'; import { DateTime, Duration } from 'luxon'; +import { AbortSignal } from 'node-abort-controller'; // Keep the IDs compatible with e.g. Prometheus export function validateId(id: string) { @@ -43,3 +44,34 @@ export function nowPlus(duration: Duration | undefined, knex: Knex) { ? knex.raw(`datetime('now', ?)`, [`${seconds} seconds`]) : knex.raw(`now() + interval '${seconds} seconds'`); } + +/** + * Sleep for the given duration, but return sooner if the abort signal + * triggers. + * + * @param duration - The amount of time to sleep, at most + * @param abortSignal - An optional abort signal that short circuits the wait + */ +export async function sleep( + duration: Duration, + abortSignal?: AbortSignal, +): Promise { + if (abortSignal?.aborted) { + return; + } + + await new Promise(resolve => { + let timeoutHandle: NodeJS.Timeout | undefined = undefined; + + const done = () => { + if (timeoutHandle) { + clearTimeout(timeoutHandle); + } + abortSignal?.removeEventListener('abort', done); + resolve(); + }; + + timeoutHandle = setTimeout(done, duration.as('milliseconds')); + abortSignal?.addEventListener('abort', done); + }); +}