<commit_before>"""Executor util helpers."""
from __future__ import annotations

from concurrent.futures import ThreadPoolExecutor
import logging
import queue
import sys
from threading import Thread
import time
import traceback

from homeassistant.util.thread import async_raise

_LOGGER = logging.getLogger(__name__)

MAX_LOG_ATTEMPTS = 2

_JOIN_ATTEMPTS = 10

EXECUTOR_SHUTDOWN_TIMEOUT = 10


def _log_thread_running_at_shutdown(name: str, ident: int) -> None:
    """Log the stack of a thread that was still running at shutdown."""
    frames = sys._current_frames()  # pylint: disable=protected-access
    stack = frames.get(ident)
    formatted_stack = traceback.format_stack(stack)
    _LOGGER.warning(
        "Thread[%s] is still running at shutdown: %s",
        name,
        "".join(formatted_stack).strip(),
    )


def join_or_interrupt_threads(
    threads: set[Thread], timeout: float, log: bool
) -> set[Thread]:
    """Attempt to join or interrupt a set of threads."""
    joined = set()
    timeout_per_thread = timeout / len(threads)

    for thread in threads:
        thread.join(timeout=timeout_per_thread)

        if not thread.is_alive() or thread.ident is None:
            joined.add(thread)
            continue

        if log:
            _log_thread_running_at_shutdown(thread.name, thread.ident)

        async_raise(thread.ident, SystemExit)<commit_msg>Handle threads exiting unexpected during shutdown (#50907)<commit_after>"""Executor util helpers."""
from __future__ import annotations

from concurrent.futures import ThreadPoolExecutor
import contextlib
import logging
import queue
import sys
from threading import Thread
import time
import traceback

from homeassistant.util.thread import async_raise

_LOGGER = logging.getLogger(__name__)

MAX_LOG_ATTEMPTS = 2

_JOIN_ATTEMPTS = 10

EXECUTOR_SHUTDOWN_TIMEOUT = 10


def _log_thread_running_at_shutdown(name: str, ident: int) -> None:
    """Log the stack of a thread that was still running at shutdown."""
    frames = sys._current_frames()  # pylint: disable=protected-access
    stack = frames.get(ident)
    formatted_stack = traceback.format_stack(stack)
    _LOGGER.warning(
        "Thread[%s] is still running at shutdown: %s",
        name,
        "".join(formatted_stack).strip(),
    )


def join_or_interrupt_threads(
    threads: set[Thread], timeout: float, log: bool
) -> set[Thread]:
    """Attempt to join or interrupt a set of threads."""
    joined = set()
    timeout_per_thread = timeout / len(threads)

    for thread in threads:
        thread.join(timeout=timeout_per_thread)

        if not thread.is_alive() or thread.ident is None:
            joined.add(thread)
            continue

        if log:
            _log_thread_running_at_shutdown(thread.name, thread.ident)

        with contextlib.suppress(SystemError):
            # SystemError at this stage is usually a race condition
            # where the thread happens to die right before we force
            # it to raise the exception
            async_raise(thread.ident, SystemExit)
{'fraction_non_alphanumeric': 0.08366013071895424, 'fraction_numerical': 0.004901960784313725, 'mean_word_length': 3.3916786226685796, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '7462089', 'n_tokens_mistral': 912, 'n_tokens_neox': 879, 'n_words': 265}