tricu

A lazy functional language based on Tree Calculus
Log | Files | Refs | README | LICENSE

io-driver.js (4321B)

      1 // Unlike repeated Promise.race against a lifetime-long promise, listeners are
      2 // removed on EVERY turn/step settlement, not retained until runtime disposal.
      3 function interrupt(promise, signal, aborted) {
      4     if (!signal)
      5         return promise;
      6     let onAbort = () => { };
      7     return new Promise((resolve, reject) => {
      8         onAbort = () => { try {
      9             resolve(aborted());
     10         }
     11         catch (error) {
     12             reject(error);
     13         } };
     14         if (signal.aborted)
     15             onAbort();
     16         else
     17             signal.addEventListener('abort', onAbort, { once: true });
     18         void promise.then(resolve, reject);
     19     }).finally(() => signal.removeEventListener('abort', onAbort));
     20 }
     21 // The queue stays alive while a read or step prompt is waiting. Stop is sticky,
     22 // independent of the resume queue, and never implemented by worker destruction.
     23 export async function drive(driver, host) {
     24     const transport = host.structured;
     25     if (!transport)
     26         throw new Error('IO version 2 requires a structured terminal host');
     27     let settled = false;
     28     const alive = (promise) => interrupt(promise, driver.signal, () => { throw driver.signal.reason; });
     29     let active = true, stopping = false, stopSent = false, wake = () => { }, step = 0;
     30     const replies = [];
     31     let boundaryFault = false, fault;
     32     const stopped = new AbortController();
     33     const enqueue = (command) => {
     34         if (!active || !command || command.run !== driver.run)
     35             return;
     36         if (command.kind === 'stop') {
     37             stopping = true;
     38             stopped.abort();
     39         }
     40         else if (command.kind === 'reply')
     41             replies.push(command);
     42         wake();
     43     };
     44     const close = () => {
     45         if (!active)
     46             return;
     47         active = false;
     48         stopped.abort();
     49         wake();
     50         wake = () => { };
     51         replies.length = 0;
     52         // A failed transport/runtime cannot be reused with admitted work abandoned.
     53         if (!settled)
     54             driver.discard?.();
     55         transport.close(driver.run, settled);
     56     };
     57     try {
     58         transport.open(driver.run, enqueue);
     59         for (;;) {
     60             if (!stopping && transport.stepping === true) {
     61                 try {
     62                     await alive(interrupt(host.before({ step: ++step, name: 'IO scheduler turn', detail: 'one scheduler advance; all tasks paused', depth: 0 }), stopped.signal, () => { }));
     63                 }
     64                 catch (error) {
     65                     if (driver.signal?.aborted)
     66                         throw error;
     67                     boundaryFault = true;
     68                     fault = error;
     69                     enqueue({ run: driver.run, kind: 'stop' });
     70                 }
     71             }
     72             const command = stopping && !stopSent ? (stopSent = true, { run: driver.run, kind: 'stop' }) :
     73                 replies.shift() ?? { run: driver.run, kind: 'poll' };
     74             const report = await alive(driver.turn(command));
     75             for (const event of report.events)
     76                 transport.event(event);
     77             if (report.done) {
     78                 if (!Object.values(report.counts).every(group => Object.values(group).every(value => value === 0)))
     79                     throw new Error('IO settlement retains work');
     80                 settled = true;
     81                 close();
     82                 if (boundaryFault)
     83                     throw fault;
     84                 if (report.outcome?.fault)
     85                     throw new Error('IO interpreter fault');
     86                 if (report.outcome?.kind !== 'completed')
     87                     throw new Error(`IO ${report.outcome?.kind ?? 'failed'}`);
     88                 return await alive(driver.view()); // Full codec, only after physical settlement.
     89             }
     90             // Yield even during pure scheduler progress so the outer worker can receive
     91             // stop/replies. This is not a hard deadline for native tree reduction.
     92             if (report.waiting && !replies.length && (!stopping || stopSent))
     93                 await alive(new Promise(resolve => { wake = resolve; }));
     94             else
     95                 await alive(new Promise(resolve => { const timer = setTimeout(resolve, 0); wake = () => { clearTimeout(timer); resolve(); }; }));
     96             wake = () => { };
     97         }
     98     }
     99     finally {
    100         close();
    101     }
    102 }