mirror of
https://codeberg.org/ziglings/exercises.git
synced 2026-06-08 07:50:00 +00:00
Merge pull request 'Improvements for async-io' (#388) from async-improvements into main
Reviewed-on: https://codeberg.org/ziglings/exercises/pulls/388
This commit is contained in:
@@ -1180,9 +1180,9 @@ const exercises = [_]Exercise{
|
|||||||
.{
|
.{
|
||||||
.main_file = "093_async9.zig",
|
.main_file = "093_async9.zig",
|
||||||
.output =
|
.output =
|
||||||
\\Main thread continues...
|
\\Computing concurrently!
|
||||||
\\Computing on a separate thread!
|
\\Main continues...
|
||||||
\\Main thread done waiting.
|
\\Main done waiting.
|
||||||
\\Result: 123
|
\\Result: 123
|
||||||
, // pay attention to the comma
|
, // pay attention to the comma
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -7,10 +7,11 @@
|
|||||||
// This interface uses a VTable pattern - a struct of function pointers -
|
// This interface uses a VTable pattern - a struct of function pointers -
|
||||||
// to abstract over different concurrency backends:
|
// to abstract over different concurrency backends:
|
||||||
//
|
//
|
||||||
// * Threaded - classic thread-pool based I/O
|
// * Threaded - thread-pool based I/O
|
||||||
// * Uring - Linux io_uring
|
// * Evented - chooses the best event-loop backend for your OS:
|
||||||
// * Kqueue - BSD/macOS
|
// * Uring on Linux (io_uring)
|
||||||
// * Dispatch - macOS Grand Central Dispatch
|
// * Kqueue on BSD/macOS
|
||||||
|
// * Dispatch on macOS (Grand Central Dispatch)
|
||||||
//
|
//
|
||||||
// The Io struct itself is tiny:
|
// The Io struct itself is tiny:
|
||||||
//
|
//
|
||||||
|
|||||||
@@ -6,41 +6,47 @@
|
|||||||
// won't necessarily be available until you call .await() on it:
|
// won't necessarily be available until you call .await() on it:
|
||||||
//
|
//
|
||||||
// var future = io.async(someFunction, .{ arg1, arg2 });
|
// var future = io.async(someFunction, .{ arg1, arg2 });
|
||||||
// // ... do other work here ...
|
|
||||||
// const result = future.await(io);
|
// const result = future.await(io);
|
||||||
//
|
//
|
||||||
// The function *may* run immediately or on another thread -
|
// The function *may* run immediately or on another thread -
|
||||||
// your code doesn't need to care! That's the beauty of the
|
// your code doesn't need to care! That's the beauty of the
|
||||||
// Io abstraction. (In the Threaded backend, if no thread is
|
// Io abstraction.
|
||||||
// available, the function runs synchronously right away and
|
|
||||||
// .await() just returns the already-computed result.)
|
|
||||||
//
|
//
|
||||||
// io.async() returns a Future(T) where T is the return type
|
// IMPORTANT: Every Future MUST be either .await()ed or .cancel()ed.
|
||||||
// of the function you passed in. Future has two key methods:
|
// Failing to do so leaks resources! A safe pattern is:
|
||||||
//
|
//
|
||||||
// .await(io) - block until the result is ready, return it
|
// var future = io.async(myFn, .{});
|
||||||
// .cancel(io) - request cancellation, then return the result
|
// defer _ = future.cancel(io); // safety net
|
||||||
|
// // ... later, if we want the result:
|
||||||
|
// const result = future.await(io);
|
||||||
|
// // (await after cancel is fine — it just returns the result)
|
||||||
|
//
|
||||||
|
// Both .await() and .cancel() block until the task finishes and
|
||||||
|
// return the result. The difference is that .cancel() also
|
||||||
|
// requests the task to stop at its next cancellation point.
|
||||||
|
// Calling either one more than once is safe — subsequent calls
|
||||||
|
// just return a copy of the result.
|
||||||
//
|
//
|
||||||
// Fix this program so that computeAnswer runs asynchronously
|
// Fix this program so that computeAnswer runs asynchronously
|
||||||
// and its result is properly awaited.
|
// and its result is properly awaited.
|
||||||
//
|
//
|
||||||
const std = @import("std");
|
const std = @import("std");
|
||||||
|
const print = std.debug.print;
|
||||||
|
|
||||||
pub fn main(init: std.process.Init) !void {
|
pub fn main(init: std.process.Init) !void {
|
||||||
const io = init.io;
|
const io = init.io;
|
||||||
|
|
||||||
// Launch computeAnswer asynchronously.
|
// Launch computeAnswer asynchronously.
|
||||||
// io.async() takes a function and a tuple of its arguments.
|
|
||||||
var future = io.async(computeAnswer, .{ 6, 7 });
|
var future = io.async(computeAnswer, .{ 6, 7 });
|
||||||
|
defer _ = future.cancel(io); // always clean up!
|
||||||
|
|
||||||
// Meanwhile, print something to show we're not blocked.
|
print("Computing... ", .{});
|
||||||
std.debug.print("Computing... ", .{});
|
|
||||||
|
|
||||||
// Now collect the result. What method on Future gives us
|
// Now collect the result. What method on Future gives us
|
||||||
// the value, blocking if it isn't ready yet?
|
// the value, blocking until it's ready?
|
||||||
const answer = future.???(io);
|
const answer = future.???(io);
|
||||||
|
|
||||||
std.debug.print("The answer is: {}\n", .{answer});
|
print("The answer is: {}\n", .{answer});
|
||||||
}
|
}
|
||||||
|
|
||||||
fn computeAnswer(a: u32, b: u32) u32 {
|
fn computeAnswer(a: u32, b: u32) u32 {
|
||||||
|
|||||||
@@ -5,20 +5,17 @@
|
|||||||
// them all. The Io backend may run them concurrently:
|
// them all. The Io backend may run them concurrently:
|
||||||
//
|
//
|
||||||
// var f1 = io.async(taskA, .{});
|
// var f1 = io.async(taskA, .{});
|
||||||
|
// defer _ = f1.cancel(io);
|
||||||
// var f2 = io.async(taskB, .{});
|
// var f2 = io.async(taskB, .{});
|
||||||
//
|
// defer _ = f2.cancel(io);
|
||||||
// // Both tasks may be running now!
|
|
||||||
// const a = f1.await(io);
|
// const a = f1.await(io);
|
||||||
// const b = f2.await(io);
|
// const b = f2.await(io);
|
||||||
//
|
//
|
||||||
// There's also io.concurrent() which provides a STRONGER guarantee:
|
// Notice the defer pattern: each async call is immediately
|
||||||
// it ensures the function gets its own unit of concurrency (e.g. a
|
// followed by a defer cancel. This ensures cleanup even if
|
||||||
// real OS thread). But it can fail with error.ConcurrencyUnavailable
|
// we return early or hit an error before reaching await.
|
||||||
// if resources are exhausted.
|
// Since await/cancel are idempotent, the defer is harmless
|
||||||
//
|
// if we've already awaited.
|
||||||
// io.async() is more portable: if no thread is available, it simply
|
|
||||||
// runs the function synchronously. This makes it the right default
|
|
||||||
// for most code.
|
|
||||||
//
|
//
|
||||||
// Fix this program to launch both tasks and collect their results.
|
// Fix this program to launch both tasks and collect their results.
|
||||||
//
|
//
|
||||||
@@ -29,12 +26,14 @@ pub fn main(init: std.process.Init) !void {
|
|||||||
const io = init.io;
|
const io = init.io;
|
||||||
|
|
||||||
// Launch both tasks asynchronously.
|
// Launch both tasks asynchronously.
|
||||||
var future_a = io.async(slowAdd, .{ 10, 20 });
|
var future_a = io.async(slowAdd, .{ 1, 2 });
|
||||||
|
defer _ = future_a.cancel(io);
|
||||||
var future_b = ???(slowMul, .{ 6, 7 });
|
var future_b = ???(slowMul, .{ 6, 7 });
|
||||||
|
defer _ = future_b.cancel(io);
|
||||||
|
|
||||||
// Await both results.
|
// Await both results.
|
||||||
const sum = future_a.await(io);
|
const sum = future_a.await(io);
|
||||||
const product = future_b.???(io);
|
const product = future_b.await(io);
|
||||||
|
|
||||||
print("{} + {} = {}\n", .{ 1, 2, sum });
|
print("{} + {} = {}\n", .{ 1, 2, sum });
|
||||||
print("{} * {} = {}\n", .{ 6, 7, product });
|
print("{} * {} = {}\n", .{ 6, 7, product });
|
||||||
|
|||||||
@@ -14,7 +14,7 @@
|
|||||||
// * Once you call group.async(), you MUST eventually call
|
// * Once you call group.async(), you MUST eventually call
|
||||||
// group.await() or group.cancel() to release resources.
|
// group.await() or group.cancel() to release resources.
|
||||||
// * group.cancel() requests cancellation on ALL members,
|
// * group.cancel() requests cancellation on ALL members,
|
||||||
// then waits for them to finish.
|
// then blocks until they all finish.
|
||||||
//
|
//
|
||||||
// Unlike Future, Group tasks don't return values to the caller.
|
// Unlike Future, Group tasks don't return values to the caller.
|
||||||
// They're ideal for parallel work that communicates through
|
// They're ideal for parallel work that communicates through
|
||||||
@@ -38,7 +38,7 @@ pub fn main(init: std.process.Init) !void {
|
|||||||
|
|
||||||
// Wait for all tasks to finish.
|
// Wait for all tasks to finish.
|
||||||
// What Group method blocks until all tasks complete?
|
// What Group method blocks until all tasks complete?
|
||||||
try group.???
|
try group.???(io);
|
||||||
|
|
||||||
print("All tasks finished!\n", .{});
|
print("All tasks finished!\n", .{});
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,7 +5,7 @@
|
|||||||
// Every Future has a .cancel() method that:
|
// Every Future has a .cancel() method that:
|
||||||
// 1. Requests the task to stop (via error.Canceled at the
|
// 1. Requests the task to stop (via error.Canceled at the
|
||||||
// next "cancellation point")
|
// next "cancellation point")
|
||||||
// 2. Waits for the task to actually finish
|
// 2. BLOCKS until the task actually finishes
|
||||||
// 3. Returns whatever result the task produced
|
// 3. Returns whatever result the task produced
|
||||||
//
|
//
|
||||||
// A "cancellation point" is any Io function that can return
|
// A "cancellation point" is any Io function that can return
|
||||||
@@ -13,7 +13,7 @@
|
|||||||
//
|
//
|
||||||
// fn myTask(io: std.Io) u32 {
|
// fn myTask(io: std.Io) u32 {
|
||||||
// io.sleep(...) catch |err| switch (err) {
|
// io.sleep(...) catch |err| switch (err) {
|
||||||
// error.Canceled => return 0, // handle gracefully
|
// error.Canceled => return 0, // error handle
|
||||||
// };
|
// };
|
||||||
// return 42;
|
// return 42;
|
||||||
// }
|
// }
|
||||||
@@ -21,6 +21,11 @@
|
|||||||
// This is fundamentally different from killing a thread -
|
// This is fundamentally different from killing a thread -
|
||||||
// the task gets a chance to clean up and return a value!
|
// the task gets a chance to clean up and return a value!
|
||||||
//
|
//
|
||||||
|
// Remember: both .await() and .cancel() block and return the
|
||||||
|
// result. The only difference is that .cancel() also sends
|
||||||
|
// the cancellation request. And both are idempotent — calling
|
||||||
|
// either one again just returns the same result.
|
||||||
|
//
|
||||||
// Fix this program: the slow task would take 10 seconds,
|
// Fix this program: the slow task would take 10 seconds,
|
||||||
// but we cancel it after 1 second. The task should detect
|
// but we cancel it after 1 second. The task should detect
|
||||||
// the cancellation and return early.
|
// the cancellation and return early.
|
||||||
@@ -32,6 +37,7 @@ pub fn main(init: std.process.Init) !void {
|
|||||||
const io = init.io;
|
const io = init.io;
|
||||||
|
|
||||||
var future = io.async(slowTask, .{io});
|
var future = io.async(slowTask, .{io});
|
||||||
|
defer _ = future.cancel(io); // safety net
|
||||||
|
|
||||||
// Wait 1 second, then cancel instead of waiting the full 10.
|
// Wait 1 second, then cancel instead of waiting the full 10.
|
||||||
io.sleep(std.Io.Duration.fromSeconds(1), .awake) catch {};
|
io.sleep(std.Io.Duration.fromSeconds(1), .awake) catch {};
|
||||||
@@ -40,7 +46,7 @@ pub fn main(init: std.process.Init) !void {
|
|||||||
|
|
||||||
// We don't want to wait 10 seconds!
|
// We don't want to wait 10 seconds!
|
||||||
// Which Future method requests cancellation AND returns the result?
|
// Which Future method requests cancellation AND returns the result?
|
||||||
const result = ???;
|
const result = future.???(io);
|
||||||
|
|
||||||
print("Task returned: {}\n", .{result});
|
print("Task returned: {}\n", .{result});
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -23,6 +23,11 @@
|
|||||||
// }
|
// }
|
||||||
// sel.cancelDiscard(); // cancel remaining, discard results
|
// sel.cancelDiscard(); // cancel remaining, discard results
|
||||||
//
|
//
|
||||||
|
// As with all async primitives: tasks spawned in a Select MUST
|
||||||
|
// be cleaned up. Use sel.cancel() to get remaining results one
|
||||||
|
// by one (for resource cleanup), or sel.cancelDiscard() if you
|
||||||
|
// don't need them.
|
||||||
|
//
|
||||||
// The buffer must be large enough for all tasks that might
|
// The buffer must be large enough for all tasks that might
|
||||||
// complete before you call cancelDiscard().
|
// complete before you call cancelDiscard().
|
||||||
//
|
//
|
||||||
@@ -47,7 +52,7 @@ pub fn main(init: std.process.Init) !void {
|
|||||||
|
|
||||||
// Wait for the first finisher.
|
// Wait for the first finisher.
|
||||||
// What Select method returns the first completed result?
|
// What Select method returns the first completed result?
|
||||||
const winner = ???;
|
const winner = try sel.???();
|
||||||
|
|
||||||
switch (winner) {
|
switch (winner) {
|
||||||
.hare => |msg| print("Hare: {s}\n", .{msg}),
|
.hare => |msg| print("Hare: {s}\n", .{msg}),
|
||||||
|
|||||||
@@ -5,22 +5,29 @@
|
|||||||
// The difference:
|
// The difference:
|
||||||
//
|
//
|
||||||
// io.async():
|
// io.async():
|
||||||
// * The function MAY run on another thread, or it may run
|
// * The function MAY run on a separate unit of concurrency,
|
||||||
// immediately on the current thread (synchronously).
|
// or it may run immediately on the caller (synchronously).
|
||||||
// * Never fails — if no thread is available, it just runs
|
// * Never fails — if no concurrency is available, it just
|
||||||
// the function right away.
|
// runs the function right away.
|
||||||
// * More portable, works with all Io backends.
|
// * More portable, works with all Io backends.
|
||||||
//
|
//
|
||||||
// io.concurrent():
|
// io.concurrent():
|
||||||
// * GUARANTEES a separate unit of concurrency (a real thread
|
// * GUARANTEES a separate unit of concurrency.
|
||||||
// in the Threaded backend).
|
|
||||||
// * Can fail with error.ConcurrencyUnavailable if resources
|
// * Can fail with error.ConcurrencyUnavailable if resources
|
||||||
// are exhausted or the backend doesn't support it.
|
// are exhausted or the backend doesn't support it.
|
||||||
// * Use when you NEED true parallelism.
|
// * Use when you NEED the task to run independently of the
|
||||||
|
// caller.
|
||||||
|
//
|
||||||
|
// What is a "unit of concurrency"? That depends on the backend!
|
||||||
|
// The Threaded backend uses OS threads. But the Evented backends
|
||||||
|
// (Uring, Kqueue, Dispatch) use M:N green threads / fibers,
|
||||||
|
// which can provide concurrency even on a SINGLE OS thread.
|
||||||
|
// Your code doesn't need to know the difference.
|
||||||
//
|
//
|
||||||
// Because concurrent() can fail, you must handle the error:
|
// Because concurrent() can fail, you must handle the error:
|
||||||
//
|
//
|
||||||
// var future = try io.concurrent(myFn, .{args});
|
// var future = try io.concurrent(myFn, .{args});
|
||||||
|
// defer _ = future.cancel(io);
|
||||||
// const result = future.await(io);
|
// const result = future.await(io);
|
||||||
//
|
//
|
||||||
// Notice the 'try' — that's the key difference in usage!
|
// Notice the 'try' — that's the key difference in usage!
|
||||||
@@ -33,25 +40,30 @@ const print = std.debug.print;
|
|||||||
pub fn main(init: std.process.Init) !void {
|
pub fn main(init: std.process.Init) !void {
|
||||||
const io = init.io;
|
const io = init.io;
|
||||||
|
|
||||||
// Launch with a guaranteed separate thread.
|
// Launch with a guaranteed separate unit of concurrency.
|
||||||
// Which Io method guarantees true concurrency?
|
// Which Io method guarantees this?
|
||||||
// (Hint: unlike io.async, this one can fail!)
|
// (Hint: unlike io.async, this one can fail!)
|
||||||
var future = try io.???(compute, .{io});
|
var future = try io.???(compute, .{io});
|
||||||
|
defer _ = future.cancel(io);
|
||||||
|
|
||||||
print("Main thread continues...\n", .{});
|
// Note: All breaks in this excercise (using sleep)
|
||||||
|
// are only necessary for a deterministic result.
|
||||||
// Wait 100 millisecond so the output order is deterministic.
|
|
||||||
io.sleep(std.Io.Duration.fromMilliseconds(100), .awake) catch {};
|
io.sleep(std.Io.Duration.fromMilliseconds(100), .awake) catch {};
|
||||||
|
|
||||||
print("Main thread done waiting.\n", .{});
|
print("Main continues...\n", .{});
|
||||||
|
|
||||||
|
// Wait 1 second for the output order.
|
||||||
|
io.sleep(std.Io.Duration.fromMilliseconds(200), .awake) catch {};
|
||||||
|
|
||||||
|
print("Main done waiting.\n", .{});
|
||||||
|
|
||||||
const result = future.await(io);
|
const result = future.await(io);
|
||||||
print("Result: {}\n", .{result});
|
print("Result: {}\n", .{result});
|
||||||
}
|
}
|
||||||
|
|
||||||
fn compute(io: std.Io) u32 {
|
fn compute(io: std.Io) u32 {
|
||||||
print("Computing on a separate thread!\n", .{});
|
print("Computing concurrently!\n", .{});
|
||||||
// Simulate some work.
|
// Simulate some work.
|
||||||
io.sleep(std.Io.Duration.fromMilliseconds(200), .awake) catch return 0;
|
io.sleep(std.Io.Duration.fromMilliseconds(400), .awake) catch return 0;
|
||||||
return 123;
|
return 123;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
//
|
//
|
||||||
// In exercise 088, we learned that cancellation happens at
|
// In exercise 089, we learned that cancellation happens at
|
||||||
// "cancellation points" — any Io function that can return
|
// "cancellation points" — any Io function that can return
|
||||||
// error.Canceled.
|
// error.Canceled.
|
||||||
//
|
//
|
||||||
@@ -11,7 +11,7 @@
|
|||||||
//
|
//
|
||||||
// const old = io.swapCancelProtection(.blocked);
|
// const old = io.swapCancelProtection(.blocked);
|
||||||
// defer _ = io.swapCancelProtection(old);
|
// defer _ = io.swapCancelProtection(old);
|
||||||
//
|
|
||||||
// // In this block, NO Io function will return error.Canceled.
|
// // In this block, NO Io function will return error.Canceled.
|
||||||
// // The cancel request is held until protection is restored.
|
// // The cancel request is held until protection is restored.
|
||||||
//
|
//
|
||||||
@@ -36,9 +36,10 @@ pub fn main(init: std.process.Init) !void {
|
|||||||
const io = init.io;
|
const io = init.io;
|
||||||
|
|
||||||
var future = io.async(importantTask, .{io});
|
var future = io.async(importantTask, .{io});
|
||||||
|
defer _ = future.cancel(io);
|
||||||
|
|
||||||
// Give the task time to start and enter its critical section.
|
// Give the task time to start and enter its critical section.
|
||||||
io.sleep(std.Io.Duration.fromMilliseconds(300), .awake) catch {};
|
io.sleep(std.Io.Duration.fromMilliseconds(200), .awake) catch {};
|
||||||
|
|
||||||
// Cancel while the task is in its protected section.
|
// Cancel while the task is in its protected section.
|
||||||
const result = future.cancel(io);
|
const result = future.cancel(io);
|
||||||
@@ -50,12 +51,12 @@ fn importantTask(io: std.Io) []const u8 {
|
|||||||
|
|
||||||
// Protect this section from cancellation.
|
// Protect this section from cancellation.
|
||||||
// What method swaps the cancel protection state?
|
// What method swaps the cancel protection state?
|
||||||
const old = io.???(. blocked);
|
const old = io.???(.blocked);
|
||||||
defer _ = io.???(old);
|
defer _ = io.???(old);
|
||||||
|
|
||||||
// This sleep will NOT return error.Canceled even though
|
// This sleep will NOT return error.Canceled even though
|
||||||
// we get canceled during it — protection is active!
|
// we get canceled during it — protection is active!
|
||||||
io.sleep(std.Io.Duration.fromMilliseconds(600), .awake) catch |err| switch (err) {
|
io.sleep(std.Io.Duration.fromMilliseconds(300), .awake) catch |err| switch (err) {
|
||||||
error.Canceled => {
|
error.Canceled => {
|
||||||
// This should never happen while protected!
|
// This should never happen while protected!
|
||||||
return "ERROR: canceled during critical section!";
|
return "ERROR: canceled during critical section!";
|
||||||
|
|||||||
@@ -16,16 +16,17 @@
|
|||||||
// by a grasshopper) and left several bugs. Can you fix them?
|
// by a grasshopper) and left several bugs. Can you fix them?
|
||||||
//
|
//
|
||||||
// Here's what the program should do:
|
// Here's what the program should do:
|
||||||
// 1. Three sensor tasks run concurrently, each sending
|
// 1. Three sensor tasks send exactly 3 readings each through
|
||||||
// exactly 3 readings through a Queue
|
// a Queue
|
||||||
// 2. A collector task receives readings, protected by a Mutex
|
// 2. A collector task receives readings concurrently,
|
||||||
|
// protected by a Mutex
|
||||||
// 3. After all sensors finish, the queue is closed
|
// 3. After all sensors finish, the queue is closed
|
||||||
// 4. The final report is written in a cancel-protected section
|
// 4. The final report is written in a cancel-protected section
|
||||||
//
|
//
|
||||||
// *************************************************************
|
// *************************************************************
|
||||||
// * A NOTE ABOUT THIS EXERCISE *
|
// * A NOTE ABOUT THIS EXERCISE *
|
||||||
// * *
|
// * *
|
||||||
// * This quiz uses concepts from exercises 084-093. *
|
// * This quiz uses concepts from exercises 085-094. *
|
||||||
// * There are 6 bugs to fix — look for the ???s! *
|
// * There are 6 bugs to fix — look for the ???s! *
|
||||||
// * *
|
// * *
|
||||||
// *************************************************************
|
// *************************************************************
|
||||||
@@ -50,8 +51,8 @@ const GardenWeather = struct {
|
|||||||
fn addReading(self: *GardenWeather, io: std.Io, reading: Reading) void {
|
fn addReading(self: *GardenWeather, io: std.Io, reading: Reading) void {
|
||||||
// Bug 1: The collector needs to lock before modifying
|
// Bug 1: The collector needs to lock before modifying
|
||||||
// shared state. What Mutex method acquires the lock?
|
// shared state. What Mutex method acquires the lock?
|
||||||
self.mutex.lock(io) catch return;
|
|
||||||
self.mutex.???(io) catch return;
|
self.mutex.???(io) catch return;
|
||||||
|
defer self.mutex.unlock(io);
|
||||||
|
|
||||||
switch (reading.sensor_type) {
|
switch (reading.sensor_type) {
|
||||||
.thermometer => self.temperature = reading.value,
|
.thermometer => self.temperature = reading.value,
|
||||||
@@ -70,39 +71,40 @@ pub fn main(init: std.process.Init) !void {
|
|||||||
var reading_buf: [8]Reading = undefined;
|
var reading_buf: [8]Reading = undefined;
|
||||||
var queue: std.Io.Queue(Reading) = .init(&reading_buf);
|
var queue: std.Io.Queue(Reading) = .init(&reading_buf);
|
||||||
|
|
||||||
// Sensor group: runs all three sensors to completion.
|
// The collector must run concurrently so it can process
|
||||||
|
// readings while the sensors are still sending.
|
||||||
|
// Start it FIRST to ensure its concurrency unit is reserved.
|
||||||
|
//
|
||||||
|
// Bug 2: The collector needs guaranteed concurrency.
|
||||||
|
// What method ensures a separate unit of concurrency?
|
||||||
|
// (Don't forget: it can fail!)
|
||||||
|
var collector_future = try io.???(collector, .{ io, &queue, &weather });
|
||||||
|
defer _ = collector_future.cancel(io);
|
||||||
|
|
||||||
|
// Sensor group: the sensors can use async — they just need
|
||||||
|
// to run, and async is more portable.
|
||||||
var sensors: std.Io.Group = .init;
|
var sensors: std.Io.Group = .init;
|
||||||
|
|
||||||
// Start three sensor tasks. They need GUARANTEED concurrency
|
sensors.async(io, sensor, .{ io, &queue, .thermometer, 20 });
|
||||||
// since they each simulate real-time measurement.
|
sensors.async(io, sensor, .{ io, &queue, .hygrometer, 60 });
|
||||||
//
|
sensors.async(io, sensor, .{ io, &queue, .anemometer, 10 });
|
||||||
// Bug 2: io.async doesn't guarantee a separate thread.
|
|
||||||
// Which Io method guarantees true concurrency?
|
|
||||||
// (Don't forget: it can fail, so you need 'try'!)
|
|
||||||
try sensors.???(io, sensor, .{ io, &queue, .thermometer, 20 });
|
|
||||||
try sensors.???(io, sensor, .{ io, &queue, .hygrometer, 60 });
|
|
||||||
try sensors.???(io, sensor, .{ io, &queue, .anemometer, 10 });
|
|
||||||
|
|
||||||
// Collector group: processes readings from the queue.
|
|
||||||
var collectors: std.Io.Group = .init;
|
|
||||||
collectors.async(io, collector, .{ io, &queue, &weather });
|
|
||||||
|
|
||||||
// Bug 3: Wait for ALL sensors to finish sending their readings.
|
// Bug 3: Wait for ALL sensors to finish sending their readings.
|
||||||
// What Group method blocks until all tasks complete?
|
// What Group method blocks until all tasks complete?
|
||||||
try sensors.await(io);
|
try sensors.???(io);
|
||||||
// try sensors.???(io);
|
|
||||||
|
|
||||||
// All sensors done — close the queue so the collector knows
|
// All sensors done — close the queue so the collector knows
|
||||||
// there's no more data coming.
|
// there's no more data coming.
|
||||||
queue.close(io);
|
queue.close(io);
|
||||||
|
|
||||||
// Wait for the collector to drain the queue.
|
// Wait for the collector to drain the remaining queue.
|
||||||
try collectors.await(io);
|
_ = collector_future.await(io);
|
||||||
|
// _ = collector_future.???(io);
|
||||||
|
|
||||||
// Now write the garden report. This is critical — it must
|
// Now write the garden report. This is critical — it must
|
||||||
// NOT be interrupted, even if something tries to cancel us!
|
// NOT be interrupted, even if something tries to cancel us!
|
||||||
//
|
//
|
||||||
// Bug 4: Protect this section from cancellation.
|
// Bug 5: Protect this section from cancellation.
|
||||||
// What Io method swaps the cancel protection state?
|
// What Io method swaps the cancel protection state?
|
||||||
const old_protection = io.???(.blocked);
|
const old_protection = io.???(.blocked);
|
||||||
defer _ = io.???(old_protection);
|
defer _ = io.???(old_protection);
|
||||||
@@ -125,7 +127,7 @@ fn sensor(
|
|||||||
.value = base_value + @as(i32, @intCast(i)),
|
.value = base_value + @as(i32, @intCast(i)),
|
||||||
};
|
};
|
||||||
|
|
||||||
// Bug 5: Send the reading into the queue.
|
// Bug 6: Send the reading into the queue.
|
||||||
// What Queue method sends a single element?
|
// What Queue method sends a single element?
|
||||||
queue.???(io, reading) catch return;
|
queue.???(io, reading) catch return;
|
||||||
}
|
}
|
||||||
@@ -163,8 +165,7 @@ fn printGardenReport(weather: *GardenWeather) void {
|
|||||||
//
|
//
|
||||||
// This quiz covered the main async I/O primitives:
|
// This quiz covered the main async I/O primitives:
|
||||||
// io.async() - launch a task (may run inline)
|
// io.async() - launch a task (may run inline)
|
||||||
// io.concurrent() - launch with guaranteed parallelism
|
// io.concurrent() - guaranteed unit of concurrency
|
||||||
// Group.concurrent() - concurrent tasks in a group
|
|
||||||
// Future.await/cancel - collect or cancel a single task
|
// Future.await/cancel - collect or cancel a single task
|
||||||
// Group.async/await/cancel - manage fire-and-forget tasks
|
// Group.async/await/cancel - manage fire-and-forget tasks
|
||||||
// Select.async/await - race tasks, act on first completion
|
// Select.async/await - race tasks, act on first completion
|
||||||
@@ -180,7 +181,8 @@ fn printGardenReport(weather: *GardenWeather) void {
|
|||||||
// Batch - submit multiple I/O operations at once
|
// Batch - submit multiple I/O operations at once
|
||||||
//
|
//
|
||||||
// The key insight: all of these work through the Io VTable,
|
// The key insight: all of these work through the Io VTable,
|
||||||
// so your code is portable across backends (Threaded, Uring,
|
// so your code is portable across backends — whether Threaded
|
||||||
// Kqueue, Dispatch) without any changes!
|
// (OS thread pool), or Evented (M:N green threads / fibers
|
||||||
|
// that can provide concurrency even on a single OS thread).
|
||||||
//
|
//
|
||||||
// Doctor Zoraptera approves.
|
// Doctor Zoraptera approves.
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
--- exercises/085_async.zig 2026-04-01 20:40:08.904999609 +0200
|
--- exercises/085_async.zig 2026-04-04 16:01:01.509555724 +0200
|
||||||
+++ answers/085_async.zig 2026-04-01 20:40:05.641933231 +0200
|
+++ answers/085_async.zig 2026-04-04 16:00:58.541495688 +0200
|
||||||
@@ -37,7 +37,7 @@
|
@@ -38,7 +38,7 @@
|
||||||
const std = @import("std");
|
const std = @import("std");
|
||||||
|
|
||||||
pub fn main(init: std.process.Init) !void {
|
pub fn main(init: std.process.Init) !void {
|
||||||
|
|||||||
@@ -1,11 +1,11 @@
|
|||||||
--- exercises/086_async2.zig 2026-04-03 19:42:15.274532915 +0200
|
--- exercises/086_async2.zig 2026-04-05 12:41:11.350626443 +0200
|
||||||
+++ answers/086_async2.zig 2026-04-03 21:30:18.180019206 +0200
|
+++ answers/086_async2.zig 2026-04-05 12:42:00.879791167 +0200
|
||||||
@@ -38,7 +38,7 @@
|
@@ -44,7 +44,7 @@
|
||||||
|
|
||||||
// Now collect the result. What method on Future gives us
|
// Now collect the result. What method on Future gives us
|
||||||
// the value, blocking if it isn't ready yet?
|
// the value, blocking until it's ready?
|
||||||
- const answer = future.???(io);
|
- const answer = future.???(io);
|
||||||
+ const answer = future.await(io);
|
+ const answer = future.await(io);
|
||||||
|
|
||||||
std.debug.print("The answer is: {}\n", .{answer});
|
print("The answer is: {}\n", .{answer});
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,18 +1,11 @@
|
|||||||
--- exercises/087_async3.zig 2026-04-01 22:51:05.540094851 +0200
|
--- exercises/087_async3.zig 2026-04-05 16:12:48.317265515 +0200
|
||||||
+++ answers/087_async3.zig 2026-04-01 22:50:44.579669189 +0200
|
+++ answers/087_async3.zig 2026-04-05 16:12:52.269343030 +0200
|
||||||
@@ -29,12 +29,12 @@
|
@@ -28,7 +28,7 @@
|
||||||
const io = init.io;
|
|
||||||
|
|
||||||
// Launch both tasks asynchronously.
|
// Launch both tasks asynchronously.
|
||||||
- var future_a = io.async(slowAdd, .{ 10, 20 });
|
var future_a = io.async(slowAdd, .{ 1, 2 });
|
||||||
|
defer _ = future_a.cancel(io);
|
||||||
- var future_b = ???(slowMul, .{ 6, 7 });
|
- var future_b = ???(slowMul, .{ 6, 7 });
|
||||||
+ var future_a = io.async(slowAdd, .{ 1, 2 });
|
|
||||||
+ var future_b = io.async(slowMul, .{ 6, 7 });
|
+ var future_b = io.async(slowMul, .{ 6, 7 });
|
||||||
|
defer _ = future_b.cancel(io);
|
||||||
|
|
||||||
// Await both results.
|
// Await both results.
|
||||||
const sum = future_a.await(io);
|
|
||||||
- const product = future_b.???(io);
|
|
||||||
+ const product = future_b.await(io);
|
|
||||||
|
|
||||||
print("{} + {} = {}\n", .{ 1, 2, sum });
|
|
||||||
print("{} * {} = {}\n", .{ 6, 7, product });
|
|
||||||
|
|||||||
@@ -1,10 +1,10 @@
|
|||||||
--- exercises/088_async4.zig 2026-04-01 23:17:31.066443941 +0200
|
--- exercises/088_async4.zig 2026-04-06 12:22:06.643385622 +0200
|
||||||
+++ answers/088_async4.zig 2026-04-01 23:17:39.251612131 +0200
|
+++ answers/088_async4.zig 2026-04-06 12:22:11.820491035 +0200
|
||||||
@@ -38,7 +38,7 @@
|
@@ -38,7 +38,7 @@
|
||||||
|
|
||||||
// Wait for all tasks to finish.
|
// Wait for all tasks to finish.
|
||||||
// What Group method blocks until all tasks complete?
|
// What Group method blocks until all tasks complete?
|
||||||
- try group.???
|
- try group.???(io);
|
||||||
+ try group.await(io);
|
+ try group.await(io);
|
||||||
|
|
||||||
print("All tasks finished!\n", .{});
|
print("All tasks finished!\n", .{});
|
||||||
|
|||||||
@@ -1,10 +1,10 @@
|
|||||||
--- exercises/089_async5.zig 2026-04-01 23:40:40.505855238 +0200
|
--- exercises/089_async5.zig 2026-04-06 14:38:54.443726849 +0200
|
||||||
+++ answers/089_async5.zig 2026-04-01 23:40:10.176236971 +0200
|
+++ answers/089_async5.zig 2026-04-06 14:38:39.945438309 +0200
|
||||||
@@ -40,7 +40,7 @@
|
@@ -46,7 +46,7 @@
|
||||||
|
|
||||||
// We don't want to wait 10 seconds!
|
// We don't want to wait 10 seconds!
|
||||||
// Which Future method requests cancellation AND returns the result?
|
// Which Future method requests cancellation AND returns the result?
|
||||||
- const result = ???;
|
- const result = future.???(io);
|
||||||
+ const result = future.cancel(io);
|
+ const result = future.cancel(io);
|
||||||
|
|
||||||
print("Task returned: {}\n", .{result});
|
print("Task returned: {}\n", .{result});
|
||||||
|
|||||||
@@ -1,10 +1,10 @@
|
|||||||
--- exercises/090_async6.zig 2026-04-02 10:25:34.016616118 +0200
|
--- exercises/090_async6.zig 2026-04-06 18:49:37.232023422 +0200
|
||||||
+++ answers/090_async6.zig 2026-04-02 10:27:48.827144051 +0200
|
+++ answers/090_async6.zig 2026-04-06 18:49:22.189720687 +0200
|
||||||
@@ -47,7 +47,7 @@
|
@@ -52,7 +52,7 @@
|
||||||
|
|
||||||
// Wait for the first finisher.
|
// Wait for the first finisher.
|
||||||
// What Select method returns the first completed result?
|
// What Select method returns the first completed result?
|
||||||
- const winner = ???;
|
- const winner = try sel.???();
|
||||||
+ const winner = try sel.await();
|
+ const winner = try sel.await();
|
||||||
|
|
||||||
switch (winner) {
|
switch (winner) {
|
||||||
|
|||||||
@@ -1,11 +1,11 @@
|
|||||||
--- exercises/093_async9.zig 2026-04-03 13:44:50.526780809 +0200
|
--- exercises/093_async9.zig 2026-04-06 19:26:11.388025362 +0200
|
||||||
+++ answers/093_async9.zig 2026-04-03 13:44:54.957870294 +0200
|
+++ answers/093_async9.zig 2026-04-06 19:18:36.242931688 +0200
|
||||||
@@ -36,7 +36,7 @@
|
@@ -43,7 +43,7 @@
|
||||||
// Launch with a guaranteed separate thread.
|
// Launch with a guaranteed separate unit of concurrency.
|
||||||
// Which Io method guarantees true concurrency?
|
// Which Io method guarantees this?
|
||||||
// (Hint: unlike io.async, this one can fail!)
|
// (Hint: unlike io.async, this one can fail!)
|
||||||
- var future = try io.???(compute, .{io});
|
- var future = try io.???(compute, .{io});
|
||||||
+ var future = try io.concurrent(compute, .{io});
|
+ var future = try io.concurrent(compute, .{io});
|
||||||
|
defer _ = future.cancel(io);
|
||||||
|
|
||||||
print("Main thread continues...\n", .{});
|
// Note: All breaks in this excercise (using sleep)
|
||||||
|
|
||||||
|
|||||||
@@ -1,10 +1,10 @@
|
|||||||
--- exercises/094_async10.zig 2026-04-03 14:25:16.600025924 +0200
|
--- exercises/094_async10.zig 2026-04-06 19:36:59.873966580 +0200
|
||||||
+++ answers/094_async10.zig 2026-04-03 14:24:56.192615893 +0200
|
+++ answers/094_async10.zig 2026-04-06 19:37:12.416216872 +0200
|
||||||
@@ -50,8 +50,8 @@
|
@@ -51,8 +51,8 @@
|
||||||
|
|
||||||
// Protect this section from cancellation.
|
// Protect this section from cancellation.
|
||||||
// What method swaps the cancel protection state?
|
// What method swaps the cancel protection state?
|
||||||
- const old = io.???(. blocked);
|
- const old = io.???(.blocked);
|
||||||
- defer _ = io.???(old);
|
- defer _ = io.???(old);
|
||||||
+ const old = io.swapCancelProtection(.blocked);
|
+ const old = io.swapCancelProtection(.blocked);
|
||||||
+ defer _ = io.swapCancelProtection(old);
|
+ defer _ = io.swapCancelProtection(old);
|
||||||
|
|||||||
@@ -1,38 +1,42 @@
|
|||||||
--- exercises/095_quiz_async.zig 2026-04-03 18:04:53.577391455 +0200
|
--- exercises/095_quiz_async.zig 2026-04-06 19:55:17.111817364 +0200
|
||||||
+++ answers/095_quiz_async.zig 2026-04-03 18:05:42.570392172 +0200
|
+++ answers/095_quiz_async.zig 2026-04-06 19:56:16.063974543 +0200
|
||||||
@@ -51,7 +51,7 @@
|
@@ -51,7 +51,7 @@
|
||||||
|
fn addReading(self: *GardenWeather, io: std.Io, reading: Reading) void {
|
||||||
// Bug 1: The collector needs to lock before modifying
|
// Bug 1: The collector needs to lock before modifying
|
||||||
// shared state. What Mutex method acquires the lock?
|
// shared state. What Mutex method acquires the lock?
|
||||||
self.mutex.lock(io) catch return;
|
|
||||||
- self.mutex.???(io) catch return;
|
- self.mutex.???(io) catch return;
|
||||||
+ defer self.mutex.unlock(io);
|
+ self.mutex.lock(io) catch return;
|
||||||
|
defer self.mutex.unlock(io);
|
||||||
|
|
||||||
switch (reading.sensor_type) {
|
switch (reading.sensor_type) {
|
||||||
.thermometer => self.temperature = reading.value,
|
@@ -78,7 +78,7 @@
|
||||||
@@ -79,9 +79,9 @@
|
// Bug 2: The collector needs guaranteed concurrency.
|
||||||
// Bug 2: io.async doesn't guarantee a separate thread.
|
// What method ensures a separate unit of concurrency?
|
||||||
// Which Io method guarantees true concurrency?
|
// (Don't forget: it can fail!)
|
||||||
// (Don't forget: it can fail, so you need 'try'!)
|
- var collector_future = try io.???(collector, .{ io, &queue, &weather });
|
||||||
- try sensors.???(io, sensor, .{ io, &queue, .thermometer, 20 });
|
+ var collector_future = try io.concurrent(collector, .{ io, &queue, &weather });
|
||||||
- try sensors.???(io, sensor, .{ io, &queue, .hygrometer, 60 });
|
defer _ = collector_future.cancel(io);
|
||||||
- try sensors.???(io, sensor, .{ io, &queue, .anemometer, 10 });
|
|
||||||
+ try sensors.concurrent(io, sensor, .{ io, &queue, .thermometer, 20 });
|
// Sensor group: the sensors can use async — they just need
|
||||||
+ try sensors.concurrent(io, sensor, .{ io, &queue, .hygrometer, 60 });
|
@@ -91,7 +91,7 @@
|
||||||
+ try sensors.concurrent(io, sensor, .{ io, &queue, .anemometer, 10 });
|
|
||||||
|
|
||||||
// Collector group: processes readings from the queue.
|
|
||||||
var collectors: std.Io.Group = .init;
|
|
||||||
@@ -90,7 +90,6 @@
|
|
||||||
// Bug 3: Wait for ALL sensors to finish sending their readings.
|
// Bug 3: Wait for ALL sensors to finish sending their readings.
|
||||||
// What Group method blocks until all tasks complete?
|
// What Group method blocks until all tasks complete?
|
||||||
try sensors.await(io);
|
- try sensors.???(io);
|
||||||
- // try sensors.???(io);
|
+ try sensors.await(io);
|
||||||
|
|
||||||
// All sensors done — close the queue so the collector knows
|
// All sensors done — close the queue so the collector knows
|
||||||
// there's no more data coming.
|
// there's no more data coming.
|
||||||
@@ -104,8 +103,8 @@
|
@@ -99,15 +99,14 @@
|
||||||
|
|
||||||
|
// Wait for the collector to drain the remaining queue.
|
||||||
|
_ = collector_future.await(io);
|
||||||
|
- // _ = collector_future.???(io);
|
||||||
|
|
||||||
|
// Now write the garden report. This is critical — it must
|
||||||
|
// NOT be interrupted, even if something tries to cancel us!
|
||||||
//
|
//
|
||||||
// Bug 4: Protect this section from cancellation.
|
// Bug 5: Protect this section from cancellation.
|
||||||
// What Io method swaps the cancel protection state?
|
// What Io method swaps the cancel protection state?
|
||||||
- const old_protection = io.???(.blocked);
|
- const old_protection = io.???(.blocked);
|
||||||
- defer _ = io.???(old_protection);
|
- defer _ = io.???(old_protection);
|
||||||
@@ -41,9 +45,9 @@
|
|||||||
|
|
||||||
printGardenReport(&weather);
|
printGardenReport(&weather);
|
||||||
}
|
}
|
||||||
@@ -127,7 +126,7 @@
|
@@ -129,7 +128,7 @@
|
||||||
|
|
||||||
// Bug 5: Send the reading into the queue.
|
// Bug 6: Send the reading into the queue.
|
||||||
// What Queue method sends a single element?
|
// What Queue method sends a single element?
|
||||||
- queue.???(io, reading) catch return;
|
- queue.???(io, reading) catch return;
|
||||||
+ queue.putOne(io, reading) catch return;
|
+ queue.putOne(io, reading) catch return;
|
||||||
|
|||||||
Reference in New Issue
Block a user