code: def call_soon_threadsafe(
      func: Callable[[], None],
      max_postpone_time: float | None = None,
      loop: asyncio.AbstractEventLoop | None = None,
  ) -> None:
      """
      Wrapper around asyncio's `call_soon_threadsafe`.
  
      This takes a `max_postpone_time` which can be used to tune the urgency of
      the method.
  
      Asyncio runs tasks in first-in-first-out. However, this is not what we
      want for the render function of the prompt_toolkit UI. Rendering is
      expensive, but since the UI is invalidated very often, in some situations
      we render the UI too often, so much that the rendering CPU usage slows down
      the rest of the processing of the application.  (Pymux is an example where
      we have to balance the CPU time spend on rendering the UI, and parsing
      process output.)
      However, we want to set a deadline value, for when the rendering should
      happen. (The UI should stay responsive).
      """
      loop2 = loop or get_running_loop()
  
      # If no `max_postpone_time` has been given, schedule right now.
      if max_postpone_time is None:
          loop2.call_soon_threadsafe(func)
          return
  
      max_postpone_until = time.time() + max_postpone_time
  
      def schedule() -> None:
          # When there are no other tasks scheduled in the event loop. Run it
          # now.
          # Notice: uvloop doesn't have this _ready attribute. In that case,
          #         always call immediately.
          if not getattr(loop2, "_ready", []):
              func()
              return
  
          # If the timeout expired, run this now.
          if time.time() > max_postpone_until:
              func()
              return
  
          # Schedule again for later.
          loop2.call_soon_threadsafe(schedule)
  
      loop2.call_soon_threadsafe(schedule)
  
file_path: catboost/contrib/python/prompt-toolkit/py3/prompt_toolkit/eventloop/utils.py
imports: from __future__ import annotations
  import asyncio
  import contextvars
  import sys
  import time
  from asyncio import get_running_loop
  from types import TracebackType
  from typing import Any, Awaitable, Callable, TypeVar, cast
label_desc: Use image node_id 2 for calling a global function with example usage: call_soon_threadsafe(func, max_postpone_time, loop) without return types