<commit_before>import logging
from multiprocessing.pool import ThreadPool
from multiprocessing import cpu_count
from threading import local, current_thread
from inspect import getmembers, ismethod, isfunction, getargspec
from functools import partial


_DOC_STRING = '''
    - factory and factory_args (Optional):
    Factory function and its arguments to create thread-local instances of 
    the thread-unsafe class.
    These only need to be provided if the thread-unsafe class has a non-empty constructor, 
    as instances of it are lazily and locally created in each thread/process.

    - pool (Optional):
    The pool of workers to parallelize processing on.
    It needs to conform to multiprocessing.Pool's API.
    If no pool is provided, default is a thread-pool (multiprocessing.pool.ThreadPool), 
    which is only recommended for I/O-bound processes, as the GIL prevents true parallelism.
'''

def concurrent_class_factory(concurrent_class_name, thread_unsafe_class):
    ''' 
    Dynamically creates a thread-safe class which exposes the same API as 
    the provided thread-unsafe class, but for which all public methods now:
    - accept an iterable of inputs, 
    - parallelize processing of this iterable on a pool of workers.

    The resulting class's __init__ method optionally accepts: %s
    '''
    concurrent_class = _create_blank_concurrent_class(concurrent_class_name, thread_unsafe_class)
    for method_name, method in getmembers(thread_unsafe_class, ismethod):
        if not method_name.startswith('_'):
            setattr(concurrent_class, method_name, _mapper_for(method, method_name))
    return concurrent_class

concurrent_class_factory.__doc__ = concurrent_class_factory.__doc__ % _DOC_STRING


def _create_blank_concurrent_class(concurrent_class_name, thread_unsafe_class):
    def __init__(self, factory=None, factory_args=None, pool=None):
        self._local = local()
        self._pool = pool or _pool_for_io_bound_workload(self._local, factory or thread_unsafe_class, factory_args)
    __init__.__doc__ = _DOC_STRING
    return type(concurrent_class_name, (object,), {'__init__': __init__})

def _pool_for_io_bound_workload(thread_local, factory, factory_args, num_threads=4*cpu_count()):
    initializer = partial(_get_or_create_object, thread_local, factory)
    if factory_args is None:
        return ThreadPool(processes=num_threads, initializer=initializer)<commit_msg>Improved getmembers' predicate to work with Python3: 'The concept of “unbound methods” has been removed from the language. When referencing a method as a class attribute, you now get a plain function object.'<commit_after>import logging
from multiprocessing.pool import ThreadPool
from multiprocessing import cpu_count
from threading import local, current_thread
from inspect import getmembers, ismethod, isfunction, getargspec
from functools import partial


_DOC_STRING = '''
    - factory and factory_args (Optional):
    Factory function and its arguments to create thread-local instances of 
    the thread-unsafe class.
    These only need to be provided if the thread-unsafe class has a non-empty constructor, 
    as instances of it are lazily and locally created in each thread/process.

    - pool (Optional):
    The pool of workers to parallelize processing on.
    It needs to conform to multiprocessing.Pool's API.
    If no pool is provided, default is a thread-pool (multiprocessing.pool.ThreadPool), 
    which is only recommended for I/O-bound processes, as the GIL prevents true parallelism.
'''

def concurrent_class_factory(concurrent_class_name, thread_unsafe_class):
    ''' 
    Dynamically creates a thread-safe class which exposes the same API as 
    the provided thread-unsafe class, but for which all public methods now:
    - accept an iterable of inputs, 
    - parallelize processing of this iterable on a pool of workers.

    The resulting class's __init__ method optionally accepts: %s
    '''
    concurrent_class = _create_blank_concurrent_class(concurrent_class_name, thread_unsafe_class)
    for method_name, method in getmembers(thread_unsafe_class, lambda x: ismethod(x) or isfunction(x)):
        if not method_name.startswith('_'):
            setattr(concurrent_class, method_name, _mapper_for(method, method_name))
    return concurrent_class

concurrent_class_factory.__doc__ = concurrent_class_factory.__doc__ % _DOC_STRING


def _create_blank_concurrent_class(concurrent_class_name, thread_unsafe_class):
    def __init__(self, factory=None, factory_args=None, pool=None):
        self._local = local()
        self._pool = pool or _pool_for_io_bound_workload(self._local, factory or thread_unsafe_class, factory_args)
    __init__.__doc__ = _DOC_STRING
    return type(concurrent_class_name, (object,), {'__init__': __init__})

def _pool_for_io_bound_workload(thread_local, factory, factory_args, num_threads=4*cpu_count()):
    initializer = partial(_get_or_create_object, thread_local, factory)
    if factory_args is None:
        return ThreadPool(processes=num_threads, initializer=initializer)
{'fraction_non_alphanumeric': 0.0627095247485703, 'fraction_numerical': 0.0005915992900808519, 'mean_word_length': 4.705286839145107, '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': '312218', 'n_tokens_mistral': 1439, 'n_tokens_neox': 1376, 'n_words': 475}