Merge pull request #438 from wtsi-hgi/feature-js-console-output

mr-c · web-flow · commit 77a352842ce6 · 2017-09-08T12:04:01.000+01:00
Adds --js-console option to facilitate debugging
diff --git a/cwltool/builder.py b/cwltool/builder.py
@@ -51,6 +51,7 @@ def __init__(self):  # type: () -> None
         self.stagedir = None  # type: Text
         self.make_fs_access = None  # type: Type[StdFsAccess]
         self.debug = False  # type: bool
+        self.js_console = False  # type: bool
         self.mutation_manager = None  # type: MutationManager
         self.force_docker_pull = False  # type: bool
 
@@ -256,5 +257,6 @@ def do_eval(self, ex, context=None, pull_image=True, recursive=False):
                                   self.resources,
                                   context=context, pull_image=pull_image,
                                   timeout=self.timeout,
-                                  force_docker_pull=self.force_docker_pull,
-                                  debug=self.debug)
+                                  debug=self.debug,
+                                  js_console=self.js_console,
+                                  force_docker_pull=self.force_docker_pull)
diff --git a/cwltool/cwlNodeEngine.js b/cwltool/cwlNodeEngine.js
@@ -5,9 +5,17 @@ process.stdin.on("data", function(chunk) {
   incoming += chunk;
   var i = incoming.indexOf("\n");
   if (i > -1) {
-    var fn = JSON.parse(incoming.substr(0, i));
-    incoming = incoming.substr(i+1);
-    process.stdout.write(JSON.stringify(require("vm").runInNewContext(fn, {})) + "\n");
+    try{
+      var fn = JSON.parse(incoming.substr(0, i));
+      incoming = incoming.substr(i+1);
+      process.stdout.write(JSON.stringify(require("vm").runInNewContext(fn, {})) + "\n");
+    }
+    catch(e){
+      console.error(e)
+    }
+    /*strings to indicate the process has finished*/
+    console.log("r1cepzbhUTxtykz5XTC4");
+    console.error("r1cepzbhUTxtykz5XTC4");
   }
 });
 process.stdin.on("end", process.exit);
diff --git a/cwltool/cwlNodeEngineJSConsole.js b/cwltool/cwlNodeEngineJSConsole.js
@@ -0,0 +1,32 @@
+"use strict";
+function js_console_log(){
+    console.error("[log] "+require("util").format.apply(this, arguments).split("\n").join("\n[log] "));
+}
+function js_console_err(){
+    console.error("[err] "+require("util").format.apply(this, arguments).split("\n").join("\n[err] "));
+}
+process.stdin.setEncoding("utf8");
+var incoming = "";
+process.stdin.on("data", function(chunk) {
+  incoming += chunk;
+  var i = incoming.indexOf("\n");
+  if (i > -1) {
+      try{
+        var fn = JSON.parse(incoming.substr(0, i));
+        incoming = incoming.substr(i+1);
+        process.stdout.write(JSON.stringify(require("vm").runInNewContext(fn, {
+        console: {
+            log: js_console_log, 
+            error: js_console_err
+        }
+        })) + "\n");
+    }
+    catch(e){
+        console.error(e)
+    }
+    /*strings to indicate the process has finished*/
+    console.log("r1cepzbhUTxtykz5XTC4");
+    console.error("r1cepzbhUTxtykz5XTC4");
+  }
+});
+process.stdin.on("end", process.exit);
diff --git a/cwltool/expression.py b/cwltool/expression.py
@@ -153,8 +153,8 @@ def next_seg(remain, obj):  # type: (Text, Any) -> Any
         return obj
 
 
-def evaluator(ex, jslib, obj, fullJS=False, timeout=None, force_docker_pull=False, debug=False):
-    # type: (Text, Text, Dict[Text, Any], bool, int, bool, bool) -> JSON
+def evaluator(ex, jslib, obj, fullJS=False, timeout=None, force_docker_pull=False, debug=False, js_console=False):
+    # type: (Text, Text, Dict[Text, Any], bool, int, bool, bool, bool) -> JSON
     m = param_re.match(ex)
     if m:
         if m.end(1)+1 == len(ex) and m.group(1) == "null":
