You can read this chapter before running the experiment. Then return to the live trace and match each concept to a real event.
Python is the language and CPython is its main implementation. CPython parses source, creates code objects with bytecode, and executes them in frames. The interpreter itself is not an event loop: an async scheduler exists only when a program runs asyncio.
Modules, function bodies, and class bodies are code blocks executed in frames. CPython compiles source to version-specific bytecode. Reference counting handles most object lifetime and cyclic GC handles cycles. In a regular GIL build, one thread at a time executes Python bytecode, while blocking I/O may release the GIL. asyncio adds an event loop, tasks, futures, and awaitables.
First ask whether bytecode is executing, threads are coordinating access to objects, or a coroutine has cooperatively yielded to an event loop.
The parser builds syntax structure and the compiler resolves scopes and emits bytecode.
The frame holds code-block context and locals while the evaluation loop executes CPython instructions.
Threads share memory, processes isolate heaps, and asyncio switches tasks at suspension points.
An object is often reclaimed immediately after its last strong reference disappears.
Supplements reference counting by finding isolated reference cycles.
In a regular build it serializes Python bytecode across threads of one interpreter.
A thread shares the heap; a process has separate memory and IPC.
The loop resumes tasks after await; blocking code does not yield control.
The language defines semantics while CPython is an implementation. PyPy and other runtimes may use different VM, JIT, memory, and cleanup behavior.
Compound operations and shared mutable state still need synchronization; C extensions and free-threaded builds change assumptions further.
create_task(A, B) → main reaches await → A/B start → await suspends them → ready B resumes → AIf a coroutine calls time.sleep or computes for a long time without await, the loop cannot run the other tasks.
Understand the words first, then the execution order.
The reference and most widely used Python implementation, written primarily in C.
A compiled code block containing bytecode, constants, names, and metadata.
Instructions for the CPython VM; not a stable machine-code contract and different across versions.
A code-block execution context with code, locals, globals, instruction state, and call relationships.
The primary CPython lifetime mechanism: an object may be reclaimed once strong references are gone.
A collector that finds unreachable object groups that reference each other.
The Global Interpreter Lock required by a thread executing Python objects and bytecode in a regular CPython build.
A suspendable computation object returned by calling async def; it is not scheduled by itself.
An asyncio wrapper that schedules a coroutine and stores its result or exception.
An object accepted by await: coroutine, Task, Future, or a compatible protocol implementation.
A low-level future-result holder connecting callback APIs to an awaiting task.
An optional CPython build without the GIL that requires compatible libraries and extensions.
Each step maps to an observable runtime state.
The parser checks grammar and builds an AST; SyntaxError happens before that code block executes.
Names and scopes are classified, literals become constants, and operations become bytecode.
The function links a code object with globals, defaults, and closure cells.
Arguments bind to parameters and the evaluation loop begins executing bytecode.
Removing a binding reduces reachability while cycles require supplemental collection.
Regular CPython requires the GIL and commonly releases it around blocking I/O.
await yields control when needed and later readiness makes the task runnable.
Long pure-Python CPU usually moves to processes, native code, or a validated free-threaded architecture.
These details explain why similar code can sometimes produce a different trace.
Calling it creates a coroutine object; await, create_task, or another scheduler is needed.
An already-ready awaitable may return without an actual suspension.
Free-threaded builds arrived in 3.13 and other implementations have their own models.
Blocking I/O and many native extensions release the GIL and overlap waiting.
Startup, IPC, and serialization can cost more than a small CPU task.
It removes a binding; other references retain the object and allocators may retain arenas.
dis is useful, but opcodes are version-specific and profiling comes first.
Most asyncio objects are not thread-safe; cross-thread calls need dedicated APIs.
First understand which parts of Node participate in execution.
Then remove instrumentation and focus on the central mechanism.
Finally match the model to the code that produces the live trace.
async def fetch_pair():
first = asyncio.create_task(fetch("/first"))
second = asyncio.create_task(fetch("/second"))
return await asyncio.gather(first, second)
# Blocking legacy I/O must leave the loop thread:
data = await asyncio.to_thread(legacy_read, path)This is not an alternative example: these are the functions and files used by the Run button.
The first file is the real CPython scenario; the second safely starts the child process and turns its JSON Lines into the live trace.
#!/usr/bin/env python3
"""Fixed, input-free CPython scenarios used by Runtime Lab."""
from __future__ import annotations
import asyncio
import dis
import gc
import io
import json
import platform
import sys
import time
import weakref
from dataclasses import dataclass
from typing import Any, Iterator
def emit(lane: str, event_type: str, key: str, **data: Any) -> None:
print(
json.dumps(
{"lane": lane, "type": event_type, "key": key, "data": data},
ensure_ascii=False,
),
flush=True,
)
def version_event() -> None:
emit(
"python",
"runtime",
"python.version",
implementation=platform.python_implementation(),
version=platform.python_version(),
)
def run_syntax() -> None:
version_event()
orders = [
{"id": "A-10", "status": "paid", "price": 120, "qty": 2},
{"id": "A-11", "status": "draft", "price": 80, "qty": 1},
{"id": "A-12", "status": "paid", "price": 50, "qty": 3},
]
emit("objects", "state", "syntax.objects", count=len(orders))
paid = [order for order in orders if order["status"] == "paid"]
total = sum(order["price"] * order["qty"] for order in paid)
emit(
"comprehension",
"result",
"syntax.comprehension",
ids=", ".join(order["id"] for order in paid),
total=total,
)
first, *middle, last = [order["id"] for order in orders]
emit(
"sequence",
"result",
"syntax.unpack",
first=first,
middle=middle,
last=last,
)
labels = [f"{index}:{order['id']}" for index, order in enumerate(orders, start=1)]
emit("loop", "result", "syntax.enumerate", labels=", ".join(labels))
def format_order(order: dict[str, Any], *, currency: str = "RUB") -> str:
return f"{order['id']} · {order['price'] * order['qty']} {currency}"
emit(
"function",
"result",
"syntax.function",
rendered=format_order(orders[0], currency="₽"),
)
emit("result", "result", "syntax.result")
@dataclass(slots=True)
class CartLine:
sku: str
price: int
quantity: int = 1
@property
def subtotal(self) -> int:
return self.price * self.quantity
def append_bad(item: str, bucket: list[str] = []) -> list[str]:
bucket.append(item)
return list(bucket)
def append_safe(item: str, bucket: list[str] | None = None) -> list[str]:
target = [] if bucket is None else bucket
target.append(item)
return target
def even_squares(limit: int) -> Iterator[int]:
for value in range(limit):
if value % 2 == 0:
yield value * value
def classify_event(event: dict[str, Any]) -> str:
match event:
case {"type": "order.paid", "payload": {"id": order_id}}:
return f"paid order {order_id}"
case {"type": event_type}:
return f"other event {event_type}"
case _:
return "invalid event"
def run_semantics() -> None:
version_event()
line = CartLine("book", 450, quantity=2)
emit(
"class",
"result",
"semantics.dataclass",
rendered=repr(line),
subtotal=line.subtotal,
)
original = ["node"]
alias = original
alias.append("python")
emit("objects", "mutation", "semantics.alias", shared=original == alias == ["node", "python"])
bad_first = append_bad("api")
bad_second = append_bad("worker")
emit(
"function",
"warning",
"semantics.mutable-default",
first=bad_first,
second=bad_second,
)
safe_first = append_safe("api")
safe_second = append_safe("worker")
emit(
"function",
"result",
"semantics.safe-default",
first=safe_first,
second=safe_second,
)
emit("generator", "result", "semantics.generator", values=list(even_squares(7)))
emit(
"pattern",
"result",
"semantics.match",
label=classify_event({"type": "order.paid", "payload": {"id": "A-42"}}),
)
try:
int("not-a-number")
except ValueError as error:
emit("exception", "caught", "semantics.exception", message=str(error))
stream = io.StringIO()
with stream:
stream.write("cleanup is deterministic")
text = stream.getvalue()
emit("context", "cleanup", "semantics.context", closed=stream.closed, text=text)
emit("result", "result", "semantics.result")
def doubled_total(values: list[int]) -> int:
return sum(value * 2 for value in values)
class CycleNode:
def __init__(self) -> None:
self.peer: CycleNode | None = None
async def traced_task(name: str, delay: float, completion: list[str]) -> str:
emit("asyncio", "start", "asyncio.started", name=name)
await asyncio.sleep(delay)
completion.append(name)
emit("asyncio", "resume", "asyncio.resumed", name=name)
return name
async def measure_timer_while(blocking_call: Any) -> float:
started = time.perf_counter()
timer = asyncio.create_task(asyncio.sleep(0.01))
await blocking_call()
await timer
return max(0.0, (time.perf_counter() - started - 0.01) * 1_000)
async def run_asyncio_round() -> None:
completion: list[str] = []
first = asyncio.create_task(traced_task("A", 0.025, completion))
second = asyncio.create_task(traced_task("B", 0.005, completion))
emit("asyncio", "schedule", "asyncio.created")
await asyncio.gather(first, second)
emit("asyncio", "result", "asyncio.result", order=" → ".join(completion))
async def block_loop() -> None:
time.sleep(0.055)
async def offload_sleep() -> None:
await asyncio.to_thread(time.sleep, 0.055)
blocked_delay = await measure_timer_while(block_loop)
emit("asyncio", "blocking", "asyncio.blocking", delay=round(blocked_delay, 1))
offloaded_delay = await measure_timer_while(offload_sleep)
emit("asyncio", "offload", "asyncio.offload", delay=round(offloaded_delay, 1))
def run_runtime() -> None:
implementation = platform.python_implementation()
gil_probe = getattr(sys, "_is_gil_enabled", None)
gil_enabled = gil_probe() if gil_probe else implementation == "CPython"
emit(
"runtime",
"config",
"runtime.config",
implementation=implementation,
version=platform.python_version(),
gil=gil_enabled,
)
operations = [instruction.opname for instruction in dis.get_instructions(doubled_total)]
emit(
"bytecode",
"result",
"runtime.bytecode",
operations=" → ".join(operations[:12]),
)
frame = sys._getframe()
visible_locals = ", ".join(sorted(name for name in frame.f_locals if not name.startswith("_")))
emit(
"frame",
"state",
"runtime.frame",
functionName=frame.f_code.co_name,
locals=visible_locals,
)
left = CycleNode()
right = CycleNode()
left.peer = right
right.peer = left
left_ref = weakref.ref(left)
right_ref = weakref.ref(right)
alive_before = left_ref() is not None and right_ref() is not None
del left, right
collected = gc.collect()
alive_after = left_ref() is not None or right_ref() is not None
emit(
"gc",
"result",
"runtime.gc",
collected=collected,
aliveBefore=alive_before,
aliveAfter=alive_after,
)
asyncio.run(run_asyncio_round())
emit("result", "result", "runtime.result")
SCENARIOS = {
"syntax": run_syntax,
"semantics": run_semantics,
"runtime": run_runtime,
}
def main() -> None:
scenario = sys.argv[1] if len(sys.argv) > 1 else ""
runner = SCENARIOS.get(scenario)
if runner is None:
raise SystemExit(f"unknown scenario: {scenario}")
runner()
if __name__ == "__main__":
main()
The scenario instruments its own trace: Python calls emit(...) and prints ordered events as JSON Lines, while Node records a timestamp as each line arrives. Source and lane labels come from application code, not a CPython/asyncio profiler. The Node bridge reads output without a shell, bounds runtime and volume, and forwards events into the HTTP stream.
Compare the goal, code, and caveats instead of memorizing syntax without a model.
Separate a created computation from scheduled work.
async def load_user(user_id: int):
return await repository.get(user_id)
coroutine = load_user(7)
task = asyncio.create_task(coroutine)
user = await taskAvoid sequentially waiting for independent operations.
user, orders = await asyncio.gather(
load_user(user_id),
load_orders(user_id),
)Keep the event-loop thread responsive.
def read_legacy_file(path: str) -> bytes:
return legacy_client.read(path)
data = await asyncio.to_thread(read_legacy_file, path)Use multiple cores for pure-Python work.
pool = ProcessPoolExecutor()
result = await asyncio.get_running_loop().run_in_executor(
pool, calculate_report, payload
)Inspect VM instructions.
import dis
def total(values):
return sum(value * 2 for value in values)
dis.dis(total)Measure collection instead of guessing from RSS.
import gc
before = gc.get_stats()
collected = gc.collect()
after = gc.get_stats()A realistic service: the original code, observable failure, corrected implementation, and why the correction works.
An async handler builds a CPU-heavy PDF and calls a synchronous provider. The author assumes async def automatically sends its body to a background thread.
Until a real suspension point, the coroutine runs in the event-loop thread. time.sleep, synchronous I/O, and pure-Python CPU delay timers, sockets, and every other task on that loop.
async def build_report(payload: dict) -> bytes:
customer = sync_crm.load(payload["customer_id"])
time.sleep(0.2)
return render_pdf(customer, payload)async def changes the call protocol but not the nature of synchronous functions inside it. None of these three calls yield control to asyncio.
from concurrent.futures import ProcessPoolExecutor
report_pool = ProcessPoolExecutor(max_workers=2)
async def build_report(payload: dict) -> bytes:
customer = await asyncio.to_thread(
sync_crm.load,
payload["customer_id"],
)
loop = asyncio.get_running_loop()
return await loop.run_in_executor(
report_pool,
render_pdf,
customer,
payload,
)Blocking I/O moves to a bounded thread pool. Pure-Python CPU runs in a bounded process pool and can use another core without the main interpreter GIL.
What the unfamiliar calls from both code samples actually do.
async defasyncio.to_thread(...)ProcessPoolExecutorget_running_loop()run_in_executor(...)max_workers=2The myth is on the left; the accurate model is on the right.
Python executes source line by line without compilation.
CPython compiles code blocks into code objects and bytecode first.
The GIL prevents race conditions.
Shared state and compound operations still require locks or queues.
asyncio makes a blocking library non-blocking.
A synchronous call blocks the loop unless replaced or explicitly offloaded.
await starts another thread.
Normally the same task and thread yield control to the event loop until readiness.
Threads speed up pure-Python CPU.
Regular CPython does not run that bytecode in parallel; use processes, native code, or a free-threaded build.
gc.collect guarantees a lower RSS.
Objects become available to the allocator, but the allocator and OS may retain arenas.
If you can explain the answer without quoting documentation, your mental model is starting to take shape.