Multithreaded char indexing

build/char-index.js

@@ -2,65 +2,146 @@
 
 'use strict';
 
-function doAdd()
+async function doAdd()
 {
-    function indexCharacters(solutionBookMap)
+    function * generateQueue(indicator, solutionBookMap)
     {
-        const progress = require('./common/progress');
-
+        const queue = new Set();
         for (const char of charSet)
         {
-            const formattedCharacter = formatCharacter(char);
-            solutionBookMap.delete(char);
-            progress
+            const promise =
             (
-                `Indexing character ${formattedCharacter}`,
-                bar =>
-                solutionBookMap.index
-                (
-                    char,
-                    progress => bar.update(progress),
-                    char =>
-                    console.warn('Required character %s not indexed', formatCharacter(char)),
-                ),
-            );
-            console.log('Character %s indexed', formattedCharacter);
-            if (!noSave)
-                solutionBookMap.save();
+                async noSave =>
+                {
+                    const formattedCharacter = formatCharacter(char);
+                    const bar = indicator.newBar(`Indexing character ${formattedCharacter}`);
+                    const serializedSolutionBook = await runWorker(solutionBookMap, bar, char);
+                    bar.hide();
+                    const solutionBook = solutionBookMap.deserialize(serializedSolutionBook);
+                    solutionBookMap.set(char, solutionBook);
+                    console.log('Character %s indexed', formattedCharacter);
+                    if (!noSave)
+                        solutionBookMap.save();
+                    queue.delete(promise);
+                }
+            )
+            (noSave);
+            queue.add(promise);
+            if (queue.size >= concurrency)
+                yield queue;
         }
+        while (queue.size)
+            yield queue;
     }
 
-    function run()
+    async function indexCharacters()
     {
+        const progress = require('./common/progress');
+
         const solutionBookMap = getSolutionBookMap(!noLoad);
-        indexCharacters(solutionBookMap);
+        await progress
+        (
+            async indicator =>
+            {
+                const generator = generateQueue(indicator, solutionBookMap);
+                for (const queue of generator)
+                    await Promise.race(queue);
+            },
+        );
+    }
+
+    function runWorker(solutionBookMap, bar, char)
+    {
+        const executor =
+        (resolve, reject) =>
+        {
+            const { join }      = require('path');
+            const { Worker }    = require('worker_threads');
+
+            const workerFileName = join(__dirname, 'common/char-index-worker.js');
+            const serializedSolutionBookMap = solutionBookMap.serialize(solutionBookMap);
+            const worker =
+            new Worker(workerFileName, { workerData: { char, serializedSolutionBookMap } });
+            worker.on
+            (
+                'message',
+                ({ progress, missingChar, serializedSolutionBook }) =>
+                {
+                    if (progress != null)
+                        bar.update(progress);
+                    if (missingChar != null)
+                    {
+                        console.warn
+                        (
+                            'Character %s required by %s is not indexed',
+                            formatCharacter(missingChar),
+                            formatCharacter(char),
+                        );
+                    }
+                    if (serializedSolutionBook != null)
+                        resolve(serializedSolutionBook);
+                },
+            );
+            worker.on('error', reject);
+            worker.on
+            (
+                'exit',
+                code => reject(new Error(`Worker stopped unexpectedly with exit code ${code}`)),
+            );
+        };
+        const promise = new Promise(executor);
+        return promise;
     }
 
     let noLoad = false;
     let noSave = false;
+    let concurrency;
     const charSet =
     parseArguments
     (
         sequence =>
         {
-            switch (sequence)
+            const match = /concurrency(?:=(?<concurrency>.*))?/.exec(sequence);
+            if (match)
+            {
+                concurrency = Number(match.groups.concurrency);
+                if (concurrency !== Math.floor(concurrency) || concurrency < 1 || concurrency > 10)
+                {
+                    console.error('Concurrency must specify an integer between 1 and 10', sequence);
+                    throw ARG_ERROR;
+                }
+            }
+            else
             {
-            case 'new':
-                noLoad = true;
-                break;
-            case 'test':
-                noSave = true;
-                break;
-            default:
-                return false;
+                switch (sequence)
+                {
+                case 'new':
+                    noLoad = true;
+                    break;
+                case 'test':
+                    noSave = true;
+                    break;
+                default:
+                    return false;
+                }
             }
             return true;
         },
     );
+    if (!concurrency)
+    {
+        const { cpus } = require('os');
+        const cpuCount = cpus().length;
+        if (cpuCount < 3)
+            concurrency = cpuCount;
+        else
+            concurrency = Math.min(Math.ceil(cpuCount / 2), 10);
+    }
+
     {
         const timeUtils = require('../tools/time-utils');
 
-        const duration = timeUtils.timeThis(run);
+        const duration = await timeUtils.timeThisAsync(indexCharacters);
         const durationStr = timeUtils.formatDuration(duration);
         console.log('%s elapsed.', durationStr);
     }
@@ -322,20 +403,24 @@ else
 
     const [,, command] = argv;
     const commandCall = COMMANDS[command];
-    try
+    (async () =>
     {
-        if (!commandCall)
+        try
         {
-            console.error
-            ('char-index: \'%s\' is not a valid command. See \'char-index help\'.', command);
-            throw ARG_ERROR;
+            if (!commandCall)
+            {
+                console.error
+                ('char-index: \'%s\' is not a valid command. See \'char-index help\'.', command);
+                throw ARG_ERROR;
+            }
+            await commandCall();
+        }
+        catch (error)
+        {
+            if (error !== ARG_ERROR)
+                throw error;
+            process.exitCode = 1;
         }
-        commandCall();
-    }
-    catch (error)
-    {
-        if (error !== ARG_ERROR)
-            throw error;
-        process.exitCode = 1;
     }
+    )();
 }