@@ -164,17 +164,17 @@ def evaluator(ex, jslib, obj, fullJS=False, timeout=None, force_docker_pull=Fals
         except Exception as w:
             raise WorkflowException("%s%s" % (m.group(1), w))
     elif fullJS:
-        return sandboxjs.execjs(ex, jslib, timeout=timeout, force_docker_pull=force_docker_pull, debug=debug)
+        return sandboxjs.execjs(ex, jslib, timeout=timeout, force_docker_pull=force_docker_pull, debug=debug, js_console=js_console)
     else:
         raise sandboxjs.JavascriptException(
             "Syntax error in parameter reference '%s' or used Javascript code without specifying InlineJavascriptRequirement.",
             ex)
 
 
 def interpolate(scan, rootvars,
-                timeout=None, fullJS=None, jslib="",force_docker_pull=False,
-                debug=False):
-    # type: (Text, Dict[Text, Any], int, bool, Union[str, Text], bool, bool) -> JSON
+                timeout=None, fullJS=None, jslib="", force_docker_pull=False,
+                debug=False, js_console=False):
+    # type: (Text, Dict[Text, Any], int, bool, Union[str, Text], bool, bool, bool) -> JSON
     scan = scan.strip()
     parts = []
     w = scanner(scan)
@@ -184,7 +184,7 @@ def interpolate(scan, rootvars,
         if scan[w[0]] == '$':
             e = evaluator(scan[w[0] + 1:w[1]], jslib, rootvars, fullJS=fullJS,
                           timeout=timeout, force_docker_pull=force_docker_pull,
-                          debug=debug)
+                          debug=debug, js_console=js_console)
             if w[0] == 0 and w[1] == len(scan):
                 return e
             leaf = json.dumps(e, sort_keys=True)
@@ -202,8 +202,8 @@ def interpolate(scan, rootvars,
 
 
 def do_eval(ex, jobinput, requirements, outdir, tmpdir, resources,
-            context=None, pull_image=True, timeout=None, force_docker_pull=False, debug=False):
-    # type: (Union[dict, AnyStr], Dict[Text, Union[Dict, List, Text]], List[Dict[Text, Any]], Text, Text, Dict[Text, Union[int, Text]], Any, bool, int, bool, bool) -> Any
+            context=None, pull_image=True, timeout=None, force_docker_pull=False, debug=False, js_console=False):
+    # type: (Union[dict, AnyStr], Dict[Text, Union[Dict, List, Text]], List[Dict[Text, Any]], Text, Text, Dict[Text, Union[int, Text]], Any, bool, int, bool, bool, bool) -> Any
 
     runtime = copy.copy(resources)
     runtime["tmpdir"] = docker_windows_path_adjust(tmpdir)
@@ -230,7 +230,9 @@ def do_eval(ex, jobinput, requirements, outdir, tmpdir, resources,
                                fullJS=fullJS,
                                jslib=jslib,
                                force_docker_pull=force_docker_pull,
-                               debug=debug)
+                               debug=debug,
+                               js_console=js_console)
+
         except Exception as e:
             raise WorkflowException("Expression evaluation error:\n%s" % e)
     else:
diff --git a/cwltool/main.py b/cwltool/main.py
@@ -158,6 +158,8 @@ def arg_parser():  # type: () -> argparse.ArgumentParser
     exgroup.add_argument("--quiet", action="store_true", help="Only print warnings and errors.")
     exgroup.add_argument("--debug", action="store_true", help="Print even more logging")
 
+    parser.add_argument("--js-console", action="store_true", help="Enable javascript console output")
+
     dependency_resolvers_configuration_help = argparse.SUPPRESS
     dependencies_directory_help = argparse.SUPPRESS
     use_biocontainers_help = argparse.SUPPRESS
