class_docstr: Standard OCRmyPDF concurrent task executor.
class_docstr_tok: ["Standard","OCRmyPDF","concurrent","task","executor","."]
code: def _execute(
      self,
      *,
      use_threads: bool,
      max_workers: int,
      progress_kwargs: dict,
      worker_initializer: Callable,
      task: Callable,
      task_arguments: Iterable,
      task_finished: Callable,
  ):
      if use_threads:
          log_queue: Queue = queue.Queue(-1)
          executor_class: FuturesExecutorClass = ThreadPoolExecutor
          initializer: WorkerInit = thread_init
      else:
          log_queue = multiprocessing.Queue(-1)
          executor_class = ProcessPoolExecutor
          initializer = process_init
  
      # Regardless of whether we use_threads for worker processes, the log_listener
      # must be a thread. Make sure we create the listener after the worker pool,
      # so that it does not get forked into the workers.
      # If use_threads is False, we are currently guilty of creating a thread before
      # forking on Linux, which is not recommended. However, we take a big
      # performance hit in pdfinfo if we can't fork. Long term solution is to
      # replace most of this with an asyncio implementation, and probably to
      # migrate some of pdfinfo into C++ or Rust.
      listener = threading.Thread(
          target=log_listener, args=(log_queue,)
      )
      listener.start()
  
      with self.pbar_class(**progress_kwargs) as pbar, executor_class(
          max_workers=max_workers,
          initializer=initializer,
          initargs=(
              log_queue,
              worker_initializer,
              logging.getLogger("").level,
          ),
      ) as executor:
          futures = [
              executor.submit(task, *args) for args in task_arguments
          ]
          try:
              for future in as_completed(futures):
                  result = future.result()
                  task_finished(result, pbar)
          except KeyboardInterrupt:
              # Terminate pool so we exit instantly
              executor.shutdown(wait=False, cancel_futures=True)
              raise
          except Exception:
              if not os.environ.get("PYTEST_CURRENT_TEST", ""):
                  # Normally we shutdown without waiting for other child workers
                  # on error, because there is no point in waiting for them. Their
                  # results will be discard. But if the condition above is True,
                  # then we are running in pytest, and we want everything to exit
                  # as cleanly as possible so that we get good error messages.
                  executor.shutdown(wait=False, cancel_futures=True)
              raise
          finally:
              # Terminate log listener
              log_queue.put_nowait(None)
  
      # When the above succeeds, wait for the listener thread to exit. (If
      # an exception occurs, we don't try to join, in case it deadlocks.)
      listener.join()
  
imports: from __future__ import annotations
  import logging
  import logging.handlers
  import multiprocessing
  import os
  import queue
  import signal
  import sys
  import threading
  from collections.abc import Iterable
  from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed
  from contextlib import suppress
  from typing import Callable, Union
  from rich.console import Console
  from ocrmypdf import Executor, hookimpl
  from ocrmypdf._logging import RichLoggingHandler
  from ocrmypdf._progressbar import RichProgressBar
  from ocrmypdf.exceptions import InputFileError
  from ocrmypdf.helpers import remove_all_log_handlers
label_desc: Use image node_id 1 for calling the StandardExecutor obj's underlying member method code with example usage: obj._execute() without return types