<commit_before>from __future__ import absolute_import
import errno
import json
import logging
import os
import select
import subprocess
import threading
import Queue
import sys
from io import BytesIO
from typing import Any, Dict, List, Mapping, Text, Tuple, Union
from .utils import onWindows
from pkg_resources import resource_stream

import six

class JavascriptException(Exception):
    pass


_logger = logging.getLogger("cwltool")

JSON = Union[Dict[Text, Any], List[Any], Text, int, float, bool, None]

localdata = threading.local()

have_node_slim = False
# minimum acceptable version of nodejs engine
minimum_node_version_str = '0.10.26'

def check_js_threshold_version(working_alias):
    # type: (str) -> bool

    """Checks if the nodeJS engine version on the system
    with the allowed minimum version.
    https://github.com/nodejs/node/blob/master/CHANGELOG.md#nodejs-changelog
    """
    # parse nodejs version into int Tuple: 'v4.2.6\n' -> [4, 2, 6]
    current_version_str = subprocess.check_output(
        [working_alias, "-v"]).decode('ascii')

    current_version = [int(v) for v in current_version_str.strip().strip('v').split('.')]
    minimum_node_version = [int(v) for v in minimum_node_version_str.split('.')]

    if current_version >= minimum_node_version:
        return True
    else:
        return False


