<commit_before>from __future__ import absolute_import

import logging
import sys
import threading

from datetime import datetime
from flask import current_app
from subprocess import PIPE, Popen, STDOUT
from time import sleep, time

from freight.notifiers import NotifierEvent
from freight.notifiers.utils import send_task_notifications
from freight.config import celery, db
from freight.constants import PROJECT_ROOT
from freight.models import LogChunk, Task, TaskStatus


@celery.task(name='freight.execute_task', max_retries=None)
def execute_task(task_id):
    logging.debug('ExecuteTask fired with %d active thread(s)',
                  threading.active_count())

    task = Task.query.get(task_id)
    if not task:
        logging.warning('ExecuteTask fired with missing Task(id=%s)', task_id)
        return

    if task.status not in (TaskStatus.pending, TaskStatus.in_progress):
        logging.warning('ExecuteTask fired with finished Task(id=%s)', task_id)
        return

    send_task_notifications(task, NotifierEvent.TASK_STARTED)

    provider_config = task.provider_config

    # wipe the log incase this is a retry
    LogChunk.query.filter(
        LogChunk.task_id == task.id,
    ).delete()

    taskrunner = TaskRunner(
        task=task,
        timeout=provider_config.get('timeout', current_app.config['DEFAULT_TIMEOUT']),
        read_timeout=provider_config.get('read_timeout', current_app.config['DEFAULT_READ_TIMEOUT']),
    )
    taskrunner.start()
    taskrunner.wait()

    # reload the task from the database due to subprocess changes
    db.session.expire(task)
    db.session.refresh(task)

    if task.status in (TaskStatus.pending, TaskStatus.in_progress):
        logging.error('Task(id=%s) did not finish cleanly', task.id)
        task.status = TaskStatus.failed
        task.date_finished = datetime.utcnow()
        db.session.add(task)
        db.session.commit()

    send_task_notifications(task, NotifierEvent.TASK_FINISHED)


def kill_subprocess(process):
    logging.debug('Sending kill() to process %s', process.pid)
    process.kill()


def forcefully_stop_process(process, timeout=10):
    timer = threading.Timer(timeout, kill_subprocess, args=[process])
    timer.start()
    try:
        logging.debug('Sending terminate() to process %s (%ds timeout)',
                      process.pid, timeout)
        process.terminate()
        process.wait()
    finally:
        timer.cancel()


class LogReporter(threading.Thread):
    def __init__(self, app_context, task_id, process, chunk_size=4096):
        self.app_context = app_context
        self.task_id = task_id
        self.process = process
        self.chunk_size = chunk_size
        self.cur_offset = 0
        self.last_recv = None
        self.active = True
        threading.Thread.__init__(self)
        self.daemon = True

    def save_chunk(self, text):
        # we also want to pipe this to stdout
        sys.stdout.write(text)

        text = text.decode('utf-8', 'replace')
        text_len = len(text)
        db.session.add(LogChunk(
            task_id=self.task_id,
            text=text,
            offset=self.cur_offset,
            size=text_len,
        ))

        # we commit immediately to ensure the API can stream logs
        db.session.commit()
        self.cur_offset += text_len

    def terminate(self):
        self.active = False

    def run(self):
        with self.app_context:
            self._run()

    def _run(self):
        self.last_recv = time()
        chunk_size = self.chunk_size
        proc = self.process
        result = ''

        while self.active:
            is_running = proc.poll() is None
            chunk = proc.stdout.read(1)
            if not (is_running or chunk):
                break

            last_write = time()<commit_msg>Add lock around offset increment (fixes GH-52)<commit_after>from __future__ import absolute_import

import logging
import sys
import threading

from datetime import datetime
from flask import current_app
from subprocess import PIPE, Popen, STDOUT
from time import sleep, time

from freight.notifiers import NotifierEvent
from freight.notifiers.utils import send_task_notifications
from freight.config import celery, db
from freight.constants import PROJECT_ROOT
from freight.models import LogChunk, Task, TaskStatus


@celery.task(name='freight.execute_task', max_retries=None)
def execute_task(task_id):
    logging.debug('ExecuteTask fired with %d active thread(s)',
                  threading.active_count())

    task = Task.query.get(task_id)
    if not task:
        logging.warning('ExecuteTask fired with missing Task(id=%s)', task_id)
        return

    if task.status not in (TaskStatus.pending, TaskStatus.in_progress):
        logging.warning('ExecuteTask fired with finished Task(id=%s)', task_id)
        return

    send_task_notifications(task, NotifierEvent.TASK_STARTED)

    provider_config = task.provider_config

    # wipe the log incase this is a retry
    LogChunk.query.filter(
        LogChunk.task_id == task.id,
    ).delete()

    taskrunner = TaskRunner(
        task=task,
        timeout=provider_config.get('timeout', current_app.config['DEFAULT_TIMEOUT']),
        read_timeout=provider_config.get('read_timeout', current_app.config['DEFAULT_READ_TIMEOUT']),
    )
    taskrunner.start()
    taskrunner.wait()

    # reload the task from the database due to subprocess changes
    db.session.expire(task)
    db.session.refresh(task)

    if task.status in (TaskStatus.pending, TaskStatus.in_progress):
        logging.error('Task(id=%s) did not finish cleanly', task.id)
        task.status = TaskStatus.failed
        task.date_finished = datetime.utcnow()
        db.session.add(task)
        db.session.commit()

    send_task_notifications(task, NotifierEvent.TASK_FINISHED)


def kill_subprocess(process):
    logging.debug('Sending kill() to process %s', process.pid)
    process.kill()


def forcefully_stop_process(process, timeout=10):
    timer = threading.Timer(timeout, kill_subprocess, args=[process])
    timer.start()
    try:
        logging.debug('Sending terminate() to process %s (%ds timeout)',
                      process.pid, timeout)
        process.terminate()
        process.wait()
    finally:
        timer.cancel()


class LogReporter(threading.Thread):
    def __init__(self, app_context, task_id, process, chunk_size=4096):
        self.app_context = app_context
        self.task_id = task_id
        self.process = process
        self.chunk_size = chunk_size
        self.cur_offset = 0
        self.last_recv = None
        self.active = True
        threading.Thread.__init__(self)
        self.daemon = True
        self.write_lock = threading.Lock()

    def save_chunk(self, text):
        # we also want to pipe this to stdout
        sys.stdout.write(text)

        text = text.decode('utf-8', 'replace')
        text_len = len(text)

        with self.write_lock:
            db.session.add(LogChunk(
                task_id=self.task_id,
                text=text,
                offset=self.cur_offset,
                size=text_len,
            ))

            # we commit immediately to ensure the API can stream logs
            db.session.commit()
            self.cur_offset += text_len

    def terminate(self):
        self.active = False

    def run(self):
        with self.app_context:
            self._run()

    def _run(self):
        self.last_recv = time()
        chunk_size = self.chunk_size
        proc = self.process
        result = ''

        while self.active:
            is_running = proc.poll() is None
            chunk = proc.stdout.read(1)
            if not (is_running or chunk):
                break

            last_write = time()
{'fraction_non_alphanumeric': 0.09302924880814328, 'fraction_numerical': 0.0025769875016106174, 'mean_word_length': 2.6287985039738198, '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': '7194600', 'n_tokens_mistral': 2287, 'n_tokens_neox': 2206, 'n_words': 547}