<commit_before>import typing as tp
from concurrent.futures.thread import ThreadPoolExecutor

from satella.coding.typing import V, U

try:
    import opentracing
except ImportError:
    opentracing = None


def parallel_construct(iterable: tp.Iterable[V],
                       function: tp.Callable[[V], tp.Optional[U]],
                       thread_pool: ThreadPoolExecutor) -> tp.List[U]:
    """
    Construct a list from executing given function in a thread pool executor.

    If opentracing is installed, and tracing is enabled, current span will be passed to child threads.

    :param iterable: iterable to apply
    :param function: function to apply. If that function returns None, no element will be added
    :param thread_pool: thread pool to execute
    :return: list that is the result of parallel application of function on each element
    """
    wrap_iterable = None

    if opentracing is not None:
        tracer = opentracing.global_tracer()
        span = tracer.active_span
        if span is not None:
            current_span = tracer.active_span

            def wrap_iterable(arg, *args,  **kwargs):
                scope = tracer.scope_manager.activate(current_span, finish_on_close=False)
                v = function(arg, *args, **kwargs)
                scope.close()
                return v

    if wrap_iterable is None:
        wrap_iterable = function

    result = []
    for item in thread_pool.map(wrap_iterable, iterable):
        if item is not None:
            result.append(item)
    return result
<commit_msg>close the scope upon creation<commit_after>import typing as tp
from concurrent.futures.thread import ThreadPoolExecutor

from satella.coding.typing import V, U

try:
    import opentracing
except ImportError:
    opentracing = None


def parallel_construct(iterable: tp.Iterable[V],
                       function: tp.Callable[[V], tp.Optional[U]],
                       thread_pool: ThreadPoolExecutor,
                       span_title: tp.Optional[str] = None) -> tp.List[U]:
    """
    Construct a list from executing given function in a thread pool executor.

    If opentracing is installed, and tracing is enabled, current span will be passed to child threads.

    :param iterable: iterable to apply
    :param function: function to apply. If that function returns None, no element will be added
    :param thread_pool: thread pool to execute
    :param span_title: span title to create. For each execution a child span will be returned
    :return: list that is the result of parallel application of function on each element
    """
    wrap_iterable = None

    if opentracing is not None:
        tracer = opentracing.global_tracer()
        span = tracer.active_span
        if span is not None:

            def wrap_iterable(arg, *args,  **kwargs):
                with tracer.start_active_span(span_title or 'New span', child_of=span):
                    return function(arg, *args, **kwargs)

    if wrap_iterable is None:
        wrap_iterable = function

    result = []
    for item in thread_pool.map(wrap_iterable, iterable):
        if item is not None:
            result.append(item)
    return result

{'fraction_non_alphanumeric': 0.06840288672732978, 'fraction_numerical': 0.0, 'mean_word_length': 2.5780022446689115, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '3822318', 'n_tokens_mistral': 871, 'n_tokens_neox': 855, 'n_words': 318}