From ffa209d6f336978f513a8770170a330b0c0272ab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Fri, 25 Aug 2023 15:44:48 +0200 Subject: [PATCH] moved around MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .../src/processing/TaskPipeline.test.ts | 87 ++++++++++++++- .../src/processing/TaskPipeline.ts | 63 +++++++++-- .../src/processing/util.test.ts | 102 ------------------ .../catalog-backend/src/processing/util.ts | 47 -------- 4 files changed, 142 insertions(+), 157 deletions(-) delete mode 100644 plugins/catalog-backend/src/processing/util.test.ts diff --git a/plugins/catalog-backend/src/processing/TaskPipeline.test.ts b/plugins/catalog-backend/src/processing/TaskPipeline.test.ts index f814e8a608..43ec118bdb 100644 --- a/plugins/catalog-backend/src/processing/TaskPipeline.test.ts +++ b/plugins/catalog-backend/src/processing/TaskPipeline.test.ts @@ -14,7 +14,7 @@ * limitations under the License. */ -import { startTaskPipeline } from './TaskPipeline'; +import { startTaskPipeline, createBarrier } from './TaskPipeline'; function createLimitedLoader(count: number, loadDelay?: number) { const items = new Array(count).fill(0).map((_, index) => index); @@ -119,3 +119,88 @@ describe('startTaskPipeline', () => { }).toThrow('must be lower'); }); }); + +describe('createBarrier', () => { + const tick = (millis: number) => + new Promise(resolve => setTimeout(resolve, millis)); + + it('abandons a wait after the timeout expires', async () => { + const abortController = new AbortController(); + const signal = abortController.signal; + const barrier = createBarrier({ waitTimeoutMillis: 100, signal }); + + const fn1 = jest.fn(); + barrier.wait().then(fn1); + + await tick(0); + expect(fn1).not.toHaveBeenCalled(); + + await tick(50); + expect(fn1).not.toHaveBeenCalled(); + + // start a new wait mid-way through the timeout + // should NOT resolve when the first one times out + const fn2 = jest.fn(); + barrier.wait().then(fn2); + + await tick(0); + expect(fn2).not.toHaveBeenCalled(); + + await tick(50); + expect(fn1).toHaveBeenCalledTimes(1); + expect(fn2).not.toHaveBeenCalled(); + + await tick(50); + expect(fn1).toHaveBeenCalledTimes(1); + expect(fn2).toHaveBeenCalledTimes(1); + }); + + it('abandons a wait after aborted', async () => { + const abortController = new AbortController(); + const signal = abortController.signal; + const barrier = createBarrier({ waitTimeoutMillis: 100, signal }); + + const fn1 = jest.fn(); + barrier.wait().then(fn1); + + // should resolve immediately, not after timeout + await tick(0); + expect(fn1).not.toHaveBeenCalled(); + abortController.abort(); + await tick(0); + expect(fn1).toHaveBeenCalledTimes(1); + + // subsequent waits should be immediate no matter what + const fn2 = jest.fn(); + barrier.wait().then(fn2); + await tick(0); + expect(fn2).toHaveBeenCalledTimes(1); + }); + + it('release immediately unblocks all waits', async () => { + const abortController = new AbortController(); + const signal = abortController.signal; + const barrier = createBarrier({ waitTimeoutMillis: 100, signal }); + + const fn1 = jest.fn(); + barrier.wait().then(fn1); + + await tick(50); + expect(fn1).not.toHaveBeenCalled(); + + // start a new wait mid-way through the timeout + // SHOULD resolve when releasing + const fn2 = jest.fn(); + barrier.wait().then(fn2); + + await tick(0); + expect(fn1).not.toHaveBeenCalled(); + expect(fn2).not.toHaveBeenCalled(); + + barrier.release(); + + await tick(0); + expect(fn1).toHaveBeenCalledTimes(1); + expect(fn2).toHaveBeenCalledTimes(1); + }); +}); diff --git a/plugins/catalog-backend/src/processing/TaskPipeline.ts b/plugins/catalog-backend/src/processing/TaskPipeline.ts index 24b4ec5ae3..06e5c5db53 100644 --- a/plugins/catalog-backend/src/processing/TaskPipeline.ts +++ b/plugins/catalog-backend/src/processing/TaskPipeline.ts @@ -14,8 +14,6 @@ * limitations under the License. */ -import { createBarrier } from './util'; - const DEFAULT_POLLING_INTERVAL_MS = 1000; type Options = { @@ -80,12 +78,12 @@ export function startTaskPipeline(options: Options) { const abortController = new AbortController(); const abortSignal = abortController.signal; - async function pipelineLoop() { - const barrier = createBarrier({ - waitTimeoutMillis: pollingIntervalMs, - signal: abortSignal, - }); + const barrier = createBarrier({ + waitTimeoutMillis: pollingIntervalMs, + signal: abortSignal, + }); + async function pipelineLoop() { while (!abortSignal.aborted) { if (state.inFlightCount <= lowWatermark) { const loadCount = highWatermark - state.inFlightCount; @@ -127,5 +125,56 @@ export function startTaskPipeline(options: Options) { return () => { abortController.abort(); + barrier.destroy(); + }; +} + +/** + * Creates a barrier with a timeout, that can be awaited or prematurely + * released either manually or by an abort signal. + */ +export function createBarrier(options: { + waitTimeoutMillis: number; + signal: AbortSignal; +}): { + wait: () => Promise; + release: () => void; + destroy: () => void; +} { + const { waitTimeoutMillis, signal } = options; + const resolvers = new Set<() => void>(); + + function wait() { + if (signal.aborted || !(waitTimeoutMillis > 0)) { + return Promise.resolve(); + } + + return new Promise(resolve => { + const timeoutHandle = setTimeout(done, waitTimeoutMillis); + + function done() { + resolvers.delete(done); + clearTimeout(timeoutHandle); + resolve(); + } + + resolvers.add(done); + }); + } + + function release() { + const resolversToCall = new Set(resolvers); + resolvers.clear(); + for (const resolver of resolversToCall) { + resolver(); + } + } + + signal.addEventListener('abort', release); + + return { + wait, + release, + destroy: () => signal.removeEventListener('abort', release), }; } diff --git a/plugins/catalog-backend/src/processing/util.test.ts b/plugins/catalog-backend/src/processing/util.test.ts deleted file mode 100644 index b8d10f7db6..0000000000 --- a/plugins/catalog-backend/src/processing/util.test.ts +++ /dev/null @@ -1,102 +0,0 @@ -/* - * Copyright 2023 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. - */ - -import { createBarrier } from './util'; - -describe('createBarrier', () => { - const tick = (millis: number) => - new Promise(resolve => setTimeout(resolve, millis)); - - it('abandons a wait after the timeout expires', async () => { - const abortController = new AbortController(); - const signal = abortController.signal; - const barrier = createBarrier({ waitTimeoutMillis: 100, signal }); - - const fn1 = jest.fn(); - barrier.wait().then(fn1); - - await tick(0); - expect(fn1).not.toHaveBeenCalled(); - - await tick(50); - expect(fn1).not.toHaveBeenCalled(); - - // start a new wait mid-way through the timeout - // should NOT resolve when the first one times out - const fn2 = jest.fn(); - barrier.wait().then(fn2); - - await tick(0); - expect(fn2).not.toHaveBeenCalled(); - - await tick(50); - expect(fn1).toHaveBeenCalledTimes(1); - expect(fn2).not.toHaveBeenCalled(); - - await tick(50); - expect(fn1).toHaveBeenCalledTimes(1); - expect(fn2).toHaveBeenCalledTimes(1); - }); - - it('abandons a wait after aborted', async () => { - const abortController = new AbortController(); - const signal = abortController.signal; - const barrier = createBarrier({ waitTimeoutMillis: 100, signal }); - - const fn1 = jest.fn(); - barrier.wait().then(fn1); - - // should resolve immediately, not after timeout - await tick(0); - expect(fn1).not.toHaveBeenCalled(); - abortController.abort(); - await tick(0); - expect(fn1).toHaveBeenCalledTimes(1); - - // subsequent waits should be immediate no matter what - const fn2 = jest.fn(); - barrier.wait().then(fn2); - await tick(0); - expect(fn2).toHaveBeenCalledTimes(1); - }); - - it('release immediately unblocks all waits', async () => { - const abortController = new AbortController(); - const signal = abortController.signal; - const barrier = createBarrier({ waitTimeoutMillis: 100, signal }); - - const fn1 = jest.fn(); - barrier.wait().then(fn1); - - await tick(50); - expect(fn1).not.toHaveBeenCalled(); - - // start a new wait mid-way through the timeout - // SHOULD resolve when releasing - const fn2 = jest.fn(); - barrier.wait().then(fn2); - - await tick(0); - expect(fn1).not.toHaveBeenCalled(); - expect(fn2).not.toHaveBeenCalled(); - - barrier.release(); - - await tick(0); - expect(fn1).toHaveBeenCalledTimes(1); - expect(fn2).toHaveBeenCalledTimes(1); - }); -}); diff --git a/plugins/catalog-backend/src/processing/util.ts b/plugins/catalog-backend/src/processing/util.ts index 2e1d3a7093..cd93be3c8b 100644 --- a/plugins/catalog-backend/src/processing/util.ts +++ b/plugins/catalog-backend/src/processing/util.ts @@ -86,50 +86,3 @@ export function isObject(value: JsonValue | undefined): value is JsonObject { export const validateEntity = entitySchemaValidator(); export const validateEntityEnvelope = entityEnvelopeSchemaValidator(); - -/** - * Creates a barrier with a timeout, that can be awaited or prematurely - * released either manually or by an abort signal. - */ -export function createBarrier(options: { - waitTimeoutMillis: number; - signal: AbortSignal; -}): { - wait: () => Promise; - release: () => void; -} { - const { waitTimeoutMillis, signal } = options; - const resolvers: Array<() => void> = []; - - function wait() { - if (signal.aborted || !(waitTimeoutMillis > 0)) { - return Promise.resolve(); - } - - return new Promise(resolve => { - const timeoutHandle = setTimeout(done, waitTimeoutMillis); - - function done() { - const index = resolvers.indexOf(done); - if (index !== -1) { - resolvers.splice(index, 1); - } - clearTimeout(timeoutHandle); - resolve(); - } - - resolvers.push(done); - }); - } - - function release() { - const resolversToCall = resolvers.splice(0, resolvers.length); - for (const resolver of resolversToCall) { - resolver(); - } - } - - signal.addEventListener('abort', release); - - return { wait, release }; -}