"""Reproduces a free-threaded CPython 3.14 SIGSEGV in `_asyncio_all_tasks`.

Spawner threads repeatedly start a short-lived worker thread that cancels the
asyncio Tasks left behind by its predecessor, creates a new batch of pending
Tasks, and exits. When a thread exits with pending Tasks, PyThreadState_Clear()
moves them onto the interpreter-wide `interp->asyncio_tasks_head` list. glibc
hands the exited thread's pthread_t to the next thread, so the successor has the
same `_Py_ThreadId()` as the creator of those Tasks. unregister_task() then
takes
its lock-free fast path and unlinks them from the shared list without stopping
the world, and concurrent unlinks corrupt the list.

Runner threads call asyncio.run(). Its shutdown calls asyncio.all_tasks() after
the loop stopped running, which takes the stop-the-world slow path in
_asyncio_all_tasks() and walks the corrupted list until it dereferences NULL.

If the runtime is not affected, the binary runs for --duration_secs and exits 0.
"""

import asyncio
import collections
from collections.abc import Sequence
import sys
import sysconfig
import threading
from typing import Any

from absl import app
from absl import flags
from absl import logging

_DURATION_SECS = flags.DEFINE_float(
    "duration_secs", 30.0, "How long to run before declaring a pass."
)
_SPAWNER_THREADS = flags.DEFINE_integer(
    "spawner_threads", 6, "Threads that churn short-lived Task-owning threads."
)
_RUNNER_THREADS = flags.DEFINE_integer(
    "runner_threads", 2, "Threads that call asyncio.run() in a loop."
)
_TASKS_PER_THREAD = flags.DEFINE_integer(
    "tasks_per_thread", 32, "Pending Tasks each short-lived thread leaves."
)

_TaskBatch = list[asyncio.Task[Any]]


class _InertLoop:
  """Minimal event loop stand-in for constructing `asyncio.Task` objects.

  `call_soon` drops the Task's first step, so a Task stays pending without a
  real event loop holding a reference to it or opening file descriptors.
  """

  def get_debug(self) -> bool:
    return False

  def call_soon(self, *args: object, **kwargs: object) -> None:
    del args, kwargs  # Unused.


_INERT_LOOP = _InertLoop()

# Keeps recently cancelled Tasks alive so their memory is not reused while a
# corrupted list may still point at them. This turns a nondeterministic
# use-after-free into the deterministic NULL dereference seen in production.
_RECENTLY_CANCELLED: collections.deque[_TaskBatch] = collections.deque(
    maxlen=4096
)


async def _noop() -> None:
  pass


def _create_pending_tasks(count: int) -> _TaskBatch:
  """Creates `count` Tasks that stay pending until they are cancelled."""
  tasks = []
  for _ in range(count):
    coro = _noop()
    task = asyncio.Task(coro, loop=_INERT_LOOP)  # type: ignore[arg-type]
    # The Task never runs: close its coroutine to avoid "never awaited", and do
    # not warn about Tasks that are still pending when the process exits.
    coro.close()
    setattr(task, "_log_destroy_pending", False)
    tasks.append(task)
  return tasks


def _short_lived_worker(
    handoff: list[_TaskBatch], tasks_per_thread: int
) -> None:
  """Cancels the predecessor's Tasks, then exits leaving new pending Tasks."""
  while handoff:
    tasks = handoff.pop()
    for task in tasks:
      # Future.cancel() completes the Task immediately, which unregisters it
      # from the asyncio task list (Task.cancel() would only request it).
      asyncio.Future.cancel(task)
    _RECENTLY_CANCELLED.append(tasks)
  handoff.append(_create_pending_tasks(tasks_per_thread))


def _spawner_loop(stop: threading.Event, tasks_per_thread: int) -> None:
  """Starts and joins short-lived worker threads until `stop` is set."""
  handoff: list[_TaskBatch] = []
  while not stop.is_set():
    worker = threading.Thread(
        target=_short_lived_worker, args=(handoff, tasks_per_thread)
    )
    worker.start()
    worker.join()


def _runner_loop(stop: threading.Event) -> None:
  """Runs short asyncio.run() calls until `stop` is set."""
  while not stop.is_set():
    asyncio.run(_noop())


def main(argv: Sequence[str]) -> None:
  if len(argv) > 1:
    raise app.UsageError("Too many command-line arguments.")
  if not sysconfig.get_config_var("Py_GIL_DISABLED"):
    raise app.UsageError("Needs a free-threaded Python build; see README.md.")
  if sys._is_gil_enabled():  # pylint: disable=protected-access
    raise app.UsageError("The GIL is enabled; run with PYTHON_GIL=0.")
  logging.info("Running on Python %s with the GIL disabled.", sys.version)

  stop = threading.Event()
  threads = [
      threading.Thread(
          target=_spawner_loop, args=(stop, _TASKS_PER_THREAD.value)
      )
      for _ in range(_SPAWNER_THREADS.value)
  ]
  threads.extend(
      threading.Thread(target=_runner_loop, args=(stop,))
      for _ in range(_RUNNER_THREADS.value)
  )
  for thread in threads:
    thread.start()
  stop.wait(_DURATION_SECS.value)
  stop.set()
  for thread in threads:
    thread.join()
  logging.info("PASS: no crash after %.1f seconds.", _DURATION_SECS.value)


if __name__ == "__main__":
  app.run(main)