Skip to content

Commit c3b4866

Browse files
feat: teach asyncio on the canvas (coroutines, tasks, event loop)
Add the concurrency exemplar: async def returns a suspended coroutine, create_task wraps it in a task, and a.run is a tick-based event loop that round-robins ready tasks to their next await, so two tasks visibly interleave. The suspension model stores a coroutine's program counter and locals across awaits. New value shapes (coroutine/task/eventloop), the async def and await parser constructs, a canvas renderer with lifecycle status colors, and five levels (ELI10 to ELI-PHD) plus the deep lesson with two schematics. Catalog is now 50 levels / 12 series.
1 parent b23ccfc commit c3b4866

10 files changed

Lines changed: 913 additions & 2 deletions

File tree

‎src/engine/checks.ts‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -385,8 +385,63 @@ export const preds = {
385385
return itv.index >= len;
386386
};
387387
},
388+
389+
/** `name` is a coroutine (the object an `async def` call returns). */
390+
isCoroutine(name: string) {
391+
return (snapshot: Snapshot, _runtime: MemoryState): boolean | string => {
392+
const view = slotView(name, snapshot);
393+
return view?.type === 'coroutine' ? true : `'${name}' is not a coroutine — call an async def.`;
394+
};
395+
},
396+
397+
/** `name` is a Task (a named handle to a coroutine). */
398+
isTask(name: string) {
399+
return (snapshot: Snapshot, _runtime: MemoryState): boolean | string => {
400+
const view = slotView(name, snapshot);
401+
return view?.type === 'task' ? true : `'${name}' is not a task — wrap a coroutine with asyncio.create_task().`;
402+
};
403+
},
404+
405+
/** `name` (a task) is currently in `status`. */
406+
taskStatus(name: string, status: string) {
407+
return (snapshot: Snapshot, _runtime: MemoryState): boolean | string => {
408+
const t = asyncTaskValue(name, snapshot);
409+
if (!t) return `Bind '${name}' as a task first.`;
410+
return t.status === status ? true : `Task '${name}' is ${t.status}, expected ${status}.`;
411+
};
412+
},
413+
414+
/** The event loop advanced at least `n` ticks (the tasks actually interleaved). */
415+
loopRan(n: number) {
416+
return (_snapshot: Snapshot, runtime: MemoryState): boolean | string => {
417+
return runtime.ticks >= n ? true : `The event loop ran ${runtime.ticks} tick(s), expected at least ${n}.`;
418+
};
419+
},
420+
421+
/** Two tasks wrap *different* coroutines (independent units of work). */
422+
differentTasks(a: string, b: string) {
423+
return (snapshot: Snapshot, _runtime: MemoryState): boolean | string => {
424+
const ta = asyncTaskValue(a, snapshot);
425+
const tb = asyncTaskValue(b, snapshot);
426+
if (!ta || !tb) return `Bind both '${a}' and '${b}' as tasks.`;
427+
return ta.coro !== tb.coro ? true : `'${a}' and '${b}' wrap the same coroutine.`;
428+
};
429+
},
388430
};
389431

