return back previous transpilationWorkerThreads flow

Author: KSDaemon

packages/cubejs-backend-native/src/transpilers.rs

@@ -39,8 +39,6 @@ pub fn transpile_js(mut cx: FunctionContext) -> JsResult<JsPromise> {
     let deserializer = JsValueDeserializer::new(&mut cx, transform_data_js_object);
     let transform_request_config = TransformRequestConfig::deserialize(deserializer);
 
-    println!("\ntransform_config {:?}\n", transform_config);
-
     let promise = cx
         .task(move || {
             let transform_config: TransformConfig = match transform_request_config {

packages/cubejs-backend-shared/src/env.ts

@@ -229,6 +229,10 @@ const variables: Record<string, (...args: any) => any> = {
   transpilationWorkerThreadsCount: () => get('CUBEJS_TRANSPILATION_WORKER_THREADS_COUNT')
     .default('0')
     .asInt(),
+  // This one takes precedence over CUBEJS_TRANSPILATION_WORKER_THREADS
+  transpilationNative: () => get('CUBEJS_TRANSPILATION_NATIVE')
+    .default('false')
+    .asBoolStrict(),
 
   /** ****************************************************************
    * Common db options                                               *

packages/cubejs-schema-compiler/package.json

@@ -54,7 +54,8 @@
     "node-dijkstra": "^2.5.0",
     "ramda": "^0.27.2",
     "syntax-error": "^1.3.0",
-    "uuid": "^8.3.2"
+    "uuid": "^8.3.2",
+    "workerpool": "^9.2.0"
   },
   "devDependencies": {
     "@clickhouse/client": "^1.7.0",

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

@@ -6,6 +6,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 { transpileJs } from '@cubejs-backend/native';
@@ -94,16 +95,32 @@ export class DataSchemaCompiler {
     const errorsReport = new ErrorReporter(null, [], this.errorReport);
     this.errorsReport = errorsReport;
 
+    const transpilationWorkerThreads = getEnv('transpilationWorkerThreads');
+    const transpilationNative = getEnv('transpilationNative');
+
+    if (!transpilationNative && transpilationWorkerThreads) {
+      const wc = getEnv('transpilationWorkerThreadsCount');
+      this.workerPool = workerpool.pool(
+        path.join(__dirname, 'transpilers/transpiler_worker'),
+        wc > 0 ? { maxWorkers: wc } : undefined,
+      );
+    }
+
     const transpile = async () => {
       let cubeNames;
       let cubeSymbols;
       let transpilerNames;
+      let results;
 
-      if (getEnv('transpilationWorkerThreads')) {
+      if (transpilationNative || transpilationWorkerThreads) {
         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.
+        // Communication with native backend also involves deserialization.
         cubeSymbols = Object.fromEntries(
           Object.entries(this.cubeSymbols.symbols)
             .map(
@@ -112,7 +129,9 @@ export class DataSchemaCompiler {
               )],
             ),
         );
+      }
 
+      if (transpilationNative) {
         // Transpilers are the same for all files within phase.
         transpilerNames = this.transpilers.map(t => t.constructor.name);
 
@@ -123,9 +142,14 @@ export class DataSchemaCompiler {
         };
 
         await this.transpileJsFile(dummyFile, errorsReport, { cubeNames, cubeSymbols, transpilerNames, contextSymbols: CONTEXT_SYMBOLS });
+
+        results = await Promise.all(toCompile.map(f => this.transpileFile(f, errorsReport, { transpilerNames })));
+      } else if (transpilationWorkerThreads) {
+        results = await Promise.all(toCompile.map(f => this.transpileFile(f, errorsReport, { cubeNames, cubeSymbols })));
+      } else {
+        results = await Promise.all(toCompile.map(f => this.transpileFile(f, errorsReport, {})));
       }
 
-      const results = await Promise.all(toCompile.map(f => this.transpileFile(f, errorsReport, { transpilerNames })));
       return results.filter(f => !!f);
     };
 
@@ -141,14 +165,16 @@ export class DataSchemaCompiler {
         contextCompilers: this.contextCompilers,
       }))
       .then(() => {
-        if (getEnv('transpilationWorkerThreads')) {
+        if (transpilationNative) {
           // Clean up cache
           const dummyFile = {
             fileName: 'terminate.js',
             content: ';',
           };
 
           return this.transpileJsFile(dummyFile, errorsReport, { cubeNames: [], cubeSymbols: {}, transpilerNames: [], contextSymbols: {} });
+        } else if (transpilationWorkerThreads && this.workerPool) {
+          this.workerPool.terminate();
         }
 
         return Promise.resolve();
@@ -195,7 +221,7 @@ export class DataSchemaCompiler {
 
   async transpileJsFile(file, errorsReport, { cubeNames, cubeSymbols, contextSymbols, transpilerNames }) {
     try {
-      if (getEnv('transpilationWorkerThreads')) {
+      if (getEnv('transpilationNative')) {
         const reqData = {
           fileName: file.fileName,
           transpilers: transpilerNames,
@@ -215,6 +241,20 @@ export class DataSchemaCompiler {
         errorsReport.exitFile();
 
         return Object.assign({}, file, { content: res.code });
+      } else if (getEnv('transpilationWorkerThreads')) {
+        const data = {
+          fileName: file.fileName,
+          content: file.content,
+          transpilers: this.transpilers.map(t => t.constructor.name),
+          cubeNames,
+          cubeSymbols,
+        };
+
+        const res = await this.workerPool.exec('transpile', [data]);
+        errorsReport.addErrors(res.errors);
+        errorsReport.addWarnings(res.warnings);
+
+        return Object.assign({}, file, { content: res.content });
       } else {
         const ast = parse(
           file.content,

packages/cubejs-schema-compiler/src/compiler/ErrorReporter.ts

@@ -108,7 +108,7 @@ export class ErrorReporter {
   }
 
   public error(e: any, fileName?: any, lineNumber?: any, position?: any) {
-    const message = `${this.context.length ? `${this.context.join(' -> ')}: ` : ''}${e instanceof UserError ? e.message : (e.stack || e)}`;
+    const message = `${this.context.length ? `${this.context.join(' -> ')}: ` : ''}${e.message ? e.message : (e.stack || e)}`;
     if (this.rootReporter().errors.find(m => (m.message || m) === message)) {
       return;
     }

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

@@ -16,7 +16,7 @@ type TransferContent = {
   content: string;
   transpilers: string[];
   cubeNames: string[];
-  cubeSymbolsNames: Record<string, Record<string, boolean>>;
+  cubeSymbols: Record<string, Record<string, boolean>>;
 };
 
 const cubeDictionary = new LightweightNodeCubeDictionary();
@@ -32,7 +32,7 @@ const transpilers = {
 
 const transpile = (data: TransferContent) => {
   cubeDictionary.setCubeNames(data.cubeNames);
-  cubeSymbols.setSymbols(data.cubeSymbolsNames);
+  cubeSymbols.setSymbols(data.cubeSymbols);
 
   const ast = parse(
     data.content,
@@ -43,15 +43,15 @@ const transpile = (data: TransferContent) => {
     },
   );
 
+  errorsReport.inFile(data);
   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`);
     }
   });
+  errorsReport.exitFile();
 
   const content = babelGenerator(ast, {}, data.content).code;
 

yarn.lock

@@ -29554,6 +29554,11 @@ wordwrap@^1.0.0:
   resolved "https://registry.yarnpkg.com/wordwrap/-/wordwrap-1.0.0.tgz#27584810891456a4171c8d0226441ade90cbcaeb"
   integrity sha1-J1hIEIkUVqQXHI0CJkQa3pDLyus=
 
+workerpool@^9.2.0:
+  version "9.2.0"
+  resolved "https://registry.yarnpkg.com/workerpool/-/workerpool-9.2.0.tgz#f74427cbb61234708332ed8ab9cbf56dcb1c4371"
+  integrity sha512-PKZqBOCo6CYkVOwAxWxQaSF2Fvb5Iv2fCeTP7buyWI2GiynWr46NcXSgK/idoV6e60dgCBfgYc+Un3HMvmqP8w==
+
 "wrap-ansi-cjs@npm:wrap-ansi@^7.0.0":
   version "7.0.0"
   resolved "https://registry.yarnpkg.com/wrap-ansi/-/wrap-ansi-7.0.0.tgz#67e145cff510a6a6984bdf1152911d69d2eb9e43"