@@ -730,6 +732,7 @@ def main(argsl=None,  # type: List[str]
                      'cachedir': None,
                      'quiet': False,
                      'debug': False,
+                     'js_console': False,
                      'version': False,
                      'enable_dev': False,
                      'enable_ext': False,
diff --git a/cwltool/process.py b/cwltool/process.py
@@ -535,6 +535,8 @@ def _init_job(self, joborder, **kwargs):
         tmpdir: tmpdir on host for this job
         stagedir: stagedir on host for this job
         select_resources: callback to select compute resources
+        debug: enable debugging output
+        js_console: enable javascript console output
         """
 
         builder = Builder()
@@ -559,6 +561,7 @@ def _init_job(self, joborder, **kwargs):
         builder.resources = {}
         builder.timeout = kwargs.get("eval_timeout")
         builder.debug = kwargs.get("debug")
+        builder.js_console = kwargs.get("js_console")
         builder.mutation_manager = kwargs.get("mutation_manager")
 
         builder.make_fs_access = kwargs.get("make_fs_access") or StdFsAccess
diff --git a/cwltool/sandboxjs.py b/cwltool/sandboxjs.py
@@ -3,6 +3,7 @@
 import json
 import logging
 import os
+import re
 import select
 import subprocess
 import threading
@@ -53,10 +54,15 @@ def check_js_threshold_version(working_alias):
         return False
 
 
-def new_js_proc(force_docker_pull=False):
-    # type: (bool) -> subprocess.Popen
+def new_js_proc(force_docker_pull=False, js_console=False):
+    # type: (bool, bool) -> subprocess.Popen
+    
+    cwl_node_engine_js = 'cwlNodeEngine.js'
+    if js_console:
+        cwl_node_engine_js = 'cwlNodeEngineJSConsole.js'
+        _logger.warn("Running with support for javascript console in expressions (DO NOT USE IN PRODUCTION)")
 
-    res = resource_stream(__name__, 'cwlNodeEngine.js')
+    res = resource_stream(__name__, cwl_node_engine_js)
     nodecode = res.read().decode('utf-8')
 
     required_node_version, docker = (False,)*2
@@ -125,10 +131,10 @@ def new_js_proc(force_docker_pull=False):
     return nodejs
 
 
-def execjs(js, jslib, timeout=None, force_docker_pull=False, debug=False):  # type: (Union[Mapping, Text], Any, int, bool, bool) -> JSON
+def execjs(js, jslib, timeout=None, force_docker_pull=False, debug=False, js_console=False):  # type: (Union[Mapping, Text], Any, int, bool, bool, bool) -> JSON
 
     if not hasattr(localdata, "proc") or localdata.proc.poll() is not None or onWindows():
-        localdata.proc = new_js_proc(force_docker_pull=force_docker_pull)
+        localdata.proc = new_js_proc(force_docker_pull=force_docker_pull, js_console=js_console)
 
     nodejs = localdata.proc
 
@@ -158,6 +164,12 @@ def terminate():
     rselect = [nodejs.stdout, nodejs.stderr]  # type: List[BytesIO]
     wselect = [nodejs.stdin]  # type: List[BytesIO]
 
+    PROCESS_FINISHED_STR = "r1cepzbhUTxtykz5XTC4\n"
+
+    def process_finished(): # type: () -> bool
+        return stdout_buf.getvalue().decode().endswith(PROCESS_FINISHED_STR) and \
+            stderr_buf.getvalue().decode().endswith(PROCESS_FINISHED_STR)
+
     # On windows system standard input/output are not handled properly by select module
     # (modules like  pywin32, msvcrt, gevent don't work either)
     if sys.platform=='win32':
@@ -208,11 +220,9 @@ def get_error(error_queue):
         error_thread.daemon=True
         error_thread.start()
 
-        # mark if output/error is ready
-        output_ready=False
-        error_ready=False
+        finished = False
 
-        while (len(wselect) + len(rselect)) > 0:
+        while not finished and tm.is_alive():
             try:
                 if nodejs.stdin in wselect:
                     if not input_queue.empty():
@@ -221,58 +231,39 @@ def get_error(error_queue):
                         wselect = []
                 if nodejs.stdout in rselect:
                     if not output_queue.empty():
-                        output_ready = True
                         stdout_buf.write(output_queue.get())
-                    elif output_ready:
-                        rselect = []
-                        no_more_output.release()
-                        no_more_error.release()
-                        output_thread.join()
 
                 if nodejs.stderr in rselect:
                     if not error_queue.empty():
-                        error_ready = True
-                        stderr_buf.write(error_queue.get())
-                    elif error_ready:
-                        rselect = []
-                        no_more_output.release()
-                        no_more_error.release()
-                        output_thread.join()
-                        error_thread.join()
-                if stdout_buf.getvalue().endswith("\n".encode()):
-                    rselect = []
+                        stderr_buf.write(error_queue.get())                
+
+                if process_finished() and error_queue.empty() and output_queue.empty():
+                    finished = True
                     no_more_output.release()
                     no_more_error.release()
-                    output_thread.join()
             except OSError as e:
                 break
 
     else:
-        while (len(wselect) + len(rselect)) > 0:
+        while not process_finished() and tm.is_alive():
             rready, wready, _ = select.select(rselect, wselect, [])
             try:
                 if nodejs.stdin in wready:
                     b = stdin_buf.read(select.PIPE_BUF)
                     if b:
                         os.write(nodejs.stdin.fileno(), b)
-                    else:
-                        wselect = []
                 for pipes in ((nodejs.stdout, stdout_buf), (nodejs.stderr, stderr_buf)):
                     if pipes[0] in rready:
                         b = os.read(pipes[0].fileno(), select.PIPE_BUF)
                         if b:
                             pipes[1].write(b)
-                        else:
-                            rselect.remove(pipes[0])
-                if stdout_buf.getvalue().endswith("\n".encode()):
-                    rselect = []
             except OSError as e:
                 break
     tm.cancel()
 
     stdin_buf.close()
-    stdoutdata = stdout_buf.getvalue()
-    stderrdata = stderr_buf.getvalue()
+    stdoutdata = stdout_buf.getvalue()[:-len(PROCESS_FINISHED_STR) - 1]
+    stderrdata = stderr_buf.getvalue()[:-len(PROCESS_FINISHED_STR) - 1]
 
     def fn_linenum():  # type: () -> Text
         lines = fn.splitlines()
@@ -290,6 +281,13 @@ def stdfmt(data):  # type: (Text) -> Text
 
     nodejs.poll()
 
+    if js_console:
+        if len(stderrdata) > 0:
+            _logger.info("Javascript console output:")
+            _logger.info("----------------------------------------")
+            _logger.info('\n'.join(re.findall(r'^[[](?:log|err)[]].*$', stderrdata.decode('utf-8'), flags=re.MULTILINE)))
+            _logger.info("----------------------------------------")
+
     if debug:
         info = u"returncode was: %s\nscript was:\n%s\nstdout was: %s\nstderr was: %s\n" %\
                (nodejs.returncode, fn_linenum(), stdfmt(stdoutdata.decode('utf-8')), stdfmt(stderrdata.decode('utf-8')))
diff --git a/cwltool/workflow.py b/cwltool/workflow.py
@@ -337,6 +337,10 @@ def receive_output(self, step, outputparms, final_output_callback, jobout, proce
 
     def try_make_job(self, step, final_output_callback, **kwargs):
         # type: (WorkflowJobStep, Callable[[Any, Any], Any], **Any) -> Generator
+
+        js_console = kwargs.get("js_console", False)
+        debug = kwargs.get("debug", False)
+
         inputparms = step.tool["inputs"]
         outputparms = step.tool["outputs"]
 
@@ -375,7 +379,7 @@ def valueFromFunc(k, v):  # type: (Any, Any) -> Any
                     if k in valueFrom:
                         return expression.do_eval(
                             valueFrom[k], shortio, self.workflow.requirements,
-                            None, None, {}, context=v)
+                            None, None, {}, context=v, debug=debug, js_console=js_console)
                     else:
                         return v
 
diff --git a/setup.py b/setup.py
@@ -44,6 +44,7 @@
                                 'schemas/v1.1.0-dev1/salad/schema_salad/metaschema/*.yml',
                                 'schemas/v1.1.0-dev1/salad/schema_salad/metaschema/*.md',
                                 'cwlNodeEngine.js',
+                                'cwlNodeEngineJSConsole.js',
                                 'extensions.yml']},
       include_package_data=True,
       install_requires=[
diff --git a/tests/test_examples.py b/tests/test_examples.py
diff --git a/tests/wf/js_output.cwl b/tests/wf/js_output.cwl
diff --git a/tests/wf/js_output_workflow.cwl b/tests/wf/js_output_workflow.cwl
