From 1fd2109739c19d1c01c0588e93abb2059d92e12e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Fri, 25 Aug 2023 12:41:53 +0200 Subject: [PATCH 1/3] make startTaskPipeline iterative MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- .changeset/cold-apricots-poke.md | 5 + .../src/processing/TaskPipeline.ts | 84 ++++++++------- .../src/processing/util.test.ts | 102 ++++++++++++++++++ .../catalog-backend/src/processing/util.ts | 47 ++++++++ 4 files changed, 201 insertions(+), 37 deletions(-) create mode 100644 .changeset/cold-apricots-poke.md create mode 100644 plugins/catalog-backend/src/processing/util.test.ts diff --git a/.changeset/cold-apricots-poke.md b/.changeset/cold-apricots-poke.md new file mode 100644 index 0000000000..b050c8082b --- /dev/null +++ b/.changeset/cold-apricots-poke.md @@ -0,0 +1,5 @@ +--- +'@backstage/plugin-catalog-backend': patch +--- + +Changed the processing loop task pipeline implementation from recursive to iterative diff --git a/plugins/catalog-backend/src/processing/TaskPipeline.ts b/plugins/catalog-backend/src/processing/TaskPipeline.ts index 4afe2408a8..24b4ec5ae3 100644 --- a/plugins/catalog-backend/src/processing/TaskPipeline.ts +++ b/plugins/catalog-backend/src/processing/TaskPipeline.ts @@ -14,6 +14,8 @@ * limitations under the License. */ +import { createBarrier } from './util'; + const DEFAULT_POLLING_INTERVAL_MS = 1000; type Options = { @@ -72,50 +74,58 @@ export function startTaskPipeline(options: Options) { throw new Error('lowWatermark must be lower than highWatermark'); } - let loading = false; - let stopped = false; - let inFlightCount = 0; + // State is in an object so that it can be stably referenced from within + // callbacks below + const state = { inFlightCount: 0 }; + const abortController = new AbortController(); + const abortSignal = abortController.signal; - async function maybeLoadMore() { - if (stopped || loading || inFlightCount > lowWatermark) { - return; - } - - // Once we hit the low watermark we load in enough items to reach the high watermark - loading = true; - const loadCount = highWatermark - inFlightCount; - const loadedItems = await loadTasks(loadCount); - loading = false; - - // We might not reach the high watermark here, in case there weren't enough items to load - inFlightCount += loadedItems.length; - loadedItems.forEach(item => { - processTask(item).finally(() => { - if (stopped) { - return; - } - - // For each item we complete we check if it's time to load more - inFlightCount -= 1; - maybeLoadMore(); - }); + async function pipelineLoop() { + const barrier = createBarrier({ + waitTimeoutMillis: pollingIntervalMs, + signal: abortSignal, }); - // We might have processed some tasks while we where loading, so check if we can load more - if (loadedItems.length > 1) { - maybeLoadMore(); + while (!abortSignal.aborted) { + if (state.inFlightCount <= lowWatermark) { + const loadCount = highWatermark - state.inFlightCount; + const loadedItems = await Promise.resolve() + .then(() => loadTasks(loadCount)) + .catch(() => { + // Silently swallow errors and go back to sleep to try again; we + // delegate to the loadTasks function itself to catch errors and log + // if it so desires + return []; + }); + if (loadedItems.length && !abortSignal.aborted) { + state.inFlightCount += loadedItems.length; + for (const item of loadedItems) { + Promise.resolve() + .then(() => processTask(item)) + .catch(() => { + // Silently swallow errors and go back to sleep to try again; we + // delegate to the processTask function itself to catch errors + // and log if it so desires + }) + .finally(() => { + state.inFlightCount -= 1; + barrier.release(); + }); + } + } + } + + await barrier.wait(); } } - // This interval makes sure that we load in new items if the loop runs - // dry because of the lack of available tasks. As long as there are - // enough items to process this will be a noop. - const intervalId = setInterval(() => { - maybeLoadMore(); - }, pollingIntervalMs); + pipelineLoop().catch(_error => { + // This should be impossible, but if it did happen, it would signal a + // programming error inside the loop (errors should definitely be caught + // inside of it). + }); return () => { - stopped = true; - clearInterval(intervalId); + abortController.abort(); }; } diff --git a/plugins/catalog-backend/src/processing/util.test.ts b/plugins/catalog-backend/src/processing/util.test.ts new file mode 100644 index 0000000000..b8d10f7db6 --- /dev/null +++ b/plugins/catalog-backend/src/processing/util.test.ts @@ -0,0 +1,102 @@ +/* + * 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 cd93be3c8b..2e1d3a7093 100644 --- a/plugins/catalog-backend/src/processing/util.ts +++ b/plugins/catalog-backend/src/processing/util.ts @@ -86,3 +86,50 @@ 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 }; +} 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 2/3] 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 }; -} From 12e081d5c51fb58929b6748f1b38abb01bf2025b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fredrik=20Adel=C3=B6w?= Date: Fri, 25 Aug 2023 16:03:29 +0200 Subject: [PATCH 3/3] rethrow better MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Fredrik Adelöw --- plugins/catalog-backend/src/processing/TaskPipeline.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/plugins/catalog-backend/src/processing/TaskPipeline.ts b/plugins/catalog-backend/src/processing/TaskPipeline.ts index 06e5c5db53..44030be9e0 100644 --- a/plugins/catalog-backend/src/processing/TaskPipeline.ts +++ b/plugins/catalog-backend/src/processing/TaskPipeline.ts @@ -117,10 +117,13 @@ export function startTaskPipeline(options: Options) { } } - pipelineLoop().catch(_error => { + pipelineLoop().catch(error => { // This should be impossible, but if it did happen, it would signal a // programming error inside the loop (errors should definitely be caught - // inside of it). + // inside of it). Let's rethrow with more information, and let it be caught + // by the process' uncaught exception handler, which will log the occurrence + // at a high level. + throw new Error(`Unexpected error in processing pipeline loop`, error); }); return () => {