<commit_before>import asyncio
import logging

from cached_property import cached_property

from .dispatchers import LoaferDispatcher
from .exceptions import ConfigurationError
from .routes import Route
from .runners import LoaferRunner

logger = logging.getLogger(__name__)


class LoaferManager:
    def __init__(self, routes, runner=None):
        self.runner = runner or LoaferRunner(on_stop_callback=self.on_loop__stop)
        self.routes = routes

    @cached_property
    def dispatcher(self):
        if not (self.routes and all(isinstance(r, Route) for r in self.routes)):
            raise ConfigurationError('invalid routes to dispatch, routes={}'.format(self.routes))

        return LoaferDispatcher(self.routes)

    def run(self, forever=True):
        loop = self.runner.loop
        self._future = asyncio.ensure_future(
            self.dispatcher.dispatch_providers(loop, forever=forever),
            loop=loop,
        )
        self._future.add_done_callback(self.on_future__errors)
        self.runner.start(self._future, run_forever=forever)

    #<commit_msg>Refactor: LoaferManager, add parameters to control concurrency limit and the number of executor threads.<commit_after>import asyncio
import logging

from cached_property import cached_property

from .dispatchers import LoaferDispatcher
from .exceptions import ConfigurationError
from .routes import Route
from .runners import LoaferRunner

logger = logging.getLogger(__name__)


class LoaferManager:

    def __init__(self, routes, runner=None, _concurrency_limit=None, _max_threads=None):
        self._concurrency_limit = _concurrency_limit
        if runner is None:
            self.runner = LoaferRunner(on_stop_callback=self.on_loop__stop, max_workers=_max_threads)
        else:
            self.runner = runner

        self.routes = routes

    @cached_property
    def dispatcher(self):
        if not (self.routes and all(isinstance(r, Route) for r in self.routes)):
            raise ConfigurationError('invalid routes to dispatch, routes={}'.format(self.routes))

        return LoaferDispatcher(self.routes, max_jobs=self._concurrency_limit)

    def run(self, forever=True):
        loop = self.runner.loop
        self._future = asyncio.ensure_future(
            self.dispatcher.dispatch_providers(loop, forever=forever),
            loop=loop,
        )
        self._future.add_done_callback(self.on_future__errors)
        self.runner.start(self._future, run_forever=forever)

    #
{'fraction_non_alphanumeric': 0.07964601769911504, 'fraction_numerical': 0.0, 'mean_word_length': 3.4972875226039783, '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': '6568068', 'n_tokens_mistral': 707, 'n_tokens_neox': 704, 'n_words': 168}