Function checkpoint

src/trio/_core/_run.py:3026–3061

A pure :ref:`checkpoint `. This checks for cancellation and allows other tasks to be scheduled, without otherwise blocking. Note that the scheduler has the option of ignoring this and continuing to run the current task if it decides this is appropriate (e.g. for increa

()

async def checkpoint() -> None:
    """A pure :ref:`checkpoint <checkpoints>`.

    This checks for cancellation and allows other tasks to be scheduled,
    without otherwise blocking.

    Note that the scheduler has the option of ignoring this and continuing to
    run the current task if it decides this is appropriate (e.g. for increased
    efficiency).

    Equivalent to ``await trio.sleep(0)`` (which is implemented by calling
    :func:`checkpoint`.)

    """
    # The scheduler is what checks timeouts and converts them into
    # cancellations. So by doing the schedule point first, we ensure that the
    # cancel point has the most up-to-date info.
    await cancel_shielded_checkpoint()
    task = current_task()
    task._cancel_points += 1
    if task._cancel_status.effectively_cancelled or (
        task is task._runner.main_task and task._runner.ki_pending
    ):
        cs = CancelScope(deadline=-inf)
        if (
            task._cancel_status._scope._cancel_reason is None
            and task is task._runner.main_task
            and task._runner.ki_pending
        ):
            task._cancel_status._scope._cancel_reason = CancelReason(
                source="KeyboardInterrupt"
            )
        assert task._cancel_status._scope._cancel_reason is not None
        cs._cancel_reason = task._cancel_status._scope._cancel_reason
        with cs:
            await _core.wait_task_rescheduled(lambda _: _core.Abort.SUCCEEDED)


async def checkpoint_if_cancelled() -> None: