use workerpool pkg

Author: KSDaemon

packages/cubejs-schema-compiler/src/compiler/DataSchemaCompiler.js

@@ -7,6 +7,7 @@ import { parse } from '@babel/parser';
 import babelGenerator from '@babel/generator';
 import babelTraverse from '@babel/traverse';
 import R from 'ramda';
+import workerpool from 'workerpool';
 
 import { getEnv, isNativeSupported } from '@cubejs-backend/shared';
 import { UserError } from './UserError';
@@ -43,6 +44,7 @@ export class DataSchemaCompiler {
     this.yamlCompiler = options.yamlCompiler;
     this.yamlCompiler.dataSchemaCompiler = this;
     this.pythonContext = null;
+    this.workerPool = null;
   }
 
   compileObjects(compileServices, objects, errorsReport) {
@@ -92,13 +94,19 @@ export class DataSchemaCompiler {
     const errorsReport = new ErrorReporter(null, [], this.errorReport);
     this.errorsReport = errorsReport;
 
+    if (getEnv('workerThreadsTranspilation')) {
+      this.workerPool = workerpool.pool(path.join(__dirname, 'transpilers/transpiler_worker'));
+    }
+
     const transpile = async () => {
       let cubeNames;
       let cubeSymbolsNames;
 
       if (getEnv('workerThreadsTranspilation')) {
         cubeNames = Object.keys(this.cubeDictionary.byId);
         // We need only cubes and all its member names for transpiling.
+        // Cubes doesn't change during transpiling, but are changed during compilation phase,
+        // so we can prepare them once for every phase.
         // Communication between main and worker threads uses
         // The structured clone algorithm (@see https://developer.mozilla.org/en-US/docs/Web/API/Web_Workers_API/Structured_clone_algorithm)
         // which doesn't allow passing any function objects, so we need to sanitize the symbols.
@@ -133,7 +141,12 @@ export class DataSchemaCompiler {
       .then(() => compilePhase({
         cubeCompilers: this.cubeCompilers,
         contextCompilers: this.contextCompilers,
-      }));
+      }))
+      .then(() => {
+        if (this.workerPool) {
+          this.workerPool.terminate();
+        }
+      });
   }
 
   compile() {
@@ -174,34 +187,18 @@ export class DataSchemaCompiler {
     }
   }
 
-  async transpileJsFile(file, errorsReport, options) {
+  async transpileJsFile(file, errorsReport, { cubeNames, cubeSymbolsNames }) {
     try {
       if (getEnv('workerThreadsTranspilation')) {
-        const res = await new Promise((resolve, reject) => {
-          const { cubeNames, cubeSymbolsNames } = options;
-
-          const worker = new Worker(
-            path.join(__dirname, 'transpilers/transpiler_worker'),
-            { workerData: { cubeNames, cubeSymbolsNames } },
-          );
-
-          const data = {
-            fileName: file.fileName,
-            content: file.content,
-            transpilers: this.transpilers.map(t => t.constructor.name),
-          };
-
-          worker.postMessage(data);
-
-          worker.on('message', resolve);
-          worker.on('error', reject);
-          worker.on('exit', (code) => {
-            if (code !== 0) {
-              reject(new Error(`Worker thread exited with code ${code}`));
-            }
-          });
-        });
-
+        const data = {
+          fileName: file.fileName,
+          content: file.content,
+          transpilers: this.transpilers.map(t => t.constructor.name),
+          cubeNames,
+          cubeSymbolsNames,
+        };
+
+        const res = await this.workerPool.exec('transpile', [data]);
         errorsReport.addErrors(res.errors);
         errorsReport.addWarnings(res.warnings);
 

packages/cubejs-schema-compiler/src/compiler/transpilers/LightweightNodeCubeDictionary.ts

@@ -1,10 +1,14 @@
 import { TranspilerCubeResolver } from './transpiler.interface';
 
 export class LightweightNodeCubeDictionary implements TranspilerCubeResolver {
-  public constructor(private readonly cubeNames: string[] = []) {
+  public constructor(private cubeNames: string[] = []) {
   }
 
   public resolveCube(name: string): boolean {
     return this.cubeNames.includes(name);
   }
+
+  public setCubeNames(cubeNames: string[]): void {
+    this.cubeNames = cubeNames;
+  }
 }

packages/cubejs-schema-compiler/src/compiler/transpilers/LightweightSymbolResolver.ts

@@ -1,8 +1,14 @@
 import { TranspilerSymbolResolver } from './transpiler.interface';
 import { CONTEXT_SYMBOLS, CURRENT_CUBE_CONSTANTS } from '../CubeSymbols';
 
+type CubeSymbols = Record<string, Record<string, boolean>>;
+
 export class LightweightSymbolResolver implements TranspilerSymbolResolver {
-  public constructor(private readonly symbols: any) {
+  public constructor(private symbols: CubeSymbols = {}) {
+  }
+
+  public setSymbols(symbols: CubeSymbols) {
+    this.symbols = symbols;
   }
 
   public isCurrentCube(name): boolean {

packages/cubejs-schema-compiler/src/compiler/transpilers/transpiler_worker.ts

@@ -1,5 +1,4 @@
-import { parentPort, workerData } from 'worker_threads';
-
+import workerpool from 'workerpool';
 import { parse } from '@babel/parser';
 import babelGenerator from '@babel/generator';
 import babelTraverse from '@babel/traverse';
@@ -12,14 +11,16 @@ import { ErrorReporter } from '../ErrorReporter';
 import { LightweightSymbolResolver } from './LightweightSymbolResolver';
 import { LightweightNodeCubeDictionary } from './LightweightNodeCubeDictionary';
 
-type FileContent = {
+type TransferContent = {
   fileName: string;
   content: string;
   transpilers: string[];
+  cubeNames: string[];
+  cubeSymbolsNames: Record<string, Record<string, boolean>>;
 };
 
-const cubeDictionary = new LightweightNodeCubeDictionary(workerData.cubeNames);
-const cubeSymbols = new LightweightSymbolResolver(workerData.cubeSymbolsNames);
+const cubeDictionary = new LightweightNodeCubeDictionary();
+const cubeSymbols = new LightweightSymbolResolver();
 const errorsReport = new ErrorReporter(null, []);
 
 const transpilers = {
@@ -29,34 +30,38 @@ const transpilers = {
   CubePropContextTranspiler: new CubePropContextTranspiler(cubeSymbols, cubeDictionary, cubeSymbols),
 };
 
-if (parentPort) {
-  parentPort.on('message', (file: FileContent) => {
-    const ast = parse(
-      file.content,
-      {
-        sourceFilename: file.fileName,
-        sourceType: 'module',
-        plugins: ['objectRestSpread']
-      },
-    );
-
-    file.transpilers.forEach(transpilerName => {
-      if (transpilers[transpilerName]) {
-        errorsReport.inFile(file);
-        babelTraverse(ast, transpilers[transpilerName].traverseObject(errorsReport));
-        errorsReport.exitFile();
-      } else {
-        throw new Error(`Transpiler ${transpilerName} not supported`);
-      }
-    });
-
-    const content = babelGenerator(ast, {}, file.content).code;
-
-    // @ts-ignore
-    parentPort.postMessage({
-      content,
-      errors: errorsReport.getErrors(),
-      warnings: errorsReport.getWarnings()
-    });
+const transpile = (data: TransferContent) => {
+  cubeDictionary.setCubeNames(data.cubeNames);
+  cubeSymbols.setSymbols(data.cubeSymbolsNames);
+
+  const ast = parse(
+    data.content,
+    {
+      sourceFilename: data.fileName,
+      sourceType: 'module',
+      plugins: ['objectRestSpread']
+    },
+  );
+
+  data.transpilers.forEach(transpilerName => {
+    if (transpilers[transpilerName]) {
+      errorsReport.inFile(data);
+      babelTraverse(ast, transpilers[transpilerName].traverseObject(errorsReport));
+      errorsReport.exitFile();
+    } else {
+      throw new Error(`Transpiler ${transpilerName} not supported`);
+    }
   });
-}
+
+  const content = babelGenerator(ast, {}, data.content).code;
+
+  return {
+    content,
+    errors: errorsReport.getErrors(),
+    warnings: errorsReport.getWarnings()
+  };
+};
+
+workerpool.worker({
+  transpile,
+});