class_docstr: A class the contains a naive implementation for concurrent sub-process
  execution. It uses `subprocess.Popen` and threads to read from stdout and
  stderr of the subprocess, and to write to stdin of the subprocess.
  
  All read data and timeouts are passed to a protocol instance, which can
  create the final result.
class_docstr_tok: ["A","class","the","contains","a","naive","implementation","for","concurrent","sub-process","execution",".","It","uses","`","subprocess.Popen","`","and","threads","to","read","from","stdout","and","stderr","of","the","subprocess",",","and","to","write","to","stdin","of","the","subprocess",".","All","read","data","and","timeouts","are","passed","to","a","protocol","instance",",","which","can","create","the","final","result","."]
code: def _locked_run(self) -> dict | _ResultGenerator:
      with self.generator_condition:
          if self.generator is not None:
              if self.owning_thread == threading.get_ident():
                  raise RuntimeError(
                      "ThreadedRunner.run() was re-entered by already owning "
                      f"thread {threading.get_ident()}. The execution is "
                      f"still owned by thread {self.owning_thread}"
                  )
              self.generator_condition.wait()
              assert self.generator is None
  
      if isinstance(self.stdin, (int, IO, type(None))):
          # We will not write anything to stdin. If the caller passed a
          # file-like he can write to it from a different thread.
          self.write_stdin = False
  
      elif isinstance(self.stdin, bytes):
          # Establish a queue to write to the process and
          # enqueue the input that is already provided.
          self.write_stdin = True
          self.stdin_queue = Queue()
          self.stdin_queue.put(self.stdin)
          self.stdin_queue.put(None)
  
      elif isinstance(self.stdin, Queue):
          # Establish a queue to write to the process.
          self.write_stdin = True
          self.stdin_queue = self.stdin
  
      else:
          # We do not recognize the input class will and just pass is through
          # to Popen(). We assume that the caller handles any writing if
          # desired.
          self.write_stdin = False
  
      self.protocol = self.protocol_class(**self.protocol_kwargs)
  
      # The following command is generated internally by datalad
      # and trusted. Security check is therefore skipped.
      kwargs = {
          **self.popen_kwargs,
          **dict(
              bufsize=0,
              stdin=subprocess.PIPE if self.write_stdin else self.stdin,
              stdout=subprocess.PIPE if self.catch_stdout else None,
              stderr=subprocess.PIPE if self.catch_stderr else None,
              shell=True
              if isinstance(self.cmd, str)
              else False,  # nosec
          ),
      }
  
      if self.process is not None:
          raise RuntimeError(
              f"Process already running {self.process.pid}"
          )
  
      self.return_code = None
      try:
          # The following command is generated internally by datalad
          # and trusted. Security check is therefore skipped.
          self.process = Popen(self.cmd, **kwargs)  # nosec
  
      except OSError as e:
          if (
              not on_windows
              and "argument list too long" in str(e).lower()
          ):
              lgr.error(
                  "Caught exception suggesting too large stack size limits. "
                  "Hint: use 'ulimit -s' command to see current limit and "
                  "e.g. 'ulimit -s 8192' to reduce it to avoid this "
                  "exception. See "
                  "https://github.com/datalad/datalad/issues/6106 for more "
                  "information."
              )
          raise
  
      self.process_running = True
      self.active_file_numbers.add(None)
  
      self.process_stdin_fileno = _get_fileno(
          self.write_stdin, self.process.stdin
      )
      self.process_stdout_fileno = _get_fileno(
          self.catch_stdout, self.process.stdout
      )
      self.process_stderr_fileno = _get_fileno(
          self.catch_stderr, self.process.stderr
      )
  
      # We pass process as transport-argument. It does not have the same
      # semantics as the asyncio-signature, but since it is only used in
      # WitlessProtocol, all necessary changes can be made there.
      self.protocol.connection_made(self.process)
  
      # Map the pipe file numbers to stdout and stderr file number, because
      # the latter are hardcoded in the protocol code
      self.fileno_mapping = {
          self.process_stdout_fileno: STDOUT_FILENO,
          self.process_stderr_fileno: STDERR_FILENO,
          self.process_stdin_fileno: STDIN_FILENO,
      }
      if None in self.fileno_mapping:
          self.fileno_mapping.pop(None)
  
      self.fileno_to_file = {
          self.process_stdout_fileno: self.process.stdout,
          self.process_stderr_fileno: self.process.stderr,
          self.process_stdin_fileno: self.process.stdin,
      }
      if None in self.fileno_to_file:
          self.fileno_to_file.pop(None)
  
      self.file_to_fileno = {
          f: f.fileno()
          for f in (
              self.process.stdout,
              self.process.stderr,
              self.process.stdin,
          )
          if f is not None
      }
  
      current_time = time.time()
      if self.timeout:
          self.last_touched[None] = current_time
  
      cmd_string = (
          self.cmd if isinstance(self.cmd, str) else " ".join(self.cmd)
      )
      if self.catch_stderr:
          if self.timeout:
              self.last_touched[
                  self.process_stderr_fileno
              ] = current_time
          self.active_file_numbers.add(self.process_stderr_fileno)
          self.last_touched[self.process_stderr_fileno] = current_time
          assert self.process.stderr is not None
          self.stderr_enqueueing_thread = ReadThread(
              identifier="STDERR: " + cmd_string[:20],
              signal_queues=[self.output_queue],
              user_info=self.process_stderr_fileno,
              source=self.process.stderr,
              destination_queue=self.output_queue,
          )
          self.stderr_enqueueing_thread.start()
  
      if self.catch_stdout:
          if self.timeout:
              self.last_touched[
                  self.process_stdout_fileno
              ] = current_time
          self.active_file_numbers.add(self.process_stdout_fileno)
          self.last_touched[self.process_stdout_fileno] = current_time
          assert self.process.stdout is not None
          self.stdout_enqueueing_thread = ReadThread(
              identifier="STDOUT: " + cmd_string[:20],
              signal_queues=[self.output_queue],
              user_info=self.process_stdout_fileno,
              source=self.process.stdout,
              destination_queue=self.output_queue,
          )
          self.stdout_enqueueing_thread.start()
  
      if self.write_stdin:
          # No timeouts for stdin
          self.active_file_numbers.add(self.process_stdin_fileno)
          assert self.stdin_queue is not None
          assert self.process.stdin is not None
          self.stdin_enqueueing_thread = WriteThread(
              identifier="STDIN: " + cmd_string[:20],
              user_info=self.process_stdin_fileno,
              signal_queues=[self.output_queue],
              source_queue=self.stdin_queue,
              destination=self.process.stdin,
          )
          self.stdin_enqueueing_thread.start()
  
      self.process_waiting_thread = WaitThread(
          "process_waiter", [self.output_queue], self.process
      )
      self.process_waiting_thread.start()
  
      if isinstance(self.protocol, GeneratorMixIn):
          self.generator = _ResultGenerator(
              self, self.protocol.result_queue
          )
          self.owning_thread = threading.get_ident()
          return self.generator
  
      return self.process_loop()
  
imports: from __future__ import annotations
  import enum
  import logging
  import subprocess
  import threading
  import time
  from collections import deque
  from collections.abc import Generator
  from queue import Empty, Queue
  from subprocess import Popen
  from typing import IO, Any, Optional
  from datalad.utils import on_windows
  from .exception import CommandError
  from .protocol import GeneratorMixIn, WitlessProtocol
  from .runnerthreads import IOState, ReadThread, WaitThread, WriteThread, _try_close
label_desc: Use image node_id 4 for calling the ThreadedRunner obj's underlying member method code with example usage: obj._locked_run() and returns: self, self