<commit_before>"""Allows running Celery tasks asynchronously using the Tornado IOLoop."""

from tornado import concurrent
from tornado import ioloop
from tornado import stack_context
import logging

logger = logging.getLogger(__name__)


class CeleryPoller(object):

    def __init__(self, poll_freq):
        self._poll_freq = poll_freq
        self._results_callbacks = []

    @concurrent.return_future
    def run_task(self, task, *args, **kwargs):
        logger.info("Running task '{}'".format(task))
        callback = kwargs["callback"]
        del kwargs["callback"]

        # Wrap the callback it takes the result and raises exceptions in the
        # correct stack context.
        def result_callback(result): callback(result.get())
        result_callback = stack_context.wrap(result_callback)

        self._results_callbacks.append((task.delay(*args, **kwargs),
                                        result_callback))
        self._poll_tasks()

    def _poll_tasks(self):
        logger.debug("Polling Celery tasks")
        results_callbacks = []
        for result, callback in self._results_callbacks:
            if result.ready():
                logger.info("Task is done")
                # Exception should never be raised here or bad things will
                # happen.
                callback(result)
            else:
                logger.debug("Task is still pending")
                results_callbacks.append((result, callback))
        self._results_callbacks = results_callbacks
        if len(self._results_callbacks) > 0:
            logger.debug("Tasks are still pending, scheduling next poll")
            ioloop.IOLoop.instance().add_timeout(self._poll_freq,
                                                 self._poll_tasks)
        else:
            logger.debug("All tasks are complete, no polling necessary")<commit_msg>Add some logging to CeleryPoller<commit_after>"""Allows running Celery tasks asynchronously using the Tornado IOLoop."""

from tornado import concurrent
from tornado import ioloop
from tornado import stack_context
import logging
import time

logger = logging.getLogger(__name__)


class CeleryPoller(object):

    def __init__(self, poll_freq):
        self._poll_freq = poll_freq
        self._results_callbacks = []

    @concurrent.return_future
    def run_task(self, task, *args, **kwargs):
        logger.info("Running task '{}'".format(task))
        callback = kwargs["callback"]
        del kwargs["callback"]

        # Wrap the callback it takes the result and raises exceptions in the
        # correct stack context.
        def result_callback(result): callback(result.get())
        result_callback = stack_context.wrap(result_callback)

        self._results_callbacks.append((task.delay(*args, **kwargs),
                                        result_callback))
        self._poll_tasks()

    def _poll_tasks(self):
        logger.debug("Polling Celery tasks")
        results_callbacks = []
        t = time.time()
        for result, callback in self._results_callbacks:
            if result.ready():
                logger.info("Finished task")
                # Exception should never be raised here or bad things will
                # happen.
                callback(result)
            else:
                logger.debug("Task is still pending")
                results_callbacks.append((result, callback))
        t = time.time() - t
        logger.info("Polled {} task(s) in {}ms"
                    .format(len(self._results_callbacks), int(t * 1000)))
        self._results_callbacks = results_callbacks
        if len(self._results_callbacks) > 0:
            logger.debug("Tasks are still pending, scheduling next poll")
            ioloop.IOLoop.instance().add_timeout(self._poll_freq,
                                                 self._poll_tasks)
        else:
            logger.debug("All tasks are complete, no polling necessary")
{'fraction_non_alphanumeric': 0.08924949290060852, 'fraction_numerical': 0.0015212981744421906, 'mean_word_length': 2.220408163265306, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '939602', 'n_tokens_mistral': 1017, 'n_tokens_neox': 967, 'n_words': 270}