Merge pull request #126 from webdoc-labs/feature/multithreading

Feature: Parallel file indexing in @webdoc/parser

Author: ShukantPal

.idea/misc.xml

@@ -25,7 +25,7 @@
     "unit-test": "",
     "build": "webdoc --tutorials ./tutorials --site-root example-documentation --site-domain https://webdoc-labs.github.io --verbose",
     "build-next": "cd .. && webdoc && cd example",
-    "build-pixi-api": "cd ../../pixi-api && webdoc && cd ../webdoc/example",
+    "build-pixi-api": "cd ../../pixi-api && webdoc --verbose && cd ../webdoc/example",
     "build-pixi-api-prod": "cd ../../pixi-api && webdoc --site-root pixi-api && cd ../webdoc/example",
     "build-pixi-api-gcp": "cd ../../pixi-api && webdoc --tutorials ./projects/guides --verbose && cd ../webdoc/example",
     "build-pixi-guides": "cd ../../guides && webdoc --tutorials docs && cd ../webdoc/example",

example/package.json

@@ -19,6 +19,7 @@ generate the `sitemap.xml`. This is useful if you want to integrate with [Algoli
 * `--site-root <path>`: If using absolute links in a template, this will set the basepath. The basepath should the directory in which the documentation is being stored relative to where the server is running. The site root is "/" by default - which means that you'll need to serve the documentation directory as top-level. Note that @webdoc/default-template uses absolute links.
 * `-c <config-path>`: This sets the path of the configuration file webdoc uses.
 * `-u <tutorials-directory>` -  (optional) This should point to a directory containing tutorials written in Markdown (".md") or HTML ".html, ".htm". JSON files can be used to configure the hierarchy and naming of tutorials (see the Tutorial Configuration section).
+* `--no-workers` - Disables usage of worker threads to parallelize parsing.
 
 ### Configuration
 

packages/webdoc-cli/README.md

@@ -63,7 +63,6 @@ async function main(argv: yargs.Argv) {
   const config = loadConfig(argv.config);
   const tutorials = loadTutorials(argv.tutorials, config.template.routes.tutorials);
 
-
   if (argv.siteRoot) {
     config.template.siteRoot = argv.siteRoot;
   }
@@ -107,7 +106,9 @@ async function main(argv: yargs.Argv) {
   }
 
   try {
-    parse(sourceFiles, documentTree);
+    await parse(sourceFiles, documentTree, {
+      mainThread: !argv.workers,
+    });
   } catch (e) {
     // Make sure we get that API structure out so the user can debug the problem!
     if (config.opts && config.opts.export) {
@@ -163,7 +164,9 @@ async function main(argv: yargs.Argv) {
 const argv = yargs.scriptName("@webdoc/cli")
   .usage("$0 -c <configFile> -u <tutorialDir> --verbose " +
     "--site-root <siteRoot> " +
-    "--site-domain <siteDomain>")
+    "--site-domain <siteDomain>" +
+    "--no-workers")
+  .default("workers", true)
   .default("config", path.join(process.cwd(), "webdoc.conf.json"), "webdoc config file")
   .alias("c", "config")
   .alias("u", "tutorials")

packages/webdoc-cli/src/index.js

@@ -5,8 +5,8 @@ const parser = require("@webdoc/parser");
 const expect = require("chai").expect;
 
 describe("@webdoc/externalize (read-write test)", function() {
-  it("should deserialize a serialized documented interface properly", function() {
-    const inputTree = parser.parse(`
+  it("should deserialize a serialized documented interface properly", async function() {
+    const inputTree = await parser.parse(`
       /** Symbol 0 */
       class Symbol0 {
         /** Symbol 1 */

packages/webdoc-externalize/test/read-write.js

@@ -1,13 +1,9 @@
 declare module "@webdoc/parser" {
-  import type {Doc} from "@webdoc/types";
+  import type {RootDoc} from "@webdoc/types";
 
-  /**
-   * The parser will accept:
-   * + a file's contents
-   * + an array of file contents
-   + + a map mapping file-names to file contents
-   */
-  declare type ParserInput = string | string[] | Map<string, string>;
-
-  declare function parse(input: ParserInput): Doc;
+  declare function parse(
+    target: string | SourceFile[],
+    root?: RootDoc,
+    options?: $Shape<{ mainThread: boolean }>,
+  ): Promise<RootDoc>;
 }

packages/webdoc-parser/index.flow.js

@@ -35,6 +35,7 @@ export function initLogger(defaultLevel: string = "INFO") {
       PartialParser: defaultLevel,
       DocParser: defaultLevel,
       DocumentTreeModifier: defaultLevel,
+      Indexer: defaultLevel,
     },
     (level, tag, msg, params) => {
       let tagPrefix = `[${tag}]:`;

packages/webdoc-parser/src/Logger.js

@@ -0,0 +1,126 @@
+// @flow
+
+import {type PackageDoc, type SourceFile} from "@webdoc/types";
+import {parserLogger, tag} from "../Logger";
+import EventEmitter from "events";
+import type {LanguageConfig} from "../types/LanguageIntegration";
+import {type Symbol} from "../types/Symbol";
+// $FlowFixMe
+import {Worker} from "worker_threads";
+import os from "os";
+import path from "path";
+
+export type JobData = {
+  jobId: number,
+  jobLanguageIntegrationModule: string,
+  jobFile: SourceFile,
+  jobConfig: LanguageConfig,
+};
+
+export type JobResult = {
+  jobId: number,
+  jobResult?: Symbol,
+  jobError?: any,
+};
+
+class IndexerWorker extends EventEmitter {
+  onError: (ev: "error", ...args: any[]) => void;
+  onMessage: (...args: any[]) => void;
+  worker: Worker;
+
+  constructor(worker: Worker) {
+    super();
+
+    this.onError = this.emit.bind(this, "error");
+    this.onMessage = this.onMessageImpl.bind(this);
+
+    this.worker = worker;
+    this.worker.on("error", this.onError);
+    this.worker.on("message", this.onMessage);
+  }
+
+  send(
+    jobId: number,
+    jobLanguageIntegrationModule: string,
+    jobFile: SourceFile,
+    jobConfig: LanguageConfig,
+  ): number {
+    this.worker.postMessage(({
+      jobId,
+      jobLanguageIntegrationModule,
+      jobFile,
+      jobConfig,
+    }: JobData));
+
+    return jobId;
+  }
+
+  onMessageImpl({jobId, jobResult, jobError}: JobResult) {
+    this.emit(`response-${jobId}`, jobResult, jobError);
+  }
+
+  destroy() {
+    this.worker.terminate();
+  }
+}
+
+export class IndexerWorkerPool {
+  workers: IndexerWorker[];
+  ptr: number = 0;
+
+  constructor(limit?: number) {
+    const workerPoolSize = Math.min(os.cpus().length, limit || 4);
+    const workers = new Array<Worker>(workerPoolSize);
+    const workerPath = path.resolve(__dirname, "./worker.js");
+
+    parserLogger.info(tag.Indexer, "Using " + workerPoolSize + " worker threads for indexing");
+
+    for (let i = 0; i < workerPoolSize; i++) {
+      workers[i] = new IndexerWorker(new Worker(workerPath));
+    }
+
+    this.workers = workers;
+  }
+
+  repair(symbol: Symbol, packages: { [id: string]: PackageDoc }): Symbol {
+    const file = symbol.loc.file;
+    const pkgId = file && file.package.id;
+
+    if (file && pkgId && packages[pkgId]) {
+      file.package = packages[pkgId];
+    }
+    for (const child of symbol.members) {
+      this.repair(child, packages);
+    }
+
+    return symbol;
+  }
+
+  index(
+    langIntegrationModule: string,
+    file: SourceFile,
+    config: LanguageConfig,
+    packages: { [id: string]: PackageDoc },
+  ): Promise<Symbol> {
+    return new Promise<Symbol>((resolve, reject) => {
+      const jobId = this.ptr++;
+      const worker = this.workers[jobId % this.workers.length];
+
+      worker.send(jobId, langIntegrationModule, file, config);
+      worker.on(`response-${jobId}`, (jobResult: Symbol, jobError: any): void => {
+        if (jobResult) {
+          resolve(this.repair(jobResult, packages));
+        } else {
+          reject(jobError);
+        }
+      });
+    });
+  }
+
+  destroy() {
+    for (const worker of this.workers) {
+      worker.destroy();
+    }
+    this.workers.length = 0;
+  }
+}

packages/webdoc-parser/src/indexer/IndexerWorkerPool.js

@@ -0,0 +1,112 @@
+// @flow
+
+import type {LanguageConfig, LanguageIntegration} from "../types/LanguageIntegration";
+import {parserLogger, tag} from "../Logger";
+import {IndexerWorkerPool} from "./IndexerWorkerPool";
+import {type SourceFile} from "@webdoc/types";
+import {type Symbol} from "../types/Symbol";
+import _ from "lodash";
+import fs from "fs";
+import os from "os";
+import path from "path";
+
+declare var globalThis: any;
+
+// File-extension -> LanguageIntegration mapping
+const languages: { [id: string]: LanguageIntegration } = {};
+
+// Register a language-integration that will be used to generate a symbol-tree for files with its
+// file-extensions.
+export function register(lang: LanguageIntegration): void {
+  for (const ext of lang.extensions) {
+    if (languages[ext]) {
+      parserLogger.warn("LanguageIntegration",
+        `.${ext} file extension has already been registered`);
+    }
+
+    languages[ext] = lang;
+  }
+}
+
+export async function run(
+  files: SourceFile[],
+  config: LanguageConfig,
+  options?: $Shape<{
+    mainThread: boolean,
+  }> = {},
+): Promise<Symbol[]> {
+  const startTime = Date.now();
+  const maxThreads = Math.min(os.cpus().length, 1 + Math.floor(files.length / 125));
+
+  parserLogger.info(tag.Indexer, "Indexing " + files.length + " files");
+
+  const symbolTrees: Array<Symbol> = new Array(files.length);
+
+  if (maxThreads > 1 && !options.mainThread) {
+    const packages = _.keyBy(
+      files.map((file) => file.package),
+      (pkg) => pkg.id,
+    );
+    const symbolIndexingOperations: Array<Promise<void>> = new Array(files.length);
+    const workerPool = new IndexerWorkerPool(maxThreads);
+
+    for (let i = 0; i < files.length; i++) {
+      const fileName = files[i].path;
+      const extension = fileName.substring(fileName.lastIndexOf(".") + 1, fileName.length);
+
+      if (!(extension in languages)) {
+        throw new Error(`.${extension} language is not registered with the Indexer!`);
+      }
+
+      const file = {
+        ...files[i],
+        path: path.resolve(globalThis.process.cwd(), files[i].path),
+      };
+      const lang = languages[extension].module;
+
+      symbolIndexingOperations[i] = workerPool.index(lang, file, config, packages).then(
+        (symbolTree: Symbol): void => {
+          symbolTrees[i] = symbolTree;
+        },
+      );
+    }
+
+    await Promise.all(symbolIndexingOperations);
+    workerPool.destroy();
+  } else {
+    for (let i = 0; i < files.length; i++) {
+      const filePath = path.resolve(globalThis.process.cwd(), files[i].path);
+      const fileContent = files[i].content || fs.readFileSync(filePath, "utf8");
+
+      symbolTrees[i] = process(fileContent, files[i], config);
+    }
+  }
+
+  const endTime = Date.now();
+
+  parserLogger.info(tag.Indexer, "Indexing took " + (endTime - startTime) + "ms");
+
+  return symbolTrees;
+}
+
+export function process(
+  file: string,
+  source: SourceFile,
+  config: LanguageConfig,
+): Symbol {
+  const fileName = source.path;
+  const lang = languages[fileName.substring(fileName.lastIndexOf(".") + 1, fileName.length)];
+
+  if (!lang) {
+    throw new Error(`.${lang} file language is not registered`);
+  }
+
+  return lang.parse(file, source, config);
+}
+
+export function lang(module: string): LanguageIntegration {
+  module = module.replace("@webdoc/parser", "../../");
+
+  // $FlowFixMe
+  return ((require(module): any).default: LanguageIntegration);
+}

packages/webdoc-parser/src/indexer/index.js

@@ -0,0 +1,40 @@
+// @flow
+
+import * as Indexer from "./";
+import type {JobData, JobResult} from "./IndexerWorkerPool";
+// $FlowFixMe
+import {isMainThread, parentPort} from "worker_threads";
+import fs from "fs";
+
+function onMessage(data: JobData) {
+  function onFileRead(err: ?Error, contents: string) {
+    if (err) {
+      return parentPort.postMessage(({
+        jobId: data.jobId,
+        jobError: err,
+      }: JobResult));
+    }
+
+    const lang = Indexer.lang(data.jobLanguageIntegrationModule);
+    const symbolTree = lang.parse(
+      contents,
+      data.jobFile,
+      data.jobConfig,
+    );
+
+    parentPort.postMessage(({
+      jobId: data.jobId,
+      jobResult: symbolTree,
+    }: JobResult));
+  }
+
+  if (!data.jobFile.content) {
+    fs.readFile(data.jobFile.path, "utf8", onFileRead);
+  } else {
+    onFileRead(null, data.jobFile.content);
+  }
+}
+
+if (!isMainThread) {
+  parentPort.on("message", onMessage);
+}