new example for concurrency

exercises/093_async9.zig

@@ -30,40 +30,74 @@
 //     defer _ = future.cancel(io);
 //     const result = future.await(io);
 //
-// Notice the 'try' — that's the key difference in usage!
+// Let's try a slightly simplified example from signal processing:
+// Suppose we're looking for the beginning of a signal above the noise
+// level. To do this, we compare each entry from beginning to end with
+// the threshold.To speed things up a bit, we split the signal into
+// two halves and have two parallel workers search for them.
+// Who finds the beginning first "wins" and thus ends the other one.
 //
-// Fix this program to launch the computation concurrently.
+// As I said, this is a simplified explanation,
+// but in practice it's done more or less like this.
 //
 const std = @import("std");
+const Io = std.Io;
 const print = std.debug.print;
 
+const SearchResult = struct {
+    worker_id: u8,
+    index: usize,
+};
+
 pub fn main(init: std.process.Init) !void {
     const io = init.io;
 
-    // Launch with a guaranteed separate unit of concurrency.
-    // Which Io method guarantees this?
-    // (Hint: unlike io.async, this one can fail!)
-    var future = try io.???(compute, .{io});
-    defer _ = future.cancel(io);
+    const data = [_]u32{ 10, 23, 45, 67, 12, 69, 3, 54, 69, 42, 68, 56, 71, 79, 79, 75, 70, 77 };
+    const threshold = 70;
+    const mid = data.len / 2;
 
-    // Note: All breaks in this exercise (using sleep)
-    // are only necessary for a deterministic result.
-    io.sleep(std.Io.Duration.fromMilliseconds(100), .awake) catch {};
+    // A queue with space for one result.
+    var buf: [1]SearchResult = undefined;
+    var queue = Io.Queue(SearchResult).init(&buf);
 
-    print("Main continues...\n", .{});
+    // Launch two workers, each searching half the array.
+    var f1 = ???(searchRange, .{ data[0..mid], target, 0, 0, &queue, io });
+    defer _ = f1.cancel(io);
 
-    // Wait 1 second for the output order.
-    io.sleep(std.Io.Duration.fromMilliseconds(200), .awake) catch {};
+    var f2 = ???(searchRange, .{ data[mid..], target, mid, 1, &queue, io });
+    defer _ = f2.cancel(io);
 
-    print("Main done waiting.\n", .{});
+    // Wait for the first result.
+    const result = try queue.getOne(io);
 
-    const result = future.await(io);
-    print("Result: {}\n", .{result});
+    print("Worker {} found signal start over threshold at index {}!\n", .{ result.worker_id, result.index });
 }
 
-fn compute(io: std.Io) u32 {
-    print("Computing concurrently!\n", .{});
-    // Simulate some work.
-    io.sleep(std.Io.Duration.fromMilliseconds(400), .awake) catch return 0;
-    return 123;
+fn searchThreshold(
+    io: Io,
+    slice: []const u32,
+    threshold: u32,
+    base_offset: usize,
+    worker_id: u8,
+    queue: *Io.Queue(SearchResult),
+) void {
+    for (slice, 0..) |val, i| {
+        // This pause is necessary so that the process can be canceled
+        // if another one has already finished. Without this pause,
+        // all workers would continue until the end.
+        io.sleep(Io.Duration.fromMilliseconds(1), .awake) catch return;
+
+        // To test this, you can view the work of the workers
+        // and then comment out the pause.
+        // print("id: {} - val: {}\n", .{ worker_id, val });
+
+        if (val >= threshold) {
+            queue.putOne(io, .{
+                .worker_id = worker_id,
+                .index = base_offset + i,
+            }) catch return;
+            return;
+        }
+    }
 }
+