class_docstr: Executor for AWS Lambda or similar environments that lack semaphores.
class_docstr_tok: ["Executor","for","AWS","Lambda","or","similar","environments","that","lack","semaphores","."]
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 and max_workers == 1:
          with self.pbar_class(**progress_kwargs) as pbar:
              for args in task_arguments:
                  result = task(*args)
                  task_finished(result, pbar)
          return
  
      task_arguments = list(task_arguments)
      grouped_args = list(
          zip_longest(*list(split_every(max_workers, task_arguments)))
      )
      if not grouped_args:
          return
  
      processes: list[Process] = []
      connections: list[Connection] = []
      for chunk in grouped_args:
          parent_conn, child_conn = Pipe()
  
          worker_args = [args for args in chunk if args is not None]
          process = Process(
              target=process_loop,
              args=(
                  child_conn,
                  worker_initializer,
                  logging.getLogger("").level,
                  task,
                  worker_args,
              ),
          )
          process.daemon = True
          processes.append(process)
          connections.append(parent_conn)
  
      for process in processes:
          process.start()
  
      with self.pbar_class(**progress_kwargs) as pbar:
          while connections:
              for result in wait(connections):
                  if not isinstance(result, Connection):
                      raise NotImplementedError(
                          "We only support Connection()"
                      )
                  try:
                      msg_type, msg = result.recv()
                  except EOFError:
                      connections.remove(result)
                      continue
  
                  if msg_type == MessageType.result:
                      task_finished(msg, pbar)
                  elif msg_type == "log":
                      record = msg
                      logger = logging.getLogger(record.name)
                      logger.handle(record)
                  elif msg_type == MessageType.complete:
                      connections.remove(result)
                  elif msg_type == MessageType.exception:
                      for process in processes:
                          process.terminate()
                      raise msg
  
      for process in processes:
          process.join()
  
imports: from __future__ import annotations
  import logging
  import logging.handlers
  import signal
  from collections.abc import Iterable, Iterator
  from contextlib import suppress
  from enum import Enum, auto
  from itertools import islice, repeat, takewhile, zip_longest
  from multiprocessing import Pipe, Process
  from multiprocessing.connection import Connection, wait
  from typing import Callable
  from ocrmypdf import Executor, hookimpl
  from ocrmypdf._concurrent import NullProgressBar
  from ocrmypdf.exceptions import InputFileError
  from ocrmypdf.helpers import remove_all_log_handlers
label_desc: Use image node_id 1 for calling the LambdaExecutor obj's underlying member method code with example usage: obj._execute() without return types