<commit_before>__all__ = [
    'Executor',
]

import itertools
import logging
import os

from g1.threads import actors
from g1.threads import futures
from g1.threads import queues

LOG = logging.getLogger(__name__)


class Executor:

    _COUNTER = itertools.count(1).__next__

    def __init__(
        self,
        max_executors=0,
        *,
        name_prefix='',
        daemon=None,
    ):

        if max_executors <= 0:
            # Use this because Executor is often used to parallelize I/O
            # instead of computationally-heavy tasks.
            max_executors = max(os.cpu_count(), 1) * 8

        if not name_prefix:
            names = (
                'executor-%02d' % self._COUNTER()
                for _ in range(max_executors)
            )
        else:
            names = (
                '%s-%02d' % (name_prefix, i) for i in range(max_executors)
            )

        self.queue = queues.Queue()
        self.stubs = tuple(
            actors.Stub(
                name=name,
                actor=actors.function_caller,
                queue=self.queue,
                daemon=daemon,
            ) for name in names
        )

    def __enter__(self):
        return self

    def __exit__(self, exc_type, *_):
        self.shutdown(graceful=not exc_type)

    def submit(self, func, *args, **kwargs):
        future = futures.Future()
        call = actors.MethodCall(
            method=func, args=args, kwargs=kwargs, future=future
        )
        self.queue.put(call)
        return future<commit_msg>Fix executor blocking main process exit<commit_after>__all__ = [
    'Executor',
]

import itertools
import logging
import os
import weakref

from g1.threads import actors
from g1.threads import futures
from g1.threads import queues

LOG = logging.getLogger(__name__)


class Executor:

    _COUNTER = itertools.count(1).__next__

    def __init__(
        self,
        max_executors=0,
        *,
        name_prefix='',
        daemon=None,
    ):

        if max_executors <= 0:
            # Use this because Executor is often used to parallelize I/O
            # instead of computationally-heavy tasks.
            max_executors = max(os.cpu_count(), 1) * 8

        if not name_prefix:
            names = (
                'executor-%02d' % self._COUNTER()
                for _ in range(max_executors)
            )
        else:
            names = (
                '%s-%02d' % (name_prefix, i) for i in range(max_executors)
            )

        self.queue = queues.Queue()
        self.stubs = tuple(
            actors.Stub(
                name=name,
                actor=actors.function_caller,
                queue=self.queue,
                daemon=daemon,
            ) for name in names
        )

        # Add this ``finalize`` so that, when the application does not
        # shut down the executor and did not set daemon to true, the
        # actor threads (and then the main process) could still exit.
        weakref.finalize(self, self.queue.close)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, *_):
        self.shutdown(graceful=not exc_type)

    def submit(self, func, *args, **kwargs):
        future = futures.Future()
        call = actors.MethodCall(
            method=func, args=args, kwargs=kwargs, future=future
        )
        self.queue.put(call)
        return future
{'fraction_non_alphanumeric': 0.09002647837599294, 'fraction_numerical': 0.00706090026478376, 'mean_word_length': 1.7463651050080775, 'pattern_counts': {'":': 0, '<': 5, '<?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': '7471050', 'n_tokens_mistral': 1023, 'n_tokens_neox': 977, 'n_words': 258}