<commit_before>import logging

import eventlet

from akanda.rug.common.exceptions import AbortTask


LOG = logging.getLogger(__name__)


class Task(object):
    def __init__(self, method, data, max_attempts=3, reason=None):
        self.method = method
        self.data = data
        self.current = 0
        self.max_attempts = max_attempts
        self.reason = reason

    def __call__(self):
        self.current += 1
        self.method(self.data)

    def should_retry(self):
        return self.current < self.max_attempts

    def __repr__(self):
        msg = '<Task method: %s reason: (%s) data: %s attempt: %d/%d >'

        return msg % (self.method.__name__,
                      self.reason,
                      self.data,
                      self.current,
                      self.max_attempts)


class TaskManager(object):
    def __init__(self, max_requeue_delay=10):
        self.task_queue = eventlet.queue.LightQueue()
        self.delay_queue = eventlet.queue.LightQueue()
        self.blocked = set()
        self.max_requeue_delay = max_requeue_delay

    def put(self, method, data, max_attempts=3, reason=None):
        self.task_queue.put(Task(method, data, max_attempts, reason))

    def start(self):
        eventlet.spawn(self._serialized_task_runner)
        eventlet.spawn(self._requeue_failed)

    def _serialized_task_runner(self):
        while True:
            LOG.info('Waiting on item')
            task = self.task_queue.get()
            try:
                LOG.debug('starting %s', task)
                task()
                LOG.debug('success for task %s', task)
            except AbortTask as e:
                LOG.warn('Task aborted: %s (%s)', e, task)
            except Exception as e:
                try:
                    if isinstance(e, Warning):
                        LOG.warn('Task: %s (%s)' % (e, task))
                    else:
                        LOG.exception('Task failed: %s (%s)' % (e, task))

                    if task.should_retry():
                        self.delay_queue.put(task)
                    else:
                        LOG.error('Task Error: %s (%s)' % (e, task))
                except Exception:
                    LOG.exception('Error processing exception in task')

    def _requeue_failed(self):
        while True:<commit_msg>Log the number of tasks waiting to be executed<commit_after>import logging

import eventlet

from akanda.rug.common.exceptions import AbortTask


LOG = logging.getLogger(__name__)


class Task(object):
    def __init__(self, method, data, max_attempts=3, reason=None):
        self.method = method
        self.data = data
        self.current = 0
        self.max_attempts = max_attempts
        self.reason = reason

    def __call__(self):
        self.current += 1
        self.method(self.data)

    def should_retry(self):
        return self.current < self.max_attempts

    def __repr__(self):
        msg = '<Task method: %s reason: (%s) data: %s attempt: %d/%d >'

        return msg % (self.method.__name__,
                      self.reason,
                      self.data,
                      self.current,
                      self.max_attempts)


class TaskManager(object):
    def __init__(self, max_requeue_delay=10):
        self.task_queue = eventlet.queue.LightQueue()
        self.delay_queue = eventlet.queue.LightQueue()
        self.blocked = set()
        self.max_requeue_delay = max_requeue_delay

    def put(self, method, data, max_attempts=3, reason=None):
        self.task_queue.put(Task(method, data, max_attempts, reason))

    def start(self):
        eventlet.spawn(self._serialized_task_runner)
        eventlet.spawn(self._requeue_failed)

    def _serialized_task_runner(self):
        while True:
            LOG.info('Waiting to execute one of the %d tasks '
                     'remaining in the queue', self.task_queue.qsize())
            task = self.task_queue.get()
            try:
                LOG.debug('starting %s', task)
                task()
                LOG.debug('success for task %s', task)
            except AbortTask as e:
                LOG.warn('Task aborted: %s (%s)', e, task)
            except Exception as e:
                try:
                    if isinstance(e, Warning):
                        LOG.warn('Task: %s (%s)' % (e, task))
                    else:
                        LOG.exception('Task failed: %s (%s)' % (e, task))

                    if task.should_retry():
                        self.delay_queue.put(task)
                    else:
                        LOG.error('Task Error: %s (%s)' % (e, task))
                except Exception:
                    LOG.exception('Error processing exception in task')

    def _requeue_failed(self):
        while True:
{'fraction_non_alphanumeric': 0.10491257285595337, 'fraction_numerical': 0.002497918401332223, 'mean_word_length': 1.7100958826847152, 'pattern_counts': {'":': 0, '<': 7, '<?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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3187208', 'n_tokens_mistral': 1412, 'n_tokens_neox': 1328, 'n_words': 325}