<commit_before>"""
Server-like task processor.
"""

import abc
import multiprocessing
import six
import time
import threading

try:
    import queue
except ImportError:
    import Queue as queue

from optinum import config


@six.add_metaclass(abc.ABCMeta)
class Daemon(object):

    """Abstract base class for simple daemons."""

    def __init__(self, name, debug=config.MISC.DEBUG,
                 delay=config.DAEMON.DELAY, loop=config.DAEMON.LOOP):
        """Setup a new instance."""
        # refine name
        if not name:
            name = self.__class__.__name__

        # save attributes
        self.name = name
        self.debug = debug
        self.delay = delay
        self.loop = loop

    def prologue(self):
        """Executed once before the main procedures."""
        pass

    def epilogue(self):
        """Executed once after the main procedures."""
        pass

    @abc.abstractmethod
    def task_generator(self):
        """Override this with your custom task generator."""
        pass

    @abc.abstractmethod
    def process(self, task):
        """Override this with your desired procedures."""
        pass

    def _process(self, task):
        """Wrapper over the `process`."""
        # pylint: disable=W0703
        try:
            result = self.process(task)
        except Exception as exc:
            self.task_fail(task, exc)
        else:
            self.task_done(task, result)

    def put_task(self, task):
        """Adds the task in the queue."""
        self._process(task)

    def task_done(self, task, result):
        """What to execute after successfully finished processing a task."""
        pass

    def task_fail(self, task, exc):
        """What to do when the program fails processing a task."""
        pass

    def finished(self):
        """What to execute after finishing processing all the tasks."""
        return self.loop    # decide to continue reprocessing or not

    def interrupted(self):
        """What to execute when keyboard interrupts arrive."""
        pass

    def start(self):
        """Starts a series of workers and processes incoming tasks."""
        self.prologue()
        while True:
            try:
                for task in self.task_generator():
                    self.put_task(task)
                loop = self.finished()
                if not loop:
                    break
                time.sleep(self.delay)
            except KeyboardInterrupt:
                self.interrupted()
                break
        self.epilogue()


class _ConcurrentDaemon(Daemon):

    """Abstract base class for concurrent daemons.

    Not inherited directly into final classes.
    """

    def __init__(self, *args, **kwargs):
        """Partial setup the new concurrent instance."""
        # maximum allowed queue size
        self.qsize = kwargs.pop('qsize', config.DAEMON.QSIZE)
        # desired number of workers
        self.wcount = kwargs.pop('wcount', config.DAEMON.WORKERS)
        super(_ConcurrentDaemon, self).__init__(*args, **kwargs)<commit_msg>Update _ConcurrentDaemon required arguments<commit_after>"""
Server-like task processor.
"""

import abc
import multiprocessing
import six
import time
import threading

try:
    import queue
except ImportError:
    import Queue as queue

from optinum import config


@six.add_metaclass(abc.ABCMeta)
class Daemon(object):

    """Abstract base class for simple daemons."""

    def __init__(self, name, debug=config.MISC.DEBUG,
                 delay=config.DAEMON.DELAY, loop=config.DAEMON.LOOP):
        """Setup a new instance."""
        # refine name
        if not name:
            name = self.__class__.__name__

        # save attributes
        self.name = name
        self.debug = debug
        self.delay = delay
        self.loop = loop

    def prologue(self):
        """Executed once before the main procedures."""
        pass

    def epilogue(self):
        """Executed once after the main procedures."""
        pass

    @abc.abstractmethod
    def task_generator(self):
        """Override this with your custom task generator."""
        pass

    @abc.abstractmethod
    def process(self, task):
        """Override this with your desired procedures."""
        pass

    def _process(self, task):
        """Wrapper over the `process`."""
        # pylint: disable=W0703
        try:
            result = self.process(task)
        except Exception as exc:
            self.task_fail(task, exc)
        else:
            self.task_done(task, result)

    def put_task(self, task):
        """Adds the task in the queue."""
        self._process(task)

    def task_done(self, task, result):
        """What to execute after successfully finished processing a task."""
        pass

    def task_fail(self, task, exc):
        """What to do when the program fails processing a task."""
        pass

    def finished(self):
        """What to execute after finishing processing all the tasks."""
        return self.loop    # decide to continue reprocessing or not

    def interrupted(self):
        """What to execute when keyboard interrupts arrive."""
        pass

    def start(self):
        """Starts a series of workers and processes incoming tasks."""
        self.prologue()
        while True:
            try:
                for task in self.task_generator():
                    self.put_task(task)
                loop = self.finished()
                if not loop:
                    break
                time.sleep(self.delay)
            except KeyboardInterrupt:
                self.interrupted()
                break
        self.epilogue()


class _ConcurrentDaemon(Daemon):

    """Abstract base class for concurrent daemons.

    Not inherited directly into final classes.
    """

    def __init__(self, qsize, wcount, *args, **kwargs):
        """Partial setup the new concurrent instance."""
        # maximum allowed queue size
        self.qsize = qsize or config.DAEMON.QSIZE
        # desired number of workers
        self.wcount = wcount or config.DAEMON.WORKERS
        super(_ConcurrentDaemon, self).__init__(*args, **kwargs)
{'fraction_non_alphanumeric': 0.09488220958570268, 'fraction_numerical': 0.0012997562956945572, 'mean_word_length': 2.0657370517928286, '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': '575282', 'n_tokens_mistral': 1654, 'n_tokens_neox': 1588, 'n_words': 536}