build/common/char-index-worker.js

@@ -0,0 +1,17 @@
+'use strict';
+
+const solutionBookMap =
+require('./solution-book-map');
+const { parentPort, workerData: { char, serializedSolutionBookMap } } = require('worker_threads');
+
+solutionBookMap.load(serializedSolutionBookMap);
+solutionBookMap.delete(char);
+const solutionBook =
+solutionBookMap.index
+(
+    char,
+    progress => parentPort.postMessage({ progress }),
+    missingChar => parentPort.postMessage({ missingChar }),
+);
+const serializedSolutionBook = solutionBookMap.serialize(solutionBook);
+parentPort.postMessage({ serializedSolutionBook });

build/common/progress.js

@@ -6,6 +6,14 @@ const COMPLETE_CHAR     = chalk.bgBlue(' ');
 const INCOMPLETE_CHAR   = chalk.bgWhite(' ');
 const BAR_WIDTH         = 20;
 
+function deleteBars()
+{
+    if (bars.length > 1)
+        stream.moveCursor(0, 1 - bars.length);
+    stream.cursorTo(0);
+    stream.clearScreenDown();
+}
+
 function formatTime(millisecs)
 {
     let str;
@@ -25,33 +33,48 @@ function formatTime(millisecs)
     return str;
 }
 
+function writeBars()
+{
+    stream.cursorTo(0);
+    stream.clearScreenDown();
+    bars.forEach
+    (
+        (bar, index) =>
+        {
+            if (index)
+                stream.moveCursor(0, 1);
+            stream.cursorTo(0);
+            stream.write(bar.lastDraw);
+        },
+    );
+}
+
 class ProgressBar
 {
-    constructor(fmt)
+    constructor(label)
     {
-        this.fmt        = fmt;
-        this.stream     = process.stdout;
+        const prefix = label == null ? '' : `${String(label).replace(/[\0-\x08\x0a-\x1f]/g, '')} `;
+        this.format     = `${prefix}\0bar \0percent / \0eta left`;
+        this.lastDraw   = undefined;
         this.progress   = 0;
-        this.lastDraw   = '';
         this.start      = new Date();
     }
 
     hide()
     {
-        const { stream } = this;
         if (!stream.isTTY)
             return;
         if (this.lastDraw)
         {
-            stream.clearLine();
-            stream.cursorTo(0);
-            this.lastDraw = '';
+            bars.splice(bars.indexOf(this), 1);
+            stream.moveCursor(0, -bars.length);
+            this.lastDraw = undefined;
+            writeBars();
         }
     }
 
     render()
     {
-        const { stream } = this;
         if (!stream.isTTY)
             return;
         const { progress } = this;
@@ -60,7 +83,9 @@ class ProgressBar
 
         // Populate the bar template with percentages and timestamps.
         let str =
-        this.fmt.replace('\0eta', formatTime(eta)).replace('\0percent', `${percent.toFixed(0)}%`);
+        this.format
+        .replace('\0eta', formatTime(eta))
+        .replace('\0percent', `${percent.toFixed(0)}%`);
 
         {
             // Compute the available space for the bar.
@@ -76,10 +101,12 @@ class ProgressBar
 
         if (this.lastDraw !== str)
         {
-            stream.clearLine();
-            stream.cursorTo(0);
-            stream.write(str);
+            if (bars.length > 1)
+                stream.moveCursor(0, 1 - bars.length);
+            if (this.lastDraw === undefined)
+                bars.push(this);
             this.lastDraw = str;
+            writeBars();
         }
     }
 
@@ -92,9 +119,16 @@ class ProgressBar
     }
 }
 
-function progress(label, fn)
+const stream = process.stdout;
+const bars = [];
+
+async function progress(label, fn)
 {
-    label = String(label).replace(/[\0-\x08\x0a-\x1f]/g, '');
+    if (fn === undefined)
+    {
+        fn = label;
+        label = undefined;
+    }
     // 'count', 'group', 'groupCollapsed', 'table', 'timeEnd' and 'timeLog' use 'log'.
     // 'trace' uses 'error'.
     // 'assert' uses 'warn'.
@@ -107,21 +141,38 @@ function progress(label, fn)
         console[propertyName] =
         (...args) =>
         {
-            bar.hide();
+            deleteBars();
             value(...args);
-            bar.render();
+            writeBars();
         };
     }
-    const bar = new ProgressBar(`${label} \0bar \0percent / \0eta left`);
-    bar.render();
     try
     {
-        const result = fn(bar);
+        let result;
+        if (label === undefined)
+        {
+            const indicator =
+            {
+                newBar(label)
+                {
+                    const bar = new ProgressBar(label);
+                    return bar;
+                },
+            };
+            result = await fn(indicator);
+        }
+        else
+        {
+            const bar = new ProgressBar(label);
+            bar.render();
+            result = await fn(bar);
+        }
         return result;
     }
     finally
     {
-        bar.hide();
+        deleteBars();
+        bars.splice(0, Infinity);
         for (const [propertyName, value] of originalValues)
             console[propertyName] = value;
     }