<commit_before># -*- coding: utf-8 -*-

from __future__ import absolute_import

import weakref
import warnings
import logging
import threading

from six.moves import queue
from .dispatcher import Service as Dispatcher
from ..base import ClientService, ClientKeyword
from ..base.core import Task as _BaseTask
from ..exc import CauldronAPINotImplementedWarning, CauldronAPINotImplemented, ServiceNotStarted, DispatcherError, TimeoutError
from .. import registry

__all__ = ['Service', 'Keyword']


class LocalTask(_BaseTask):
    """A simple object to mock asynchronous operations."""
    
    def __call__(self):
        """Make sure that errors are properly set."""
        super(LocalTask, self).__call__()
        if self.error is not None:
            self.error = DispatcherError(str(self.error))
        if self.exc_info is not None:
            self.exc_info = (DispatcherError, self.error, self.exc_info[2])
    
class LocalTaskQueue(threading.Thread):
    
    def __init__(self, name, log=None):
        super(LocalTaskQueue, self).__init__(name="Task Queue for {0:s}".format(name))
        self.queue = queue.Queue()
        self.log = log or logging.getLogger("ktl.local.TaskQueue.{0:s}".format(name))
        self.shutdown = threading.Event()
    
    def run(self):
        """Run the task queue thread."""
        while not self.shutdown.isSet():
            try:
                task = self.queue.get()
                if task is None:<commit_msg>Fix error raising from local dispatchers?<commit_after># -*- coding: utf-8 -*-

from __future__ import absolute_import

import weakref
import warnings
import logging
import threading

from six.moves import queue
from .dispatcher import Service as Dispatcher
from ..base import ClientService, ClientKeyword
from ..base.core import Task as _BaseTask
from ..exc import CauldronAPINotImplementedWarning, CauldronAPINotImplemented, ServiceNotStarted, DispatcherError, TimeoutError
from .. import registry

__all__ = ['Service', 'Keyword']

class LocalTask(_BaseTask):
    """A simple object to mock asynchronous operations."""
    
    def reraise(self):
        """Wrap raised errors in dispatcher errors."""
        if self.error is not None:
            self.error = DispatcherError(str(self.error))
        if self.exc_info is not None:
            self.exc_info = (DispatcherError, self.error, self.exc_info[2])
        super(LocalTask, self).reraise()
    
class LocalTaskQueue(threading.Thread):
    
    def __init__(self, name, log=None):
        super(LocalTaskQueue, self).__init__(name="Task Queue for {0:s}".format(name))
        self.queue = queue.Queue()
        self.log = log or logging.getLogger("ktl.local.TaskQueue.{0:s}".format(name))
        self.shutdown = threading.Event()
    
    def run(self):
        """Run the task queue thread."""
        while not self.shutdown.isSet():
            try:
                task = self.queue.get()
                if task is None:
{'fraction_non_alphanumeric': 0.10610169491525424, 'fraction_numerical': 0.002711864406779661, 'mean_word_length': 3.0204359673024523, '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': '778240', 'n_tokens_mistral': 855, 'n_tokens_neox': 836, 'n_words': 247}