<commit_before>#! coding:utf-8
# python2 requires: pip install futures

import time
from concurrent.futures import (ProcessPoolExecutor, ThreadPoolExecutor,
                                as_completed, wait)
from concurrent.futures._base import Executor, Future, TimeoutError
from concurrent.futures.thread import _WorkItem
from functools import wraps
from weakref import WeakSet

from requests import Session
from requests.adapters import HTTPAdapter

from .configs import Config
from .exceptions import FailureException
from .versions import PY2, PY3

if PY3:
    from concurrent.futures.process import BrokenProcessPool


def get_cpu_count():
    try:
        from multiprocessing import cpu_count
        return cpu_count()
    except Exception as e:
        Config.main_logger.error('get_cpu_count failed for %s' % e)


class NewExecutorPool(Executor):
    """add async_func(function decorator) for submitting called-function into NewExecutorPool obj."""

    def __init__(self, max_workers=None, timeout=None, default_callback=None):

        # trace the next node of NewExecutorPool.__mro__
        super(NewExecutorPool, self).__init__(max_workers)
        self._timeout = timeout
        self.default_callback = default_callback
        self._all_futures = WeakSet()

    def async_func(self, function):

        @wraps(function)
        def wrapped(*args, **kwargs):
            return self.submit(function, *args, **kwargs)

        return wrapped

    def close(self, wait=True):
        self.shutdown(wait=wait)

    @property
    def x(self):
        return self.wait_futures_done(list(self._all_futures))

    def wait_futures_done(self, tasks=None):
        # ignore the order of tasks
        tasks = tasks or self._all_futures
        fs = {f.x for f in wait(tasks).done}
        return fs


class Pool(NewExecutorPool, ThreadPoolExecutor):

    def __init__(self, n=None, *args, **kwargs):
        n = n or kwargs.pop('max_workers', None)
        if PY2 and n is None:
            # python2 n!=None
            n = (get_cpu_count() or 1) * 5
        super(Pool, self).__init__(n, *args, **kwargs)

    def submit(self, func, *args, **kwargs):
        """self.submit(function,arg1,arg2,arg3=3)"""

        with self._shutdown_lock:
            if self._shutdown:
                raise RuntimeError('cannot schedule new futures after shutdown')
            future = NewFuture(self._timeout, args, kwargs)
            callback = kwargs.pop('callback', self.default_callback)
            if callback:
                if not isinstance(callback, (list, tuple)):
                    callback = [callback]
                for fn in callback:
                    future.add_done_callback(future.wrap_callback(fn))
            w = _WorkItem(future, func, args, kwargs)
            self._work_queue.put(w)
            self._adjust_thread_count()
            self._all_futures.add(future)
            return future


class ProcessPool(NewExecutorPool, ProcessPoolExecutor):

    def __init__(self, n=None, *args, **kwargs):
        n = n or kwargs.pop('max_workers', None)
        if PY2 and n is None:
            # python2 n!=None
            n = get_cpu_count() or 1
        super(ProcessPool, self).__init__(n, *args, **kwargs)

    def submit(self, func, *args, **kwargs):<commit_msg>recoding for NewExecutorPoolMixin to simplify __mro__<commit_after>#! coding:utf-8
# python2 requires: pip install futures

import time
from concurrent.futures import (ProcessPoolExecutor, ThreadPoolExecutor,
                                as_completed, wait)
from concurrent.futures._base import Executor, Future, TimeoutError
from concurrent.futures.thread import _WorkItem
from functools import wraps
from weakref import WeakSet

from requests import Session
from requests.adapters import HTTPAdapter

from .configs import Config
from .exceptions import FailureException
from .versions import PY2, PY3

if PY3:
    from concurrent.futures.process import BrokenProcessPool


def get_cpu_count():
    try:
        from multiprocessing import cpu_count
        return cpu_count()
    except Exception as e:
        Config.main_logger.error('get_cpu_count failed for %s' % e)


class NewExecutorPoolMixin(Executor):
    """add async_func(function decorator) for submitting."""

    def async_func(self, function):

        @wraps(function)
        def wrapped(*args, **kwargs):
            return self.submit(function, *args, **kwargs)

        return wrapped

    def close(self, wait=True):
        self.shutdown(wait=wait)

    @property
    def x(self):
        return self.wait_futures_done(list(self._all_futures))

    def wait_futures_done(self, tasks=None):
        # ignore the order of tasks
        tasks = tasks or self._all_futures
        fs = {f.x for f in wait(tasks).done}
        return fs


class Pool(ThreadPoolExecutor, NewExecutorPoolMixin):

    def __init__(self,
                 n=None,
                 timeout=None,
                 default_callback=None,
                 *args,
                 **kwargs):
        n = n or kwargs.pop('max_workers', None)
        if PY2 and n is None:
            # python2 n!=None
            n = (get_cpu_count() or 1) * 5
        super(Pool, self).__init__(n, *args, **kwargs)
        self._timeout = timeout
        self.default_callback = default_callback
        self._all_futures = WeakSet()

    def submit(self, func, *args, **kwargs):
        """self.submit(function,arg1,arg2,arg3=3)"""

        with self._shutdown_lock:
            if self._shutdown:
                raise RuntimeError('cannot schedule new futures after shutdown')
            future = NewFuture(self._timeout, args, kwargs)
            callback = kwargs.pop('callback', self.default_callback)
            if callback:
                if not isinstance(callback, (list, tuple)):
                    callback = [callback]
                for fn in callback:
                    future.add_done_callback(future.wrap_callback(fn))
            w = _WorkItem(future, func, args, kwargs)
            self._work_queue.put(w)
            self._adjust_thread_count()
            self._all_futures.add(future)
            return future


class ProcessPool(ProcessPoolExecutor, NewExecutorPoolMixin):

    def __init__(self,
                 n=None,
                 timeout=None,
                 default_callback=None,
                 *args,
                 **kwargs):
        n = n or kwargs.pop('max_workers', None)
        if PY2 and n is None:
            # python2 n!=None
            n = get_cpu_count() or 1
        super(ProcessPool, self).__init__(n, *args, **kwargs)
        self._timeout = timeout
        self.default_callback = default_callback
        self._all_futures = WeakSet()

    def submit(self, func, *args, **kwargs):
{'fraction_non_alphanumeric': 0.09031877213695395, 'fraction_numerical': 0.004722550177095631, 'mean_word_length': 2.448854961832061, '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': '2597290', 'n_tokens_mistral': 1961, 'n_tokens_neox': 1890, 'n_words': 523}