diff --git a/packages/controller/src/Controller.ts b/packages/controller/src/Controller.ts index 10bfacf6..740b54d2 100644 --- a/packages/controller/src/Controller.ts +++ b/packages/controller/src/Controller.ts @@ -376,9 +376,6 @@ export default class Controller { } } - private _saveDataCurrentlyRunning = false; - private _saveDataCurrentlyWaiting: (() => void)[] = []; - /** * Save all data currently in memory to disk * @@ -386,31 +383,18 @@ export default class Controller { * completes and start another save. */ async saveData() { - if (this._saveDataCurrentlyRunning) { - const waitForCompletion = () => new Promise(resolve => { - this._saveDataCurrentlyWaiting.push(resolve); - }); - await waitForCompletion(); - if (this._saveDataCurrentlyRunning) { - await waitForCompletion(); - return; - } - } + await this._saveDataAsyncSerial.invoke(); + } - this._saveDataCurrentlyRunning = true; + private _saveDataAsyncSerial = new lib.AsyncSerialMergingCallback(async () => { try { await this._saveDataInternal(); } catch (err: any) { logger.error(`Unexpected error during saveData:\n${err.stack}`); } - this._saveDataCurrentlyRunning = false; - for (const callback of this._saveDataCurrentlyWaiting) { - callback(); - } - this._saveDataCurrentlyWaiting.length = 0; - } + }); - async _saveDataInternal() { + private async _saveDataInternal() { if (this.config.dirty) { this.config.dirty = false; await lib.safeOutputFile(this.configPath, JSON.stringify(this.config.serialize(), null, "\t")); diff --git a/packages/lib/src/helpers.ts b/packages/lib/src/helpers.ts index a3e58571..38c804b3 100644 --- a/packages/lib/src/helpers.ts +++ b/packages/lib/src/helpers.ts @@ -51,6 +51,62 @@ export async function timeout(promise: Promise, time: number, timeoutResul } } +/** + * Helper to serialise and merge waiting calls to an async callback. + * + * Provides a convenient interface to serialise the calls made to an async + * callback such that no overlapping invocations of the callback is made. + * If the callback is already running when another invocation is requested + * the request will be queued until the callback returns and then an the + * callback called again. + * + * If multiple calls are made while the callback is running then these will + * be merged into one call of the callback. + */ +export class AsyncSerialMergingCallback { + private _currentlyRunning = false; + private _currentlyWaiting: (() => void)[] = []; + + /** + * @param callback - Async function to serialise access to + */ + constructor( + public callback: () => Promise, + ) { } + + /** + * Invoke the assosiated callback. + * + * If the callback is currently running then this will wait until the + * existing invocation finishes and then invoke it again. If called + * multiple times while an invocation is running the calls will be + * merged into one call. + */ + async invoke() { + if (this._currentlyRunning) { + const waitForCompletion = () => new Promise(resolve => { + this._currentlyWaiting.push(resolve); + }); + await waitForCompletion(); + if (this._currentlyRunning) { + await waitForCompletion(); + return; + } + } + + this._currentlyRunning = true; + try { + await this.callback(); + } finally { + this._currentlyRunning = false; + for (const waiter of this._currentlyWaiting) { + waiter(); + } + this._currentlyWaiting.length = 0; + } + } +} + /** * Read stream to the end and return its content diff --git a/test/lib/helpers.js b/test/lib/helpers.js index 57b23772..f9466d75 100644 --- a/test/lib/helpers.js +++ b/test/lib/helpers.js @@ -44,6 +44,66 @@ describe("lib/helpers", function() { }); }); + describe("class AsyncSerialMergingCallback", function() { + let started; + let ended; + let merger; + beforeEach(function() { + started = 0; + ended = 0; + merger = new lib.AsyncSerialMergingCallback( + () => new Promise(resolve => { + started += 1; + process.nextTick(() => { + ended += 1; + resolve(); + }); + }) + ); + }); + it("should await the callback when invoked", async function() { + await merger.invoke(); + assert.equal(started, 1, "incorrect started count"); + assert.equal(ended, 1, "incorrect ended count"); + }); + it("should immediately call the callback", async function() { + let promise = merger.invoke(); + assert.equal(started, 1, "incorrect started count"); + assert.equal(ended, 0, "incorrect ended count"); + await promise; + assert.equal(started, 1, "incorrect started count"); + assert.equal(ended, 1, "incorrect ended count"); + }); + it("should call the callback twice if invoked in twice in paralell", async function() { + await Promise.all([ + merger.invoke(), + merger.invoke(), + ]); + assert.equal(started, 2, "incorrect started count"); + assert.equal(ended, 2, "incorrect ended count"); + }); + it("should call the callback twice if invoked many times in paralell", async function() { + await Promise.all([ + merger.invoke(), + merger.invoke(), + merger.invoke(), + merger.invoke(), + ]); + assert.equal(started, 2, "incorrect started count"); + assert.equal(ended, 2, "incorrect ended count"); + }); + it("should serialise calls", async function() { + let events = []; + await Promise.all([ + merger.invoke().then(() => { events.push(1); }), + merger.invoke().then(() => { events.push(2); }), + merger.invoke().then(() => { events.push(3); }), + merger.invoke().then(() => { events.push(4); }), + ]); + assert.deepEqual(events, [1, 2, 3, 4]); + }); + }); + function parse(input, attributes) { const result = lib.parseSearchString(input, attributes); if (!result.issues.length) {