<commit_before>"""
Skipping shell invocations is good, when possible. This wrapper around subprocess does dirty work of
parsing command lines into argv and making sure that no shell magic is being used.
"""

import subprocess, shlex, re, logging, sys, threading, traceback
import command

_log = logging.getLogger('pymake.process')

blacklist = re.compile(r'[=\\$><;*?[{~`|&]')
def clinetoargv(cline):
    """
    If this command line can safely skip the shell, return an argv array.
    """

    if blacklist.search(cline) is not None:
        return None

    return shlex.split(cline, comments=True)

shellwords = (':', '.', 'break', 'cd', 'continue', 'exec', 'exit', 'export',
              'getopts', 'hash', 'pwd', 'readonly', 'return', 'shift', 
              'test', 'times', 'trap', 'umask', 'unset', 'alias',
              'set', 'bind', 'builtin', 'caller', 'command', 'declare',
              'echo', 'enable', 'help', 'let', 'local', 'logout', 
              'printf', 'read', 'shopt', 'source', 'type', 'typeset',
              'ulimit', 'unalias', 'set')

def call(cline, env, cwd, loc, cb, context, echo):
    argv = clinetoargv(cline)
    if argv is None or (len(argv) and argv[0] in shellwords):
        _log.debug("%s: Running command through shell because of shell metacharacters" % (loc,))
        context.call(cline, shell=True, env=env, cwd=cwd, cb=cb, echo=echo)
        return

    if not len(argv):
        cb(res=0)
        return

    if argv[0] == command.makepypath:
        command.main(argv[1:], env, cwd, context, cb)
        return

    if argv[0:2] == [sys.executable, command.makepypath]:
        command.main(argv[2:], env, cwd, context, cb)
        return

    _log.debug("%s: skipping shell, no metacharacters found" % (loc,))
    context.call(argv, shell=False, env=env, cwd=cwd, cb=cb, echo=echo)

class ParallelContext(object):
    """
    Manages the parallel execution of processes.
    """

    def __init__(self, jcount):
        self.jcount = jcount
        self.exit = False

        self.lock = threading.Lock()
        self.threadcount = 0

        self.notifypending = threading.Condition(self.lock)
        self.pending = []
        self.running = 0

        self.notifyresult = threading.Condition(self.lock)
        self.results = []

    def _run(self):
        self.notifypending.acquire()
        try:
            while True:
                if len(self.pending) == 0:
                    self.notifypending.wait()
                    continue

                argv, shell, env, cwd, cb, echo, idx = self.pending.pop(0)
                self.running += 1

                _log.debug("running <%i>: pending: %i running %i threads: %i: results: %i" % (idx,
                                                                                              len(self.pending),
                                                                                              self.running,
                                                                                              self.threadcount,
                                                                                              len(self.results)))

                self.notifypending.release()
                if echo is not None:
                    print echo
                res = subprocess.call(argv, shell=shell, env=env, cwd=cwd)
                self.notifypending.acquire()

                self.running -= 1
                self.results.append((cb, res, idx))
                self.notifyresult.notify()

                continue
        finally:
            self.notifypending.release()

    counter = 0

    def call(self, argv, shell, env, cwd, cb, echo):
        """
        Asynchronously call the process
        """
        self.notifypending.acquire()
        try:
            idx = ParallelContext.counter
            ParallelContext.counter += 1

            _log.debug("call <%i>: pending: %i running %i threads: %i: results: %i: %r" % (idx,
                                                                                           len(self.pending),
                                                                                           self.running,
                                                                                           self.threadcount,
                                                                                           len(self.results), argv))

            self.pending.append((argv, shell, env, cwd, cb, echo, idx))
            self.notifypending.notify()
            if len(self.pending) and self.threadcount < self.jcount:
                _log.debug("Creating a new thread for process execution")
                self.threadcount += 1
                t = threading.Thread(target=self._run)
                t.setDaemon(True)
                t.start()
        finally:
            self.notifypending.release()

    def spin(self):
        """
        Spin the 'event loop', and return only when it is empty.
        """
        self.notifyresult.acquire()
        while True:
            _log.debug("spin: pending: %i running %i threads: %i results: %i" % (len(self.pending),
                                                                                 self.running,
                                                                                 self.threadcount,
                                                                                 len(self.results)))
            if len(self.results) > 0:
                lresults = self.results
                self.results = []
                self.notifyresult.release()
                try:
                    for cb, res, idx in lresults:
                        _log.debug("results of <%i>: %s" % (idx, res))
                        try:
                            cb(res)
                        except Exception, e:
                            _log.debug("Exception throwing during callback: %s" % (e,))
                            raise
                finally:
                    self.notifyresult.acquire()

            if len(self.pending) == 0 and self.running == 0 and len(self.results) == 0:
                _log.debug("exiting spin")
                self.notifyresult.release()
                return

            self.notifyresult.wait(20)
            if len(self.results) == 0:
                raise Exception("wait timeout")
