Give other requests a turn: yield between batch entries, every 10 ms, and during the startup scan

tests/tx/batch-validate-yield.test.js

@@ -0,0 +1,42 @@
+/**
+ * $batch-validate-code gives the event loop a turn between entries.
+ *
+ * Validating one entry never yields by itself, so a 1,000-entry batch from an IG build used to
+ * hold the event loop for its whole duration (~1 s on production hardware), and every other
+ * request arriving meanwhile waited for it (a single lookup typically waited 0.4-0.9 s).
+ * processBatch now calls checkAndYield before each entry; checkAndYield only actually yields
+ * after YIELD_INTERVAL ms of compute, so small batches are unaffected.
+ */
+const request = require('supertest');
+const { getTestApp, shutdownTestApp } = require('./setup');
+const { OperationContext } = require('../../tx/operation-context');
+
+const CS_URL = 'http://example.org/fhir/CodeSystem/batch-yield-cs';
+const CS = {
+  resourceType: 'CodeSystem', url: CS_URL, version: '1.0.0', status: 'active', content: 'complete',
+  concept: [{ code: 'a', display: 'Alpha' }, { code: 'b', display: 'Beta' }]
+};
+const entry = (code) => ({ name: 'validation', resource: { resourceType: 'Parameters', parameter: [{ name: 'code', valueCode: code }] } });
+
+describe('$batch-validate-code yields between entries', () => {
+  let app;
+  beforeAll(async () => { app = await getTestApp(); }, 60000);
+  afterAll(async () => { await shutdownTestApp(); });
+
+  test('checkAndYield is called once per entry and the results are unchanged', async () => {
+    const spy = jest.spyOn(OperationContext.prototype, 'checkAndYield');
+    try {
+      const codes = ['a', 'zz', 'b', 'a', 'b'];
+      const res = await request(app).post('/tx/r5/CodeSystem/$batch-validate-code')
+        .set('Content-Type', 'application/json')
+        .send({ resourceType: 'Parameters', parameter: [{ name: 'tx-resource', resource: CS }, { name: 'url', valueUri: CS_URL }, ...codes.map(entry)] });
+      expect(res.status).toBe(200);
+      const calls = spy.mock.calls.filter((c) => c[0] === 'batch-validate').length;
+      expect(calls).toBe(codes.length);
+      const results = res.body.parameter.map((r) => r.resource.parameter.find((p) => p.name === 'result').valueBoolean);
+      expect(results).toEqual([true, false, true, true, true]);
+    } finally {
+      spy.mockRestore();
+    }
+  });
+});

tests/tx/operation-context-yield.test.js

@@ -199,3 +199,24 @@ describe('OperationContext.waitFor', () => {
     expect(ctx.computeElapsed()).toBeLessThan(200);
   });
 });
+
+describe('OperationContext.checkAndYield: the slice length', () => {
+  // A request that arrives while a long operation runs waits up to one slice for its turn
+  test('yields once 10 ms of compute have passed since the last yield', async () => {
+    const ctx = makeContext(30);
+    const now = jest.spyOn(performance, 'now');
+    try {
+      now.mockReturnValue(1000);
+      ctx._clock.lastYield = 1000;
+      ctx._clock.sliceStart = 1000;
+      now.mockReturnValue(1009);
+      await ctx.checkAndYield('test');
+      expect(ctx._clock.lastYield).toBe(1000); // 9 ms: no yield yet
+      now.mockReturnValue(1010);
+      await ctx.checkAndYield('test');
+      expect(ctx._clock.lastYield).toBe(1010); // 10 ms: yielded, and a new slice started
+    } finally {
+      now.mockRestore();
+    }
+  });
+});

tests/tx/usage-tracker-yield.test.js

@@ -0,0 +1,37 @@
+// The ConceptUsageTracker's startup scan reads every value set, about a second of work on a full
+// source list; it used to run without a break, so requests arriving just after startup waited
+// for all of it. It now gives the event loop a turn every few milliseconds.
+
+const ConceptUsageTracker = require('../../tx/usage-tracker');
+
+function busy(ms) {
+  const end = Date.now() + ms;
+  while (Date.now() < end) { /* compute */ }
+}
+
+function libraryOf(count) {
+  const vs = (i) => ({ jsonObj: { compose: { include: [{ system: 'http://example.org/cs', concept: [{ code: 'c' + (i % 7) }] }] } } });
+  return {
+    valueSetProviders: [{
+      listAllValueSets: async () => Array.from({ length: count }, (_, i) => 'http://example.org/vs/' + i),
+      fetchValueSet: async (url) => { busy(1); return vs(Number(url.split('/').pop())); }
+    }]
+  };
+}
+
+describe('ConceptUsageTracker scan', () => {
+  test('other work runs during the scan, and the counts are unchanged', async () => {
+    const tracker = new ConceptUsageTracker();
+    let ranDuringScan = 0;
+    let scanning = true;
+    const tick = () => { if (scanning) { ranDuringScan++; setImmediate(tick); } };
+    setImmediate(tick);
+    const count = await tracker.scanValueSets(libraryOf(100));
+    scanning = false;
+    expect(count).toBe(100);
+    expect(ranDuringScan).toBeGreaterThan(3);
+    const usages = tracker.usages('http://example.org/cs');
+    expect(usages.get('c0').count).toBe(15);
+    expect(usages.get('c6').count).toBe(14);
+  });
+});

tx/operation-context.js

@@ -824,7 +824,7 @@ const CHECK_FREQUENCY = 100;
 // I/O. checkAndYield() awaits a real setImmediate every YIELD_INTERVAL ms of
 // compute, so pending I/O (new connections, /metadata, timers) runs between
 // slices and one heavy $expand can no longer time out every other client.
-const YIELD_INTERVAL = 25;
+const YIELD_INTERVAL = 10;
 const { setImmediate: yieldToEventLoop } = require('timers/promises');
 
 class OperationContext {

tx/usage-tracker.js

@@ -1,4 +1,10 @@
 
+const { setImmediate: yieldToEventLoop } = require('timers/promises');
+
+// The startup scan reads every value set (thousands with the npm packages, VSAC), about a second
+// of work: give requests a turn every YIELD_INTERVAL_MS of it.
+const YIELD_INTERVAL_MS = 10;
+
 class ConceptUsageTracker {
 
   constructor() {
@@ -7,9 +13,14 @@ class ConceptUsageTracker {
 
   async scanValueSets(library) {
     let c = 0;
+    let sliceStart = Date.now();
     for (let vsp of library.valueSetProviders) {
       let list = await vsp.listAllValueSets();
       for (let url of list) {
+        if (Date.now() - sliceStart >= YIELD_INTERVAL_MS) {
+          await yieldToEventLoop();
+          sliceStart = Date.now();
+        }
         let vs = await vsp.fetchValueSet(url);
         if (vs && vs.jsonObj.compose) {
           if (await this.scanValueSet(vs.jsonObj.compose)) {

tx/workers/batch-validate.js

@@ -105,6 +105,10 @@ class BatchValidateWorker extends TerminologyWorker {
 
       for (const p of params.parameter) {
         if (p.name == 'validation') {
+          // Validating an entry never yields the event loop by itself, and IG builds send batches
+          // of up to 1,000 entries (~1 s of compute): give other requests a turn between entries
+          // (checkAndYield yields only after YIELD_INTERVAL ms of compute, so small batches don't).
+          await this.opContext.checkAndYield('batch-validate');
           let op = new Parameters();
           op.jsonObj.parameter = [];
           for (const gp of globalParams) {