432+
/** Read a task's normalized `{ coro, name, status }` from the snapshot, or null. */
433+
function asyncTaskValue(name: string, snapshot: Snapshot): { coro: string; name: string; status: string } | null {
434+
const module = snapshot.frames[snapshot.frames.length - 1];
435+
if (!module) return null;
436+
const id = module.locals[name];
437+
if (typeof id !== 'string' || !snapshot.heap[id]) return null;
438+
const obj = snapshot.heap[id];
439+
if (obj.type !== 'task') return null;
440+
const v = obj.value as { coro?: unknown; name?: unknown; status?: unknown };
441+
if (typeof v.coro !== 'string' || typeof v.status !== 'string') return null;
442+
return { coro: v.coro, name: typeof v.name === 'string' ? v.name : '', status: v.status };
443+
}
444+
390445
/** Read a pandas Series/DataFrame's normalized `buffer` id from the snapshot. */
391446
function pandasBufferId(name: string, snapshot: Snapshot): string | null {
392447
const module = snapshot.frames[snapshot.frames.length - 1];

‎src/engine/evaluator.ts‎

Lines changed: 286 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,18 @@ function execStatement(node: AstNode, state: MemoryState): void {
112112
state.heap[fn.id] = fn;
113113
return;
114114
}
115+
case 'CoroutineDef': {
116+
// An `async def` binds a coroutine-function code object (async: true).
117+
const fn = allocObject('function', {
118+
name: node.name,
119+
params: [...node.params],
120+
body: node.body,
121+
async: true,
122+
} as FunctionCode, `async function ${node.name}`);
123+
setFrameLocal(currentFrame(state), node.name, fn.id);
124+
state.heap[fn.id] = fn;
125+
return;
126+
}
115127
case 'Import': {
116128
const mod = allocObject('module', { name: node.name }, `<module '${node.name}'>`);
117129
state.heap[mod.id] = mod;
@@ -134,6 +146,25 @@ class PyReturn extends Error {
134146
}
135147
}
136148

149+
/**
150+
* Raised by `await a.sleep(...)` inside a coroutine to hand the tick back to the
151+
* event loop (a cooperative-yield suspension point). The coroutine executor
152+
* catches it, saves the suspended state, and resumes on the next tick.
153+
*/
154+
class PySuspend extends Error {
155+
constructor() {
156+
super('suspend');
157+
}
158+
}
159+
160+
/**
161+
* The id of the coroutine currently being stepped by the event loop, or null.
162+
* Lets `evalExpr` on an `Await` know whether a suspension is in force (inside a
163+
* coroutine) or a plain no-op (a bare `await` outside a running task). The
164+
* engine is single-threaded, so a module pointer is safe.
165+
*/
166+
let activeCoroutineId: string | null = null;
167+
137168
function currentFrame(state: MemoryState): Frame {
138169
return state.frames[state.frames.length - 1];
139170
}
@@ -314,6 +345,15 @@ function evalExpr(node: AstNode, state: MemoryState): RtValue {
314345
case 'Call': {
315346
return evalCall(node, state);
316347
}
348+
case 'Await': {
349+
// `await a.sleep(...)` inside a running coroutine is a suspension point:
350+
// it hands the tick back to the event loop (a cooperative yield). Outside a
351+
// running coroutine it is a no-op that just evaluates its operand.
352+
if (activeCoroutineId) {
353+
throw new PySuspend();
354+
}
355+
return evalExpr(node.value, state);
356+
}
317357
default:
318358
throw new Error(`RuntimeError: unsupported expression ${node.kind}`);
319359
}
@@ -435,12 +475,17 @@ function callUserFunction(fnId: RtValue, args: RtValue[], state: MemoryState, li
435475
if (!fn || fn.type !== 'function') {
436476
throw new Error(`TypeError: object is not callable (line ${line})`);
437477
}
438-
const { name, params, body } = fn.value as FunctionCode;
478+
const { name, params, body, async: isAsync } = fn.value as FunctionCode;
439479
if (args.length !== params.length) {
440480
throw new Error(
441481
`TypeError: ${name}() takes ${params.length} positional argument(s) but ${args.length} were given (line ${line})`,
442482
);
443483
}
484+
// Calling a coroutine function does NOT run it — it returns a suspended
485+
// coroutine object (the whole point of async: the body runs later, on ticks).
486+
if (isAsync) {
487+
return makeCoroutine(name, params, body, args, fnId as string, state);
488+
}
444489
const frame: Frame = { name, locals: {} };
445490
params.forEach((p, i) => {
446491
frame.locals[p] = args[i];
@@ -498,6 +543,17 @@ function callMethod(objId: RtValue, method: string, args: RtValue[], state: Memo
498543
throw new Error(`AttributeError: module 'pandas' has no attribute '${method}' (line ${line})`);
499544
}
500545

546+
if (obj.type === 'module' && (obj.value as { name: string }).name === 'asyncio') {
547+
if (method === 'create_task') return makeTask(args, state, line);
548+
if (method === 'gather') return gather(args, state, line);
549+
if (method === 'sleep') return asyncSleep(args, state, line);
550+
if (method === 'run') {
551+
runEventLoop(args, state, line);
552+
return null;
553+
}
554+
throw new Error(`AttributeError: module 'asyncio' has no attribute '${method}' (line ${line})`);
555+
}
556+
501557
if (obj.type === 'ndarray') {
502558
if (method === 'copy') return copyNdarray(objId as string, state, line);
503559
if (method === 'sum') return ndarraySum(objId as string, state, line);
@@ -1034,6 +1090,235 @@ function ndarraySum(ndId: string, state: MemoryState, line: number): RtValue {
10341090
return ndarrayWindow(ndId, state).reduce((acc, n) => acc + n, 0);
10351091
}
10361092

1093+
/* ————————————————— asyncio (concurrency exemplar) —————————————————
1094+
*
1095+
* A coroutine is an object that holds a function's state across `await` points;
1096+
* a Task is a named handle to one; the event loop is a tick-based scheduler that
1097+
* round-robins ready tasks — each tick advances one task to its next `await`,
1098+
* then moves on. There is no real loop or wall clock (the fidelity boundary):
1099+
* the *schedule* is simulated so two coroutines visibly interleave.
1100+
*/
1101+
1102+
type AsyncStatus = 'pending' | 'ready' | 'suspended' | 'done';
1103+
1104+
interface CoroutineValue {
1105+
fn: string;
1106+
status: AsyncStatus;
1107+
pc: number;
1108+
name: string;
1109+
params: string[];
1110+
body: AstNode[];
1111+
locals: Record<string, unknown>;
1112+
returnValue: RtValue;
1113+
}
1114+
1115+
interface TaskValue {
1116+
coro: string;
1117+
name: string;
1118+
status: AsyncStatus;
1119+
}
1120+
1121+
interface EventLoopValue {
1122+
running: boolean;
1123+
ready: string[];
1124+
done: string[];
1125+
tick: number;
1126+
}
1127+
1128+
function asyncRepr(type: string, v: unknown, state: MemoryState): string {
1129+
if (type === 'coroutine') {
1130+
const c = v as CoroutineValue;
1131+
return `coro ${c.name}() · ${c.status}`;
1132+
}
1133+
if (type === 'task') {
1134+
const t = v as TaskValue;
1135+
return `Task(${t.name}) · ${t.status}`;
1136+
}
1137+
const l = v as EventLoopValue;
1138+
const readyNames = l.ready.map((id) => (state.heap[id]?.value as TaskValue | undefined)?.name ?? '?').join(', ');
1139+
const doneNames = l.done.map((id) => (state.heap[id]?.value as TaskValue | undefined)?.name ?? '?').join(', ');
1140+
return `EventLoop · tick ${l.tick} · running ${l.running ? 'True' : 'False'} · [ready ${readyNames || '∅'} | done ${doneNames || '∅'}]`;
1141+
}
1142+
1143+
/**
1144+
* Bind a coroutine function's call to a *suspended* coroutine object. Its
1145+
* `locals` are seeded with the parameters; `pc` starts at 0 and the body has not
1146+
* yet run — it only runs when the event loop ticks it.
1147+
*/
1148+
function makeCoroutine(
1149+
name: string,
1150+
params: string[],
1151+
body: AstNode[],
1152+
args: RtValue[],
1153+
fnId: string,
1154+
state: MemoryState,
1155+
): string {
1156+
const locals: Record<string, unknown> = {};
1157+
params.forEach((p, i) => (locals[p] = args[i]));
1158+
const v: CoroutineValue = {
1159+
fn: fnId,
1160+
status: 'pending',
1161+
pc: 0,
1162+
name,
1163+
params: [...params],
1164+
body,
1165+
locals,
1166+
returnValue: null,
1167+
};
1168+
const obj = allocObject('coroutine', v, asyncRepr('coroutine', v, state));
1169+
state.heap[obj.id] = obj;
1170+
return obj.id;
1171+
}
1172+
1173+
/** `a.create_task(coro)` → a Task wrapping the given coroutine (initially ready). */
1174+
function makeTask(args: RtValue[], state: MemoryState, line: number): RtValue {
1175+
if (args.length !== 1) throw new Error(`TypeError: create_task() takes one argument (a coroutine) (line ${line})`);
1176+
const coroId = args[0];
1177+
if (typeof coroId !== 'string' || state.heap[coroId]?.type !== 'coroutine') {
1178+
throw new Error(`TypeError: create_task() takes a coroutine, got ${typeName(coroId, state)} (line ${line})`);
1179+
}
1180+
const coro = state.heap[coroId].value as CoroutineValue;
1181+
const v: TaskValue = { coro: coroId, name: coro.name, status: coro.status === 'done' ? 'done' : 'ready' };
1182+
const obj = allocObject('task', v, asyncRepr('task', v, state));
1183+
state.heap[obj.id] = obj;
1184+
return obj.id;
1185+
}
1186+
1187+
/** `a.sleep(secs)` → a no-op builtin; the waiting is the surrounding `await`. */
1188+
function asyncSleep(args: RtValue[], state: MemoryState, line: number): RtValue {
1189+
void state; // sleep is a no-op builtin — the waiting is the surrounding `await`
1190+
if (args.length !== 1) throw new Error(`TypeError: sleep() takes one argument (line ${line})`);
1191+
const secs = Number(args[0]);
1192+
if (Number.isNaN(secs)) throw new Error(`TypeError: sleep() takes a number (line ${line})`);
1193+
return null;
1194+
}
1195+
1196+
/** `a.gather(t1, t2, ...)` → a list of the task results (eager, for display). */
1197+
function gather(args: RtValue[], state: MemoryState, line: number): RtValue {
1198+
if (args.length < 1) throw new Error(`TypeError: gather() takes at least one awaitable (line ${line})`);
1199+
const results: RtValue[] = args.map((a) => taskReturnValue(a, state, line));
1200+
const obj = allocObject('list', results, '');
1201+
state.heap[obj.id] = obj;
1202+
obj.repr = formatObject(obj, state);
1203+
return obj.id;
1204+
}
1205+
1206+
/** The value a finished task's coroutine returned (None if not done yet). */
1207+
function taskReturnValue(v: RtValue, state: MemoryState, line: number): RtValue {
1208+
if (typeof v !== 'string' || state.heap[v]?.type !== 'task') {
1209+
throw new Error(`TypeError: gather() takes tasks, got ${typeName(v, state)} (line ${line})`);
1210+
}
1211+
const t = state.heap[v].value as TaskValue;
1212+
const coro = state.heap[t.coro];
1213+
if (!coro || coro.type !== 'coroutine') throw new Error(`RuntimeError: task coroutine was released (line ${line})`);
1214+
const cv = coro.value as CoroutineValue;
1215+
return cv.status === 'done' ? cv.returnValue : null;
1216+
}
1217+
1218+
/**
1219+
* Advance one coroutine by exactly one tick: run statements from its `pc` until
1220+
* it suspends on a cooperative `await` (handing the tick back to the loop) or it
1221+
* returns. The suspended frame's locals are saved back onto the coroutine object
1222+
* so the next tick resumes where it left off — a coroutine is a function that
1223+
* remembers its place.
1224+
*/
1225+
function stepCoroutine(state: MemoryState): void {
1226+
const coroId = activeCoroutineId!;
1227+
const c = state.heap[coroId].value as CoroutineValue;
1228+
const frame: Frame = { name: c.name, locals: {} };
1229+
for (const [k, v] of Object.entries(c.locals)) frame.locals[k] = v as RtValue;
1230+
state.frames.push(frame);
1231+
try {
1232+
while (c.pc < c.body.length) {
1233+
try {
1234+
execStatement(c.body[c.pc]!, state);
1235+
} catch (err) {
1236+
if (err instanceof PySuspend) {
1237+
// Cooperative yield: move past the `await` statement so the next tick
1238+
// resumes after it, not by re-running it.
1239+
c.pc += 1;
1240+
c.status = 'suspended';
1241+
return;
1242+
}
1243+
if (err instanceof PyReturn) {
1244+
c.returnValue = err.value;
1245+
c.pc = c.body.length;
1246+
c.status = 'done';
1247+
return;
1248+
}
1249+
throw err;
1250+
}
1251+
c.pc += 1;
1252+
}
1253+
c.returnValue = null;
1254+
c.status = 'done';
1255+
} finally {
1256+
for (const [k, v] of Object.entries(frame.locals)) c.locals[k] = v;
1257+
state.frames.pop();
1258+
refreshAsyncReprs(state);
1259+
}
1260+
}
1261+
1262+
/** Recompute the repr of every async object (status / queue change each tick). */
1263+
function refreshAsyncReprs(state: MemoryState): void {
1264+
for (const obj of Object.values(state.heap)) {
1265+
if (obj.type === 'coroutine' || obj.type === 'task' || obj.type === 'eventloop') {
1266+
obj.repr = asyncRepr(obj.type, obj.value, state);
1267+
}
1268+
}
1269+
}
1270+
1271+
/**
1272+
* `a.run(t1, t2, ...)` — the event loop. It builds a loop object whose ready
1273+
* queue holds the tasks, then steps them round-robin, one task per tick, until
1274+
* all are done. Each tick is one unit of interleaving; the round-robined stdout
1275+
* is the visible proof the tasks ran cooperatively (A, B, A, B, …), not one
1276+
* after the other.
1277+
*/
1278+
function runEventLoop(args: RtValue[], state: MemoryState, line: number): void {
1279+
if (args.length < 1) throw new Error(`TypeError: run() takes at least one task (line ${line})`);
1280+
const taskIds = args.map((a) => ensureTaskId(a, state, line));
1281+
const ready = taskIds.filter((id) => (state.heap[id].value as TaskValue).status !== 'done');
1282+
const done = taskIds.filter((id) => (state.heap[id].value as TaskValue).status === 'done');
1283+
const loop: EventLoopValue = { running: true, ready, done, tick: 0 };
1284+
const loopObj = allocObject('eventloop', loop, asyncRepr('eventloop', loop, state));
1285+
state.heap[loopObj.id] = loopObj;
1286+
1287+
let guard = 0;
1288+
while (loop.ready.length > 0) {
1289+
const taskId = loop.ready.shift()!;
1290+
const task = state.heap[taskId];
1291+
if (!task || task.type !== 'task') {
1292+
loop.ready = [];
1293+
break;
1294+
}
1295+
const tv = task.value as TaskValue;
1296+
activeCoroutineId = tv.coro; // this step runs inside the task's coroutine
1297+
stepCoroutine(state);
1298+
activeCoroutineId = null;
1299+
loop.tick += 1;
1300+
state.ticks += 1;
1301+
const nowDone = (state.heap[tv.coro].value as CoroutineValue).status === 'done';
1302+
tv.status = nowDone ? 'done' : 'ready';
1303+
if (nowDone) loop.done.push(taskId);
1304+
else loop.ready.push(taskId);
1305+
// Refresh all async reprs after the state/queue update so the boxes agree.
1306+
refreshAsyncReprs(state);
1307+
if (guard++ > 10000) {
1308+
loop.running = false;
1309+
break;
1310+
}
1311+
}
1312+
loop.running = false;
1313+
refreshAsyncReprs(state);
1314+
}
1315+
1316+
/** Resolve a run() argument to a task object id (rejects lists / non-tasks). */
1317+
function ensureTaskId(v: RtValue, state: MemoryState, line: number): string {
1318+
if (typeof v === 'string' && state.heap[v]?.type === 'task') return v;
1319+
throw new Error(`TypeError: run() takes tasks, got ${typeName(v, state)} (line ${line})`);
1320+
}
1321+
10371322
/* ————————————————— pandas (Phase 2) —————————————————
10381323
*
10391324
* A Series/DataFrame is *windows over a buffer*, the numpy picture lifted to a

0 commit comments

Comments
 (0)