<commit_before>import json
import os
import signal
import tempfile
import logging

from twisted.internet.protocol import ProcessProtocol
from twisted.internet import defer, task
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

OOB_FD = 3


class OOBDataProtocol(LineReceiver):

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

    delimiter = '\n'

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

    def lineReceived(self, line):
        if ':' not in line:
            self.logger.error('malformed oob data: %r' % line)
            return
        key, value = line.split(':', 1)
        self._source_lock.run(
            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, _source_lock):
        self.deferred = deferred
        self.log_file = log_file
        self.source = source
        self.oob_data = OOBDataProtocol(source, board_name, _source_lock)

    def childDataReceived(self, childFD, data):
        if childFD == OOB_FD:
            self.oob_data.dataReceived(data)
        self.log_file.write(data)
        # Check size of file here, terminate if too big.
        self.log_file.flush()

    def processEnded(self, reason):
        self.log_file.close()
        self.deferred.callback(reason.value.exitCode)


class Job(object):

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

    def __init__(self, job_data, dispatcher, source, board_name, reactor,
                 log_file, log_level):
        self.job_data = job_data
        self.dispatcher = dispatcher
        self.source = source
        self.board_name = board_name
        self.reactor = reactor
        self._json_file = None
        self._source_lock = defer.DeferredLock()
        self._checkCancel_call = task.LoopingCall(self._checkCancel)
        self._signals = ['SIGINT', 'SIGINT', 'SIGTERM', 'SIGTERM', 'SIGKILL']
        self._time_limit_call = None
        self._killing = False
        self.job_log_file = None

    def _checkCancel(self):
        if self._killing:
            self._cancel()
        else:
            return self._source_lock.run(
                self.source.jobCheckForCancellation, self.board_name).addCallback(
                self._maybeCancel)

    def _cancel(self):
        self._killing = True
        if self._signals:
            signame = self._signals.pop(0)
        else:
            self.logger.warning("self._signals is empty!")
            signame = 'SIGKILL'
        self.logger.info(
            'attempting to kill job with signal %s' % signame)
        self._protocol.transport.signalProcess(getattr(signal, signame))

    def _maybeCancel(self, cancel):
        if cancel:
            self.logger.info("killing job by user request")
            self.job_log_file.write("\nKILLING JOB BY USER REQUEST\n")
            self._cancel()
        else:
            logging.debug('not cancelling')

    def _time_limit_exceeded(self):
        self.job_log_file.write("\nJOB EXCEEDED TIMEOUT %s, KILLING\n")
        self._cancel()

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

    def _run(self, job_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._protocol = DispatcherProcessProtocol(
            d, job_log_file, self.source, self.board_name, self._source_lock)
        self.job_log_file = job_log_file
        self.reactor.spawnProcess(
            self._protocol, self.dispatcher, args=[
                self.dispatcher, self._json_file, '--oob-fd', str(OOB_FD)],
            childFDs={0:0, 1:'r', 2:'r', OOB_FD:'r'}, env=None)
        self._checkCancel_call.start(10)
        timeout = max(self.json_data['timeout'], 24*60*60)
        self._time_limit_call = self.reactor.callLater(
            timeout, self._time_limit_exceeded)
        d.addBoth(self._exited)
        return d


    def _exited(self, exit_code):
        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._time_limit_call.cancel()
        self._checkCancel_call.stop()
        return self._source_lock.run(
            self.source.jobCompleted, self.board_name, exit_code).addCallback(
            lambda r:exit_code)


class SimplePP(ProcessProtocol):
    def __init__(self, d):
        self.d = d
    def processEnded(self, reason):
        self.d.callback(None)
<commit_msg>kill on log file getting too big too<commit_after>import json
import os
import signal
import tempfile
import logging

from twisted.internet.protocol import ProcessProtocol
from twisted.internet import defer, task
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

OOB_FD = 3


class OOBDataProtocol(LineReceiver):

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

    delimiter = '\n'

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

    def lineReceived(self, line):
        if ':' not in line:
            self.logger.error('malformed oob data: %r' % line)
            return
        key, value = line.split(':', 1)
        self._source_lock.run(
            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, job):
        self.deferred = deferred
        self.log_file = log_file
        self.job = job
        self.oob_data = OOBDataProtocol(
            job.source, job.board_name, job._source_lock)

    def childDataReceived(self, childFD, data):
        if childFD == OOB_FD:
            self.oob_data.dataReceived(data)
        self.log_file.write(data)
        # Check size of file here, terminate if too big.
        if self.log_file.tell() > 1e9:
            self.job.cancel("EXCEEDED LOG SIZE LIMIT")
        self.log_file.flush()

    def processEnded(self, reason):
        self.log_file.close()
        self.deferred.callback(reason.value.exitCode)


class Job(object):

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

    def __init__(self, job_data, dispatcher, source, board_name, reactor,
                 log_file, log_level):
        self.job_data = job_data
        self.dispatcher = dispatcher
        self.source = source
        self.board_name = board_name
        self.reactor = reactor
        self._json_file = None
        self._source_lock = defer.DeferredLock()
        self._checkCancel_call = task.LoopingCall(self._checkCancel)
        self._signals = ['SIGINT', 'SIGINT', 'SIGTERM', 'SIGTERM', 'SIGKILL']
        self._time_limit_call = None
        self._killing = False
        self.job_log_file = None

    def _checkCancel(self):
        if self._killing:
            self._cancel()
        else:
            return self._source_lock.run(
                self.source.jobCheckForCancellation, self.board_name).addCallback(
                self._maybeCancel)

    def cancel(self, reason=None):
        if not self._killing and reason is None:
            reason = "killing job for unknown reason"
        if reason is not None:
            self.logger.info(reason)
            self.job_log_file.write("\n%s\n" % reason.upper())
        self._killing = True
        if self._signals:
            signame = self._signals.pop(0)
        else:
            self.logger.warning("self._signals is empty!")
            signame = 'SIGKILL'
        self.logger.info(
            'attempting to kill job with signal %s' % signame)
        self._protocol.transport.signalProcess(getattr(signal, signame))

    def _maybeCancel(self, cancel):
        if cancel:
            self.cancel("killing job by user request")
        else:
            logging.debug('not cancelling')

    def _time_limit_exceeded(self):
        self.cancel("killing job for exceeding timeout")

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

    def _run(self, job_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._protocol = DispatcherProcessProtocol(
            d, job_log_file, self)
        self.job_log_file = job_log_file
        self.reactor.spawnProcess(
            self._protocol, self.dispatcher, args=[
                self.dispatcher, self._json_file, '--oob-fd', str(OOB_FD)],
            childFDs={0:0, 1:'r', 2:'r', OOB_FD:'r'}, env=None)
        self._checkCancel_call.start(10)
        timeout = max(self.json_data['timeout'], 24*60*60)
        self._time_limit_call = self.reactor.callLater(
            timeout, self._time_limit_exceeded)
        d.addBoth(self._exited)
        return d


    def _exited(self, exit_code):
        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._time_limit_call.cancel()
        self._checkCancel_call.stop()
        return self._source_lock.run(
            self.source.jobCompleted, self.board_name, exit_code).addCallback(
            lambda r:exit_code)


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

{'fraction_non_alphanumeric': 0.1035132334934539, 'fraction_numerical': 0.003014034096260714, 'mean_word_length': 2.385841836734694, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, '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': '337579', 'n_tokens_mistral': 3299, 'n_tokens_neox': 3188, 'n_words': 672}