<commit_before>import json
import os
import tempfile
import logging

from twisted.internet.protocol import ProcessProtocol
from twisted.internet import defer
from twisted.protocols.basic import LineReceiver


def catchall_errback(logger):
    def eb(failure):
        logger.error(
            '%s: %s\n%s', failure.type.__name__, failure.value,
            failure.getTraceback())
    return eb


class OOBDataProtocol(LineReceiver):

    logger = logging.getLogger(__name__ + '.OOBDataProtocol')

    delimiter = '\n'

    def __init__(self, source, board_name):
        self.source = source
        self.board_name = board_name

    def lineReceived(self, line):
        if ':' not in line:
            self.logger.error('malformed oob data: %r' % line)
            return
        key, value = line.split(':', 1)
        self.source.jobOobData(
            self.board_name, key, value.lstrip()).addErrback(
                catchall_errback(self.logger))


class DispatcherProcessProtocol(ProcessProtocol):

    logger = logging.getLogger(__name__ + '.DispatcherProcessProtocol')

    def __init__(self, deferred, log_file, source, board_name):
        self.deferred = deferred
        self.log_file = log_file
        self.source = source
        self.oob_data = OOBDataProtocol(source, board_name)

    def childDataReceived(self, childFD, data):
        if childFD == 3:
            self.oob_data.dataReceived(data)
        self.log_file.write(data)
        self.log_file.flush()

    def processEnded(self, reason):
        # This discards the process exit value.
        self.log_file.close()
        self.deferred.callback(None)


class Job(object):

    logger = logging.getLogger(__name__ + '.Job')

    def __init__(self, job_data, dispatcher, source, board_name, reactor):
        self.job_data = job_data
        self.dispatcher = dispatcher
        self.source = source
        self.board_name = board_name
        self.reactor = reactor
        self._json_file = None

    def run(self):
        d = self.source.getLogFileForJobOnBoard(self.board_name)
        return d.addCallback(self._run).addErrback(
            catchall_errback(self.logger))

    def _run(self, log_file):
        d = defer.Deferred()
        json_data = self.job_data
        fd, self._json_file = tempfile.mkstemp()
        with os.fdopen(fd, 'wb') as f:
            json.dump(json_data, f)
        self.reactor.spawnProcess(
            DispatcherProcessProtocol(
                d, log_file, self.source, self.board_name),
            self.dispatcher, args=[
                self.dispatcher, self._json_file, '--oob-fd', '3'],
            childFDs={0:0, 1:'r', 2:'r', 3:'r'}, env=None)
        d.addBoth(self._exited)
        return d

    def _exited(self, result):
        self.logger.info("job finished on %s", self.job_data['target'])
        if self._json_file is not None:
            os.unlink(self._json_file)
        self.logger.info("reporting job completed")
        self.source.jobCompleted(self.board_name)
        return result


class SimplePP(ProcessProtocol):
    def __init__(self, d):
        self.d = d
    def processEnded(self, reason):
        self.d.callback(None)
<commit_msg>oops, wait for jobComplete to finish before exiting<commit_after>import json
import os
import tempfile
import logging

from twisted.internet.protocol import ProcessProtocol
from twisted.internet import defer
from twisted.protocols.basic import LineReceiver


def catchall_errback(logger):
    def eb(failure):
        logger.error(
            '%s: %s\n%s', failure.type.__name__, failure.value,
            failure.getTraceback())
    return eb


class OOBDataProtocol(LineReceiver):

    logger = logging.getLogger(__name__ + '.OOBDataProtocol')

    delimiter = '\n'

    def __init__(self, source, board_name):
        self.source = source
        self.board_name = board_name

    def lineReceived(self, line):
        if ':' not in line:
            self.logger.error('malformed oob data: %r' % line)
            return
        key, value = line.split(':', 1)
        self.source.jobOobData(
            self.board_name, key, value.lstrip()).addErrback(
                catchall_errback(self.logger))


class DispatcherProcessProtocol(ProcessProtocol):

    logger = logging.getLogger(__name__ + '.DispatcherProcessProtocol')

    def __init__(self, deferred, log_file, source, board_name):
        self.deferred = deferred
        self.log_file = log_file
        self.source = source
        self.oob_data = OOBDataProtocol(source, board_name)

    def childDataReceived(self, childFD, data):
        if childFD == 3:
            self.oob_data.dataReceived(data)
        self.log_file.write(data)
        self.log_file.flush()

    def processEnded(self, reason):
        # This discards the process exit value.
        self.log_file.close()
        self.deferred.callback(None)


class Job(object):

    logger = logging.getLogger(__name__ + '.Job')

    def __init__(self, job_data, dispatcher, source, board_name, reactor):
        self.job_data = job_data
        self.dispatcher = dispatcher
        self.source = source
        self.board_name = board_name
        self.reactor = reactor
        self._json_file = None

    def run(self):
        d = self.source.getLogFileForJobOnBoard(self.board_name)
        return d.addCallback(self._run).addErrback(
            catchall_errback(self.logger))

    def _run(self, log_file):
        d = defer.Deferred()
        json_data = self.job_data
        fd, self._json_file = tempfile.mkstemp()
        with os.fdopen(fd, 'wb') as f:
            json.dump(json_data, f)
        self.reactor.spawnProcess(
            DispatcherProcessProtocol(
                d, log_file, self.source, self.board_name),
            self.dispatcher, args=[
                self.dispatcher, self._json_file, '--oob-fd', '3'],
            childFDs={0:0, 1:'r', 2:'r', 3:'r'}, env=None)
        d.addBoth(self._exited)
        return d

    def _exited(self, result):
        self.logger.info("job finished on %s", self.job_data['target'])
        if self._json_file is not None:
            os.unlink(self._json_file)
        self.logger.info("reporting job completed")
        return self.source.jobCompleted(self.board_name).addCallback(
            lambda r:result)


class SimplePP(ProcessProtocol):
    def __init__(self, d):
        self.d = d
    def processEnded(self, reason):
        self.d.callback(None)

{'fraction_non_alphanumeric': 0.10680514648891645, 'fraction_numerical': 0.0024802356223841263, 'mean_word_length': 2.4989154013015185, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4867486', 'n_tokens_mistral': 1997, 'n_tokens_neox': 1921, 'n_words': 426}