def new_js_proc():
    # type: () -> subprocess.Popen

    res = resource_stream(__name__, 'cwlNodeEngine.js')
    nodecode = res.read()

    required_node_version, docker = (False,)*2
    nodejs = None
    trynodes = ("nodejs", "node")
    for n in trynodes:
        try:
            if subprocess.check_output([n, "--eval", "process.stdout.write('t')"]) != "t":
                continue
            else:
                nodejs = subprocess.Popen([n, "--eval", nodecode],
                                          stdin=subprocess.PIPE,
                                          stdout=subprocess.PIPE,
                                          stderr=subprocess.PIPE)

                required_node_version = check_js_threshold_version(n)
                break
        except subprocess.CalledProcessError:
            pass
        except OSError as e:
            if e.errno == errno.ENOENT:
                pass
            else:
                raise

    if nodejs is None or nodejs is not None and required_node_version is False:
        try:
            nodeimg = "node:slim"
            global have_node_slim
            if not have_node_slim:
                dockerimgs = subprocess.check_output(["docker", "images", nodeimg]).decode('utf-8')
                if len(dockerimgs.split("\n")) <= 1:
                    nodejsimg = subprocess.check_output(["docker", "pull", nodeimg])
                    _logger.info("Pulled Docker image %s %s", nodeimg, nodejsimg)
                have_node_slim = True
            nodejs = subprocess.Popen(["docker", "run",
                                       "--attach=STDIN", "--attach=STDOUT", "--attach=STDERR",
                                       "--sig-proxy=true", "--interactive",
                                       "--rm", nodeimg, "node", "--eval", nodecode],
                                      stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            docker = True
        except OSError as e:
            if e.errno == errno.ENOENT:
                pass
            else:
                raise
        except subprocess.CalledProcessError:
            pass

    # docker failed and nodejs not on system
    if nodejs is None:
        raise JavascriptException(
            u"cwltool requires Node.js engine to evaluate Javascript "
            "expressions, but couldn't find it.  Tried %s, docker run "
            "node:slim" % u", ".join(trynodes))

    # docker failed, but nodejs is installed on system but the version is below the required version
    if docker is False and required_node_version is False:
        raise JavascriptException(
            u'cwltool requires minimum v{} version of Node.js engine.'.format(minimum_node_version_str),
            u'Try updating: https://docs.npmjs.com/getting-started/installing-node')

    return nodejs


def execjs(js, jslib, timeout=None, debug=False):  # type: (Union[Mapping, Text], Any, int, bool) -> JSON

    if not hasattr(localdata, "proc") or localdata.proc.poll() is not None or onWindows():
        localdata.proc = new_js_proc()

    nodejs = localdata.proc

    fn = u"\"use strict\";\n%s\n(function()%s)()" %\
         (jslib, js if isinstance(js, six.string_types) and len(js) > 1 and js[0] == '{' else ("{return (%s);}" % js))

    killed = []

    def term():
        try:
            killed.append(True)
            nodejs.kill()
        except OSError:
            pass

    if timeout is None:
        timeout = 20

    tm = threading.Timer(timeout, term)
    tm.start()

    stdin_buf = BytesIO((json.dumps(fn) + "\n").encode('utf-8'))
    stdout_buf = BytesIO()
    stderr_buf = BytesIO()

    rselect = [nodejs.stdout, nodejs.stderr]  # type: List[BytesIO]
    wselect = [nodejs.stdin]  # type: List[BytesIO]

    # 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':
        READ_BYTES_SIZE = 512

        # creating queue for reading from a thread to queue
        input_queue = Queue.Queue()
        output_queue = Queue.Queue()
        error_queue = Queue.Queue()

        # To tell threads that output has ended and threads can safely exit
        no_more_output = threading.Lock()
        no_more_output.acquire()
        no_more_error = threading.Lock()
        no_more_error.acquire()

        # put constructed command to input queue which then will be passed to nodejs's stdin
        def put_input(input_queue):
            while True:
                b = stdin_buf.read(READ_BYTES_SIZE)
                if b:
                    input_queue.put(b)
                else:
                    break

        # get the output from nodejs's stdout and continue till otuput ends
        def get_output(output_queue):
            while not no_more_output.acquire(False):
                b=os.read(nodejs.stdout.fileno(), READ_BYTES_SIZE)
                if b:
                    output_queue.put(b)<commit_msg>Py 3 doesn't have Queue module, use multiprocessing.queue instead<commit_after>from __future__ import absolute_import
import errno
import json
import logging
import os
import select
import subprocess
import threading
from multiprocessing import Queue
import sys
from io import BytesIO
from typing import Any, Dict, List, Mapping, Text, Tuple, Union
from .utils import onWindows
from pkg_resources import resource_stream

import six

class JavascriptException(Exception):
    pass


_logger = logging.getLogger("cwltool")

JSON = Union[Dict[Text, Any], List[Any], Text, int, float, bool, None]

localdata = threading.local()

have_node_slim = False
# minimum acceptable version of nodejs engine
minimum_node_version_str = '0.10.26'

def check_js_threshold_version(working_alias):
    # type: (str) -> bool

    """Checks if the nodeJS engine version on the system
    with the allowed minimum version.
    https://github.com/nodejs/node/blob/master/CHANGELOG.md#nodejs-changelog
    """
    # parse nodejs version into int Tuple: 'v4.2.6\n' -> [4, 2, 6]
    current_version_str = subprocess.check_output(
        [working_alias, "-v"]).decode('ascii')

    current_version = [int(v) for v in current_version_str.strip().strip('v').split('.')]
    minimum_node_version = [int(v) for v in minimum_node_version_str.split('.')]

    if current_version >= minimum_node_version:
        return True
    else:
        return False


def new_js_proc():
    # type: () -> subprocess.Popen

    res = resource_stream(__name__, 'cwlNodeEngine.js')
    nodecode = res.read()

    required_node_version, docker = (False,)*2
    nodejs = None
    trynodes = ("nodejs", "node")
    for n in trynodes:
        try:
            if subprocess.check_output([n, "--eval", "process.stdout.write('t')"]) != "t":
                continue
            else:
                nodejs = subprocess.Popen([n, "--eval", nodecode],
                                          stdin=subprocess.PIPE,
                                          stdout=subprocess.PIPE,
                                          stderr=subprocess.PIPE)

                required_node_version = check_js_threshold_version(n)
                break
        except subprocess.CalledProcessError:
            pass
        except OSError as e:
            if e.errno == errno.ENOENT:
                pass
            else:
                raise

    if nodejs is None or nodejs is not None and required_node_version is False:
        try:
            nodeimg = "node:slim"
            global have_node_slim
            if not have_node_slim:
                dockerimgs = subprocess.check_output(["docker", "images", nodeimg]).decode('utf-8')
                if len(dockerimgs.split("\n")) <= 1:
                    nodejsimg = subprocess.check_output(["docker", "pull", nodeimg])
                    _logger.info("Pulled Docker image %s %s", nodeimg, nodejsimg)
                have_node_slim = True
            nodejs = subprocess.Popen(["docker", "run",
                                       "--attach=STDIN", "--attach=STDOUT", "--attach=STDERR",
                                       "--sig-proxy=true", "--interactive",
                                       "--rm", nodeimg, "node", "--eval", nodecode],
                                      stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            docker = True
        except OSError as e:
            if e.errno == errno.ENOENT:
                pass
            else:
                raise
        except subprocess.CalledProcessError:
            pass

    # docker failed and nodejs not on system
    if nodejs is None:
        raise JavascriptException(
            u"cwltool requires Node.js engine to evaluate Javascript "
            "expressions, but couldn't find it.  Tried %s, docker run "
            "node:slim" % u", ".join(trynodes))

    # docker failed, but nodejs is installed on system but the version is below the required version
    if docker is False and required_node_version is False:
        raise JavascriptException(
            u'cwltool requires minimum v{} version of Node.js engine.'.format(minimum_node_version_str),
            u'Try updating: https://docs.npmjs.com/getting-started/installing-node')

    return nodejs


def execjs(js, jslib, timeout=None, debug=False):  # type: (Union[Mapping, Text], Any, int, bool) -> JSON

    if not hasattr(localdata, "proc") or localdata.proc.poll() is not None or onWindows():
        localdata.proc = new_js_proc()

    nodejs = localdata.proc

    fn = u"\"use strict\";\n%s\n(function()%s)()" %\
         (jslib, js if isinstance(js, six.string_types) and len(js) > 1 and js[0] == '{' else ("{return (%s);}" % js))

    killed = []

    def term():
        try:
            killed.append(True)
            nodejs.kill()
        except OSError:
            pass

    if timeout is None:
        timeout = 20

    tm = threading.Timer(timeout, term)
    tm.start()

    stdin_buf = BytesIO((json.dumps(fn) + "\n").encode('utf-8'))
    stdout_buf = BytesIO()
    stderr_buf = BytesIO()

    rselect = [nodejs.stdout, nodejs.stderr]  # type: List[BytesIO]
    wselect = [nodejs.stdin]  # type: List[BytesIO]

    # 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':
        READ_BYTES_SIZE = 512

        # creating queue for reading from a thread to queue
        input_queue = Queue()
        output_queue = Queue()
        error_queue = Queue()

        # To tell threads that output has ended and threads can safely exit
        no_more_output = threading.Lock()
        no_more_output.acquire()
        no_more_error = threading.Lock()
        no_more_error.acquire()

        # put constructed command to input queue which then will be passed to nodejs's stdin
        def put_input(input_queue):
            while True:
                b = stdin_buf.read(READ_BYTES_SIZE)
                if b:
                    input_queue.put(b)
                else:
                    break

        # get the output from nodejs's stdout and continue till otuput ends
        def get_output(output_queue):
            while not no_more_output.acquire(False):
                b=os.read(nodejs.stdout.fileno(), READ_BYTES_SIZE)
                if b:
                    output_queue.put(b)
{'fraction_non_alphanumeric': 0.10118440662012707, 'fraction_numerical': 0.0041571887991215, 'mean_word_length': 2.0879147493339794, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 15, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4366787', 'n_tokens_mistral': 3818, 'n_tokens_neox': 3604, 'n_words': 1013}