Fb/exclusive init (#52)

* serialize init so that double execution is really avoided

* start of semaphore.test.js

* makeExclusiveQueuing and makeExclusiveReturning

* add semaphore test

* semaphore test 2

* semaphore test 3

* semaphore test 4

Author: rlindner81

package.json

@@ -9,6 +9,7 @@
     "index.cds"
   ],
   "scripts": {
+    "postinstall": "npx patch-package",
     "test": "jest",
     "test:remove-inline-snapshots": "npx replace '\\.toMatchInlineSnapshot\\(\\s*`[\\s\\S]*?`\\s*\\);' '.toMatchInlineSnapshot();' test -r --include='*.test.js'",
     "lint": "npm run prettier && npm run eslint",

src/featureToggles.js

@@ -26,6 +26,7 @@ const { HandlerCollection } = require("./shared/handlerCollection");
 const { ENV, isObject, tryRequire } = require("./shared/static");
 const { promiseAllDone } = require("./shared/promiseAllDone");
 const { LimitedLazyCache } = require("./shared/cache");
+const { Semaphore } = require("./shared/semaphore");
 
 const ENV_UNIQUE_NAME = process.env[ENV.UNIQUE_NAME];
 const DEFAULT_REDIS_CHANNEL = process.env[ENV.REDIS_CHANNEL] || "features";
@@ -644,7 +645,7 @@ class FeatureToggles {
    * Initialize needs to run and finish before other APIs are called. It processes the configuration, sets up
    * related internal state, and starts communication with redis.
    */
-  async initializeFeatures({ config: configInput, configFile: configFilepath = DEFAULT_CONFIG_FILEPATH }) {
+  async _initializeFeatures({ config: configInput, configFile: configFilepath = DEFAULT_CONFIG_FILEPATH } = {}) {
     if (this.__isInitialized) {
       return;
     }
@@ -760,6 +761,10 @@ class FeatureToggles {
     return this;
   }
 
+  async initializeFeatures(options) {
+    return await Semaphore.makeExclusiveReturning(this._initializeFeatures.bind(this))(options);
+  }
+
   // ========================================
   // END OF INITIALIZE SECTION
   // ========================================

src/redisWrapper.js

@@ -22,7 +22,6 @@ const INTEGRATION_MODE = Object.freeze({
 const CF_REDIS_SERVICE_LABEL = "redis-cache";
 
 const logger = new Logger(COMPONENT_NAME);
-const watchedGetSetSemaphore = new Semaphore();
 
 const MODE = Object.freeze({
   RAW: "raw",
@@ -300,15 +299,6 @@ const setObject = async (key, value, options) => {
  */
 const del = async (key) => await _clientExec("DEL", { key });
 
-const _watchedGetSetExclusive = async (key, newValueCallback, options) => {
-  await watchedGetSetSemaphore.acquire();
-  try {
-    return await _watchedGetSet(key, newValueCallback, options);
-  } finally {
-    watchedGetSetSemaphore.release();
-  }
-};
-
 const _watchedGetSet = async (key, newValueCallback, { field, mode = MODE.OBJECT, attempts = 10 } = {}) => {
   const useHash = field !== undefined;
   if (!mainClient) {
@@ -373,6 +363,8 @@ const _watchedGetSet = async (key, newValueCallback, { field, mode = MODE.OBJECT
   );
 };
 
+const _watchedGetSetExclusive = Semaphore.makeExclusiveQueuing(_watchedGetSet);
+
 /**
  * @callback NewValueCallback
  * @see watchedGetSet

src/shared/semaphore.js

@@ -53,6 +53,41 @@ class Semaphore {
       this.resolveCurrentSemaphore();
     }
   }
+
+  /**
+   * Take an async function and turn it into an exclusively executing async function. Calls during async execution will
+   * be queued and executed serially.
+   */
+  static makeExclusiveQueuing(cb) {
+    const semaphore = new Semaphore();
+    return async (...args) => {
+      await semaphore.acquire();
+      try {
+        return await cb(...args);
+      } finally {
+        semaphore.release();
+      }
+    };
+  }
+
+  /**
+   * Take an async function and turn it into an exclusively executing async function. Calls during async execution will
+   * be returned.
+   */
+  static makeExclusiveReturning(cb) {
+    let isRunning;
+    return async (...args) => {
+      if (isRunning) {
+        return;
+      }
+      isRunning = true;
+      try {
+        return await cb(...args);
+      } finally {
+        isRunning = false;
+      }
+    };
+  }
 }
 
 module.exports = { Semaphore };

test/shared/semaphore.test.js

@@ -0,0 +1,118 @@
+"use strict";
+const { promisify, format } = require("util");
+
+const { Semaphore } = require("../../src/shared/semaphore");
+
+const sleep = promisify(setTimeout);
+
+describe("semaphore", () => {
+  let executionLog;
+  const n = 3;
+  const m = 2;
+  const result = "finished";
+  const runner = async (index) => {
+    executionLog.push(format("started %d", index));
+    await sleep(10);
+    executionLog.push(format("finished %d", index));
+    return result;
+  };
+
+  beforeEach(() => {
+    executionLog = [];
+  });
+
+  test("non-exclusive", async () => {
+    const resultsPrimary = await Promise.all(Array.from({ length: n }, (_, i) => runner(i + 1)));
+
+    const resultsSecondary = await Promise.all(Array.from({ length: m }, (_, i) => runner(n + i + 1)));
+    expect(resultsPrimary).toMatchInlineSnapshot(`
+      [
+        "finished",
+        "finished",
+        "finished",
+      ]
+    `);
+    expect(resultsSecondary).toMatchInlineSnapshot(`
+      [
+        "finished",
+        "finished",
+      ]
+    `);
+    expect(executionLog).toMatchInlineSnapshot(`
+      [
+        "started 1",
+        "started 2",
+        "started 3",
+        "finished 1",
+        "finished 2",
+        "finished 3",
+        "started 4",
+        "started 5",
+        "finished 4",
+        "finished 5",
+      ]
+    `);
+  });
+
+  test("exclusive queueing", async () => {
+    const exclusiveRunner = Semaphore.makeExclusiveQueuing(runner);
+    const resultsPrimary = await Promise.all(Array.from({ length: n }, (_, i) => exclusiveRunner(i + 1)));
+
+    const resultsSecondary = await Promise.all(Array.from({ length: m }, (_, i) => exclusiveRunner(n + i + 1)));
+    expect(resultsPrimary).toMatchInlineSnapshot(`
+      [
+        "finished",
+        "finished",
+        "finished",
+      ]
+    `);
+    expect(resultsSecondary).toMatchInlineSnapshot(`
+      [
+        "finished",
+        "finished",
+      ]
+    `);
+    expect(executionLog).toMatchInlineSnapshot(`
+      [
+        "started 1",
+        "finished 1",
+        "started 2",
+        "finished 2",
+        "started 3",
+        "finished 3",
+        "started 4",
+        "finished 4",
+        "started 5",
+        "finished 5",
+      ]
+    `);
+  });
+
+  test("exclusive returning", async () => {
+    const exclusiveRunner = Semaphore.makeExclusiveReturning(runner);
+    const resultsPrimary = await Promise.all(Array.from({ length: n }, (_, i) => exclusiveRunner(i + 1)));
+
+    const resultsSecondary = await Promise.all(Array.from({ length: m }, (_, i) => exclusiveRunner(n + i + 1)));
+    expect(resultsPrimary).toMatchInlineSnapshot(`
+      [
+        "finished",
+        undefined,
+        undefined,
+      ]
+    `);
+    expect(resultsSecondary).toMatchInlineSnapshot(`
+      [
+        "finished",
+        undefined,
+      ]
+    `);
+    expect(executionLog).toMatchInlineSnapshot(`
+      [
+        "started 1",
+        "finished 1",
+        "started 4",
+        "finished 4",
+      ]
+    `);
+  });
+});