esp: RTOS work (#907)

- yield when a higher priority task wakes in sync primitives
- fix priority inheritance in `Mutex`
- refactor sync primitive timeouts
- add `Signal` sync primitive (alternative to direct to task
notifications)
- remove `Condition` (as well as the event waiting system from wifi
which seemed a bit hacky)
- foundation libc: implement `strnlen`

Author: tact1m4n3

examples/espressif/esp/src/lwip/exports.zig

@@ -113,16 +113,13 @@ export fn sys_mbox_set_invalid(ptr: *c.sys_mbox_t) void {
 
 export fn sys_mbox_post(ptr: *c.sys_mbox_t, element: MailboxElement) void {
     const mailbox: *Mailbox = @ptrCast(@alignCast(ptr.*));
-    mailbox.put_one(element, null) catch unreachable;
+    mailbox.put_one(element, .never) catch unreachable;
 }
 
 export fn sys_mbox_trypost(ptr: *c.sys_mbox_t, element: MailboxElement) c.err_t {
     const mailbox: *Mailbox = @ptrCast(@alignCast(ptr.*));
-    if (mailbox.put_one_non_blocking(element)) {
-        return c.ERR_OK;
-    } else {
-        return c.ERR_MEM;
-    }
+    mailbox.put_one(element, .non_blocking) catch return c.ERR_MEM;
+    return c.ERR_OK;
 }
 
 comptime {
@@ -132,7 +129,7 @@ comptime {
 export fn sys_arch_mbox_fetch(ptr: *c.sys_mbox_t, element_ptr: *MailboxElement, timeout: u32) u32 {
     const mailbox: *Mailbox = @ptrCast(@alignCast(ptr.*));
     const now = esp.time.get_time_since_boot();
-    element_ptr.* = mailbox.get_one(if (timeout != 0) .from_ms(timeout) else null) catch {
+    element_ptr.* = mailbox.get_one(if (timeout != 0) .{ .after = .from_ms(timeout) } else .never) catch {
         return c.SYS_ARCH_TIMEOUT;
     };
     // returns waiting time in ms
@@ -141,12 +138,9 @@ export fn sys_arch_mbox_fetch(ptr: *c.sys_mbox_t, element_ptr: *MailboxElement,
 
 export fn sys_arch_mbox_tryfetch(ptr: *c.sys_mbox_t, element_ptr: *MailboxElement) u32 {
     const mailbox: *Mailbox = @ptrCast(@alignCast(ptr.*));
-    if (mailbox.get_one_non_blocking()) |element| {
-        element_ptr.* = element;
-        return 0;
-    } else {
-        return c.SYS_MBOX_EMPTY;
-    }
+    const element = mailbox.get_one(.non_blocking) catch return c.SYS_MBOX_EMPTY;
+    element_ptr.* = element;
+    return 0;
 }
 
 export fn sys_sem_new(ptr: *c.sys_sem_t, count: u8) c.err_t {
@@ -172,7 +166,7 @@ export fn sys_sem_signal(ptr: *c.sys_sem_t) void {
 export fn sys_arch_sem_wait(ptr: *c.sys_sem_t, timeout: u32) u32 {
     const sem: *rtos.Semaphore = @ptrCast(@alignCast(ptr.*));
     const now = esp.time.get_time_since_boot();
-    sem.take_with_timeout(if (timeout != 0) .from_ms(timeout) else null) catch {
+    sem.take_with_timeout(if (timeout != 0) .{ .after = .from_ms(timeout) } else .never) catch {
         return c.SYS_ARCH_TIMEOUT;
     };
     // returns waiting time in ms

examples/espressif/esp/src/rtos.zig

@@ -8,7 +8,7 @@ const rtos = esp.rtos;
 pub const microzig_options: microzig.Options = .{
     .logFn = usb_serial_jtag.logger.log,
     .interrupts = .{
-        .interrupt31 = rtos.tick_interrupt_handler,
+        .interrupt31 = rtos.interrupt_handler,
     },
     .log_level = .debug,
     .cpu = .{
@@ -26,7 +26,7 @@ pub const microzig_options: microzig.Options = .{
 
 fn task1(queue: *rtos.Queue(u32)) void {
     for (0..5) |i| {
-        queue.put_one(i, null) catch unreachable;
+        queue.put_one(i, .never) catch unreachable;
         rtos.sleep(.from_ms(500));
     }
 }
@@ -44,7 +44,7 @@ pub fn main() !void {
     defer rtos.wait_and_free(gpa, task);
 
     while (true) {
-        const item = try queue.get_one(.from_ms(1000));
+        const item = try queue.get_one(.{ .after = .from_ms(1000) });
         std.log.info("got item: {}", .{item});
     }
 }

examples/espressif/esp/src/tcp_server.zig

@@ -24,7 +24,7 @@ pub const microzig_options: microzig.Options = .{
     .logFn = usb_serial_jtag.logger.log,
     .interrupts = .{
         .interrupt30 = radio.interrupt_handler,
-        .interrupt31 = rtos.tick_interrupt_handler,
+        .interrupt31 = rtos.interrupt_handler,
     },
     .cpu = .{
         .interrupt_stack = .{
@@ -52,8 +52,7 @@ const SERVER_PORT = 3333;
 
 var maybe_netif: ?*lwip.c.netif = null;
 
-var ip_ready_semaphore: rtos.Semaphore = .init(0, 1);
-var ip: lwip.c.ip_addr_t = undefined;
+var wifi_ready_signal: rtos.Signal(void) = .{};
 
 extern fn netconn_new_with_proto_and_callback(t: lwip.c.enum_netconn_type, proto: lwip.c.u8_t, callback: ?*const anyopaque) [*c]lwip.c.struct_netconn;
 pub fn main() !void {
@@ -86,7 +85,11 @@ pub fn main() !void {
             .auth_method = AUTH_METHOD,
         },
     });
-    try radio.wifi.start_blocking();
+
+    try radio.wifi.start();
+    while (!radio.wifi.get_sta_state().is_started()) {
+        rtos.sleep(.from_ms(100));
+    }
 
     {
         std.log.info("Scanning for access points...", .{});
@@ -102,10 +105,14 @@ pub fn main() !void {
         }
     }
 
-    try radio.wifi.connect_blocking();
-    try lwip.c_err(lwip.c.netifapi_netif_common(&netif, lwip.c.netif_set_link_up, null));
+    try radio.wifi.connect();
+    while (radio.wifi.get_sta_state() != .connected) {
+        rtos.sleep(.from_ms(100));
+    }
+    _ = lwip.c.netifapi_netif_common(maybe_netif.?, lwip.c.netif_set_link_up, null);
+
+    wifi_ready_signal.wait();
 
-    ip_ready_semaphore.take();
     std.log.info("Listening on {f}:{}", .{ IP_Formatter.init(netif.ip_addr), SERVER_PORT });
 
     const server_conn = netconn_new_with_proto_and_callback(lwip.c.NETCONN_TCP, 0, null) orelse {
@@ -164,6 +171,13 @@ fn handle_client(conn: ?*lwip.c.netconn) !void {
     }
 }
 
+fn netif_status_callback(netif_ptr: [*c]lwip.c.netif) callconv(.c) void {
+    const netif = &netif_ptr[0];
+    if (netif.ip_addr.u_addr.ip4.addr != 0) {
+        wifi_ready_signal.put({});
+    }
+}
+
 fn on_packet_received(comptime _: radio.wifi.Interface, data: []const u8) void {
     const pbuf: *lwip.c.struct_pbuf = lwip.c.pbuf_alloc(lwip.c.PBUF_RAW, @intCast(data.len), lwip.c.PBUF_POOL) orelse {
         std.log.err("failed to allocate receive pbuf", .{});
@@ -197,14 +211,6 @@ fn netif_init(netif_ptr: [*c]lwip.c.struct_netif) callconv(.c) lwip.c.err_t {
     return lwip.c.ERR_OK;
 }
 
-fn netif_status_callback(netif_ptr: [*c]lwip.c.netif) callconv(.c) void {
-    const netif = &netif_ptr[0];
-    if (netif.ip_addr.u_addr.ip4.addr != 0) {
-        ip = netif.ip_addr;
-        ip_ready_semaphore.give();
-    }
-}
-
 var packet_buf: [1600]u8 = undefined;
 
 fn netif_output(_: [*c]lwip.c.struct_netif, pbuf_c: [*c]lwip.c.struct_pbuf) callconv(.c) lwip.c.err_t {

modules/foundation-libc/include/string.h

@@ -15,6 +15,7 @@ char * strcpy(char * restrict s1, char const * restrict s2);
 size_t strcspn(char const * s1, char const * s2);
 char * strerror(int errnum);
 size_t strlen(char const * str);
+size_t strnlen(char const * str, size_t n);
 char * strncat(char * restrict s1, char const * restrict s2, size_t n);
 int    strncmp(char const * lhs, char const * rhs, size_t count);
 char * strncpy(char * restrict s1, char const * restrict s2, size_t n);

modules/foundation-libc/src/modules/string.zig

@@ -44,6 +44,11 @@ export fn strlen(str: ?[*:0]const c_char) usize {
     return std.mem.len(s);
 }
 
+export fn strnlen(str: ?[*:0]const u8, n: usize) usize {
+    const s = str orelse return 0;
+    return if (std.mem.indexOfScalar(u8, s[0..n], 0)) |index| index else n;
+}
+
 /// https://en.cppreference.com/w/c/string/byte/strncmp
 export fn strncmp(lhs: ?[*:0]const c_char, rhs: ?[*:0]const c_char, count: usize) c_int {
     const lhs_s: [*:0]const u8 = @ptrCast(lhs orelse return if (rhs != null) 1 else 0);

port/espressif/esp/src/hal.zig

@@ -116,4 +116,5 @@ fn str(comptime l: usize, comptime s: []const u8) [l]u8 {
 
 test {
     _ = rtos;
+    _ = radio;
 }

port/espressif/esp/src/hal/radio.zig

@@ -166,3 +166,7 @@ pub const interrupt_handler: microzig.cpu.InterruptHandler = .{
         }
     }.handler_fn,
 };
+
+test {
+    _ = osi;
+}

port/espressif/esp/src/hal/radio/osi.zig

@@ -44,12 +44,14 @@ pub fn strlen(str: ?[*:0]const u8) callconv(.c) usize {
     return std.mem.len(s);
 }
 
-pub fn strnlen(str: ?[*:0]const u8, _: usize) callconv(.c) usize {
-    // const s = str orelse return 0;
-    // return if (std.mem.indexOfScalar(u8, s[0..n], 0)) |index| index + 1 else n;
+pub fn strnlen(str: ?[*:0]const u8, n: usize) callconv(.c) usize {
     const s = str orelse return 0;
+    return if (std.mem.indexOfScalar(u8, s[0..n], 0)) |index| index else n;
+}
 
-    return std.mem.len(s);
+test "strnlen" {
+    try std.testing.expect(strnlen(&.{ 10, 20, 30, 0 }, 4) == 3);
+    try std.testing.expect(strnlen(&.{ 10, 20, 30, 0, 10, 20, 30 }, 7) == 3);
 }
 
 pub fn strrchr(str: ?[*:0]const u8, chr: u32) callconv(.c) ?[*:0]const u8 {
@@ -319,11 +321,12 @@ pub fn semphr_take(ptr: ?*anyopaque, tick: u32) callconv(.c) i32 {
     log.debug("semphr_take {?} {}", .{ ptr, tick });
 
     const sem: *rtos.Semaphore = @ptrCast(@alignCast(ptr));
-    const maybe_timeout: ?rtos.Duration = if (tick == c.OSI_FUNCS_TIME_BLOCKING)
-        .from_ticks(tick)
-    else
-        null;
-    sem.take_with_timeout(maybe_timeout) catch {
+    const timeout: rtos.Timeout = switch (tick) {
+        0 => .non_blocking,
+        c.OSI_FUNCS_TIME_BLOCKING => .never,
+        else => |ticks| .{ .after = .from_ticks(ticks) },
+    };
+    sem.take_with_timeout(timeout) catch {
         log.debug(">>>> return from semaphore take with timeout: {*}", .{sem});
         return 1;
     };
@@ -368,10 +371,11 @@ const RecursiveMutex = struct {
             if (@intFromEnum(current_task.priority) > @intFromEnum(owning_task.priority)) {
                 mutex.prev_priority = owning_task.priority;
                 owning_task.priority = current_task.priority;
-                rtos.make_ready(owning_task);
+                var _hptw = false;
+                rtos.make_ready(owning_task, &_hptw);
             }
 
-            mutex.wait_queue.wait(current_task, null);
+            mutex.wait_queue.wait(null);
         }
 
         assert(mutex.value == 0);
@@ -395,8 +399,13 @@ const RecursiveMutex = struct {
                 owning_task.priority = prev_priority;
                 mutex.prev_priority = null;
             }
+
             mutex.owning_task = null;
-            mutex.wait_queue.wake_one();
+
+            var hptw = false;
+            mutex.wait_queue.wake_one(&hptw);
+            if (hptw) rtos.yield_from_cs(.reschedule);
+
             return true;
         } else {
             return false;
@@ -505,17 +514,12 @@ pub fn queue_send(ptr: ?*anyopaque, item_ptr: ?*anyopaque, block_time_tick: u32)
     const queue: *QueueWrapper = @ptrCast(@alignCast(ptr));
     const item: [*]const u8 = @ptrCast(@alignCast(item_ptr));
 
-    const size = switch (block_time_tick) {
-        0 => queue.inner.put_non_blocking(item[0..queue.item_len]),
-        else => queue.inner.put(
-            item[0..queue.item_len],
-            1,
-            if (block_time_tick != c.OSI_FUNCS_TIME_BLOCKING)
-                .from_ticks(block_time_tick)
-            else
-                null,
-        ),
+    const timeout: rtos.Timeout = switch (block_time_tick) {
+        0 => .non_blocking,
+        c.OSI_FUNCS_TIME_BLOCKING => .never,
+        else => |ticks| .{ .after = .from_ticks(ticks) },
     };
+    const size = queue.inner.put(item[0..queue.item_len], 1, timeout);
     if (size == 0) return -1;
     return 1;
 }
@@ -525,9 +529,11 @@ pub fn queue_send_from_isr(ptr: ?*anyopaque, item_ptr: ?*anyopaque, _hptw: ?*any
 
     const queue: *QueueWrapper = @ptrCast(@alignCast(ptr));
     const item: [*]const u8 = @ptrCast(@alignCast(item_ptr));
-    const n = @divExact(queue.inner.put_non_blocking(item[0..queue.item_len]), queue.item_len);
 
-    @as(*u32, @ptrCast(@alignCast(_hptw))).* = @intFromBool(rtos.is_a_higher_priority_task_ready());
+    var hptw = false;
+    const n = @divExact(queue.inner.put_from_isr(item[0..queue.item_len], &hptw), queue.item_len);
+
+    @as(*u32, @ptrCast(@alignCast(_hptw))).* |= @intFromBool(hptw);
 
     return @intCast(n);
 }
@@ -546,17 +552,12 @@ pub fn queue_recv(ptr: ?*anyopaque, item_ptr: ?*anyopaque, block_time_tick: u32)
     const queue: *QueueWrapper = @ptrCast(@alignCast(ptr));
     const item: [*]u8 = @ptrCast(@alignCast(item_ptr));
 
-    const size = switch (block_time_tick) {
-        0 => queue.inner.get_non_blocking(item[0..queue.item_len]),
-        else => queue.inner.get(
-            item[0..queue.item_len],
-            queue.item_len,
-            if (block_time_tick != c.OSI_FUNCS_TIME_BLOCKING)
-                .from_ticks(block_time_tick)
-            else
-                null,
-        ),
+    const timeout: rtos.Timeout = switch (block_time_tick) {
+        0 => .non_blocking,
+        c.OSI_FUNCS_TIME_BLOCKING => .never,
+        else => |ticks| .{ .after = .from_ticks(ticks) },
     };
+    const size = queue.inner.get(item[0..queue.item_len], 1, timeout);
     if (size == 0) return -1;
     return 1;
 }

port/espressif/esp/src/hal/radio/timer.zig

@@ -150,16 +150,16 @@ fn task_fn() void {
             callback(arg);
         }
 
-        const sleep_duration: ?rtos.Duration = blk: {
+        const timeout: rtos.Timeout = blk: {
             mutex.lock();
             defer mutex.unlock();
             break :blk if (find_next_wake_absolute()) |next_wake_absolute|
-                .from_us(@truncate(next_wake_absolute.diff(now).to_us()))
+                .{ .after = .from_us(@truncate(next_wake_absolute.diff(now).to_us())) }
             else
-                null;
+                .never;
         };
 
-        reload_semaphore.take_with_timeout(sleep_duration) catch {};
+        reload_semaphore.take_with_timeout(timeout) catch {};
     }
 }