<commit_msg>Switch from threads to waitpid, and take deferred Windows pain.<commit_after>"""
Skipping shell invocations is good, when possible. This wrapper around subprocess does dirty work of
parsing command lines into argv and making sure that no shell magic is being used.
"""

import subprocess, shlex, re, logging, sys, traceback, os
import command

_log = logging.getLogger('pymake.process')

blacklist = re.compile(r'[=\\$><;*?[{~`|&]')
def clinetoargv(cline):
    """
    If this command line can safely skip the shell, return an argv array.
    """

    if blacklist.search(cline) is not None:
        return None

    return shlex.split(cline, comments=True)

shellwords = (':', '.', 'break', 'cd', 'continue', 'exec', 'exit', 'export',
              'getopts', 'hash', 'pwd', 'readonly', 'return', 'shift', 
              'test', 'times', 'trap', 'umask', 'unset', 'alias',
              'set', 'bind', 'builtin', 'caller', 'command', 'declare',
              'echo', 'enable', 'help', 'let', 'local', 'logout', 
              'printf', 'read', 'shopt', 'source', 'type', 'typeset',
              'ulimit', 'unalias', 'set')

def call(cline, env, cwd, loc, cb, context, echo):
    argv = clinetoargv(cline)
    if argv is None or (len(argv) and argv[0] in shellwords):
        _log.debug("%s: Running command through shell because of shell metacharacters" % (loc,))
        context.call(cline, shell=True, env=env, cwd=cwd, cb=cb, echo=echo)
        return

    if not len(argv):
        cb(res=0)
        return

    if argv[0] == command.makepypath:
        command.main(argv[1:], env, cwd, context, cb)
        return

    if argv[0:2] == [sys.executable, command.makepypath]:
        command.main(argv[2:], env, cwd, context, cb)
        return

    _log.debug("%s: skipping shell, no metacharacters found" % (loc,))
    context.call(argv, shell=False, env=env, cwd=cwd, cb=cb, echo=echo)

def statustoresult(status):
    """
    Convert the status returned from waitpid into a prettier numeric result.
    """
    sig = status & 0xFF
    if sig:
        return -sig

    return status >>8

class ParallelContext(object):
    """
    Manages the parallel execution of processes.
    """

    def __init__(self, jcount):
        self.jcount = jcount
        self.exit = False

        self.pending = [] # list of (argv, shell, env, cwd, cb, echo)
        self.running = [] # list of (subprocess, cb)

    def run(self):
        while len(self.running) < self.jcount and len(self.pending):
            argv, shell, env, cwd, cb, echo = self.pending.pop(0)

            if echo is not None:
                print echo
            p = subprocess.Popen(argv, shell=shell, env=env, cwd=cwd)
            self.running.append((p, cb))

    counter = 0

    def call(self, argv, shell, env, cwd, cb, echo):
        """
        Asynchronously call the process
        """

        self.pending.append((argv, shell, env, cwd, cb, echo))
        self.run()

    def spin(self):
        """
        Spin the 'event loop', and return only when it is empty.
        """

        while len(self.pending) or len(self.running):
            self.run()
            assert len(self.running)

            _log.debug("spin: pending: %i running %i" % (len(self.pending),
                                                         len(self.running)))
            
            pid, status = os.waitpid(-1, 0)
            result = statustoresult(status)

            for i in xrange(0, len(self.running)):
                p, cb = self.running[i]
                if p.pid == pid:
                    cb(result)
                    del self.running[i]
                    break

        _log.debug("exiting spin")

{'fraction_non_alphanumeric': 0.12069136783550213, 'fraction_numerical': 0.0036753749875831927, 'mean_word_length': 1.5007451564828613, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 43, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2834437', 'n_tokens_mistral': 2780, 'n_tokens_neox': 2636, 'n_words': 762}