Sync primitives timeout refactor and high prio task yield

Author: tact1m4n3

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

@@ -107,16 +107,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 {
@@ -126,7 +123,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
@@ -135,12 +132,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 {
@@ -166,7 +160,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

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

@@ -321,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;
     };
@@ -370,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);
@@ -397,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;
@@ -507,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;
 }
@@ -527,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);
 }
@@ -548,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 {};
     }
 }