util/ipc: add IPC abstraction

[rusty1968]

Summary

Add util/ipc, a #![no_std] IpcChannel trait over the kernel's channel IPC syscalls, covering both synchronous and asynchronous transactions:

[chrysh]

Design review of the IPC abstraction. Five comments inline, all about the shape of the API rather than the syscall forwarding, which looks right. Checked against pw_kernel/kernel/object/channel.rs at this head.

[chrysh]

Two follow-ups on the async lifetime, both about giving the transaction an owner.

[chrysh]

Wrote the three cases from the test-coverage thread and ran them on QEMU:
drop-cancel, double start, and try_recv before the response. All pass on top
of 008045a, and no_panics_test stays green. The whole sequence runs 50
times over.

Every case is mutation-checked, so none of them passes by accident:

Mutation	Result
Drop no longer cancels	FAIL
local pending check removed from start()	FAIL
handler gate never fires	FAIL
handler stops lowering the parked USER signal	FAIL
the same, with ROUNDS set back to 1	PASS

Two things worth knowing if you write this differently. Calling try_recv()
straight after start() is a race, not a test: my first version passed even
with the handler responding immediately. The handler now raises USER on the
initiator to say it has read the request and parked, the initiator waits for
that before calling try_recv(), and both sides lower their peer's USER
afterwards.

The repeat loop is the other half. A signal left raised makes the handshake
vacuous from round two on, and the case still passed, so the gated case polls
for a stale parked signal before it starts. That pair is the last row of the
table: the leak is caught at 50 rounds and invisible at 1.

Apply with git am or take it as a starting point.

Patch

diff --git a/target/ast10x0/tests/util_ipc/async_transaction/handler_main.rs b/target/ast10x0/tests/util_ipc/async_transaction/handler_main.rs
index d1087396..a9d77831 100644
--- a/target/ast10x0/tests/util_ipc/async_transaction/handler_main.rs
+++ b/target/ast10x0/tests/util_ipc/async_transaction/handler_main.rs
@@ -4,7 +4,8 @@
 //! Handler side of the util/ipc AsyncTransaction QEMU test.
 //!
 //! Exercises `util_ipc::IpcHandler`: waits for a request, reads it, and
-//! responds with the request byte incremented by one.
+//! responds with the request byte incremented by one. A request of
+//! `GATED_REQUEST` is held until the initiator raises Signals::USER.
 
 #![no_main]
 #![no_std]
@@ -16,6 +17,12 @@ use userspace::syscall::{self, Signals};
 use userspace::time::Instant;
 use util_ipc::{IpcHandle, IpcHandler};
 
+/// Request byte that parks the handler instead of responding: it raises
+/// Signals::USER on the initiator to say it is parked, then waits for the
+/// initiator to raise USER back before responding. The parked signal is
+/// what makes the initiator's "pending" observation deterministic.
+const GATED_REQUEST: u8 = 0x40;
+
 #[entry]
 fn entry() {
     let ipc = IpcHandle::new(handle::IPC);
@@ -28,6 +35,16 @@ fn entry() {
         let mut buf = [0u8; 1];
         match ipc.read(0, &mut buf) {
             Ok(1) => {
+                if buf[0] == GATED_REQUEST {
+                    // Tell the initiator we are parked, wait for its
+                    // release, then lower the parked signal again.
+                    if ipc.set_peer_user_signal(true).is_err()
+                        || syscall::object_wait(handle::IPC, Signals::USER, Instant::MAX).is_err()
+                    {
+                        continue;
+                    }
+                    let _ = ipc.set_peer_user_signal(false);
+                }
                 buf[0] = buf[0].wrapping_add(1);
                 let _ = ipc.respond(&buf);
             }
diff --git a/target/ast10x0/tests/util_ipc/async_transaction/initiator_main.rs b/target/ast10x0/tests/util_ipc/async_transaction/initiator_main.rs
index 29bb0bdf..eebb3804 100644
--- a/target/ast10x0/tests/util_ipc/async_transaction/initiator_main.rs
+++ b/target/ast10x0/tests/util_ipc/async_transaction/initiator_main.rs
@@ -3,15 +3,20 @@
 
 //! Initiator side of the util/ipc AsyncTransaction QEMU test.
 //!
-//! Runs three cases against `handler` and calls `debug_shutdown(Ok(()))` on
-//! full pass or `debug_shutdown(Err(_))` on the first failure. The kernel
-//! target writes `TEST_RESULT:PASS/FAIL` to UART.
+//! Runs the cases below against `handler`, `ROUNDS` times over, and calls
+//! `debug_shutdown(Ok(()))` on full pass or `debug_shutdown(Err(_))` on the
+//! first failure. The kernel target writes `TEST_RESULT:PASS/FAIL` to UART.
+//! Repeating catches state left behind by an earlier pass, a signal never
+//! lowered above all.
 //!
 //! | Case               | Exercises                          | Expect            |
 //! |--------------------|-------------------------------------|-------------------|
 //! | blocking transact  | `IpcInitiator::transact`            | byte incremented  |
 //! | async cancel       | `AsyncTransaction::start`/`cancel`  | channel freed     |
 //! | async roundtrip    | `AsyncTransaction::start`/`try_recv`| byte incremented  |
+//! | drop cancels       | `AsyncTransaction::drop`            | channel freed     |
+//! | double start       | `start` while pending               | buffers returned  |
+//! | try_recv early     | `try_recv` before the response      | `Unavailable`     |
 
 #![no_main]
 #![no_std]
@@ -20,12 +25,20 @@ use app_initiator::handle;
 use pw_status::{Error, Result};
 use userspace::entry;
 use userspace::syscall::{self, Signals};
-use userspace::time::Instant;
+use userspace::time::{Clock, Instant, SystemClock};
 use util_ipc::{AsyncTransaction, IpcHandle, IpcInitiator};
 
 static SEND_BUF: [u8; 1] = [0x10];
 static mut RECV_BUF: [u8; 1] = [0u8; 1];
 
+/// Second receive buffer, for the case that starts a transaction while
+/// another one still borrows `RECV_BUF`.
+static mut RECV_BUF2: [u8; 1] = [0u8; 1];
+
+/// Request byte the handler holds until Signals::USER is raised, matching
+/// `GATED_REQUEST` in handler_main.rs.
+static SEND_GATED: [u8; 1] = [0x40];
+
 /// # Safety
 /// Only called from this single-threaded app, and only while no
 /// `AsyncTransaction` still holds a prior borrow of `RECV_BUF`.
@@ -34,6 +47,13 @@ unsafe fn recv_buf() -> &'static mut [u8] {
     unsafe { &mut *core::ptr::addr_of_mut!(RECV_BUF) }
 }
 
+/// # Safety
+/// Same contract as `recv_buf`, for the second buffer.
+unsafe fn recv_buf2() -> &'static mut [u8] {
+    // Safety: see function doc.
+    unsafe { &mut *core::ptr::addr_of_mut!(RECV_BUF2) }
+}
+
 fn test_blocking_transact() -> Result<()> {
     let ipc = IpcHandle::new(handle::IPC);
     let send = [0x20u8];
@@ -80,11 +100,143 @@ fn test_async_roundtrip() -> Result<()> {
     Ok(())
 }
 
+/// Dropping a pending transaction has to cancel it, otherwise the channel
+/// stays busy and the next start fails with `Unavailable`.
+fn test_drop_cancels() -> Result<()> {
+    {
+        let mut txn = AsyncTransaction::new(IpcHandle::new(handle::IPC));
+        // Safety: no other AsyncTransaction is live right now.
+        txn.start(&SEND_BUF, unsafe { recv_buf() })?;
+        // Dropped here with the transaction still pending.
+    }
+
+    let mut txn = AsyncTransaction::new(IpcHandle::new(handle::IPC));
+    // Safety: the drop above cancelled the transaction. The kernel clears
+    // its transaction slot on every path out of the cancel syscall, so it
+    // no longer points into RECV_BUF.
+    txn.start(&SEND_BUF, unsafe { recv_buf() })?;
+    txn.cancel()?;
+    Ok(())
+}
+
+/// A second start while one is pending fails locally and hands both
+/// buffers back in the error.
+fn test_double_start() -> Result<()> {
+    let mut txn = AsyncTransaction::new(IpcHandle::new(handle::IPC));
+    // Safety: no other AsyncTransaction is live right now.
+    txn.start(&SEND_BUF, unsafe { recv_buf() })?;
+
+    // RECV_BUF is still borrowed by the pending transaction, so the second
+    // start gets its own buffer.
+    // Safety: nothing else borrows RECV_BUF2.
+    let result = txn.start(&SEND_BUF, unsafe { recv_buf2() });
+
+    let outcome = match result {
+        Ok(()) => {
+            pw_log::error!("double start: second start() succeeded");
+            Err(Error::Internal)
+        }
+        Err(e) if e.error != Error::FailedPrecondition => {
+            pw_log::error!("double start: status code {}", e.error as u32);
+            Err(Error::Internal)
+        }
+        Err(e) => {
+            let send_returned = core::ptr::eq(e.buffers.send.as_ptr(), SEND_BUF.as_ptr());
+            let recv_returned = core::ptr::eq(
+                e.buffers.recv.as_ptr(),
+                core::ptr::addr_of!(RECV_BUF2).cast(),
+            );
+            if send_returned && recv_returned {
+                Ok(())
+            } else {
+                pw_log::error!("double start: buffers not returned");
+                Err(Error::Internal)
+            }
+        }
+    };
+
+    txn.cancel()?;
+    outcome
+}
+
+/// `try_recv` before the handler responds reports Unavailable and keeps the
+/// transaction pending. The handler parks on this request and raises USER
+/// to say so, so the observation is not a race: by the time USER arrives
+/// the handler has read the request and is not going to respond until
+/// released.
+fn test_try_recv_before_response() -> Result<()> {
+    let ipc = IpcHandle::new(handle::IPC);
+
+    // The handler lowers the parked signal before it responds, so USER is
+    // clear on entry. If it is still set, it leaked from an earlier round
+    // and the wait below would return without the handler having parked.
+    // A deadline of now makes this a poll rather than a wait.
+    if syscall::object_wait(ipc.as_raw(), Signals::USER, SystemClock::now()).is_ok() {
+        pw_log::error!("try_recv early: stale parked signal");
+        return Err(Error::Internal);
+    }
+
+    let mut txn = AsyncTransaction::new(ipc);
+    // Safety: no other AsyncTransaction is live right now.
+    txn.start(&SEND_GATED, unsafe { recv_buf() })?;
+
+    // Wait for the handler to say it is parked.
+    syscall::object_wait(txn.as_raw(), Signals::USER, Instant::MAX)?;
+
+    match txn.try_recv() {
+        Err(Error::Unavailable) => {}
+        Ok(_) => {
+            pw_log::error!("try_recv early: completed before the handler responded");
+            return Err(Error::Internal);
+        }
+        Err(e) => {
+            pw_log::error!("try_recv early: status code {}", e as u32);
+            return Err(Error::Internal);
+        }
+    }
+    if !txn.is_pending() {
+        pw_log::error!("try_recv early: transaction no longer pending");
+        return Err(Error::Internal);
+    }
+
+    // Release the handler, then complete as usual.
+    ipc.set_peer_user_signal(true)?;
+    syscall::object_wait(txn.as_raw(), Signals::READABLE, Instant::MAX)?;
+    let completion = txn.try_recv()?;
+    ipc.set_peer_user_signal(false)?;
+
+    if completion.len != 1 || completion.buffers.recv[0] != 0x41 {
+        pw_log::error!("try_recv early: unexpected response");
+        return Err(Error::Internal);
+    }
+    Ok(())
+}
+
+/// How often the whole sequence runs. Every case leaves the channel idle
+/// and both USER signals lowered, so a repeat that fails means state leaked
+/// from the pass before it.
+const ROUNDS: u32 = 50;
+
+fn run_all() -> Result<()> {
+    for round in 0..ROUNDS {
+        let ret = test_blocking_transact()
+            .and_then(|_| test_async_cancel())
+            .and_then(|_| test_async_roundtrip())
+            .and_then(|_| test_drop_cancels())
+            .and_then(|_| test_double_start())
+            .and_then(|_| test_try_recv_before_response());
+
+        if ret.is_err() {
+            pw_log::error!("failed in round {}", round as u32);
+            return ret;
+        }
+    }
+    Ok(())
+}
+
 #[entry]
 fn entry() {
-    let ret = test_blocking_transact()
-        .and_then(|_| test_async_cancel())
-        .and_then(|_| test_async_roundtrip());
+    let ret = run_all();
 
     match &ret {
         Ok(()) => pw_log::info!("All test cases PASSED"),

[rusty1968]

@chrysh , we are deferring host testability for now - doing it properly requires upstream changes. The QEMU coverage exercises this code against the real kernel.

[rusty1968]

Wrote the three cases from the test-coverage thread and ran them on QEMU: drop-cancel, double start, and try_recv before the response. All pass on top of 008045a, and no_panics_test stays green. The whole sequence runs 50 times over.

Every case is mutation-checked, so none of them passes by accident:

Mutation Result
Drop no longer cancels FAIL
local pending check removed from start() FAIL
handler gate never fires FAIL
handler stops lowering the parked USER signal FAIL
the same, with ROUNDS set back to 1 PASS
Two things worth knowing if you write this differently. Calling try_recv() straight after start() is a race, not a test: my first version passed even with the handler responding immediately. The handler now raises USER on the initiator to say it has read the request and parked, the initiator waits for that before calling try_recv(), and both sides lower their peer's USER afterwards.

The repeat loop is the other half. A signal left raised makes the handshake vacuous from round two on, and the case still passed, so the gated case polls for a stale parked signal before it starts. That pair is the last row of the table: the leak is caught at 50 rounds and invisible at 1.

Apply with git am or take it as a starting point.

Patch

Test cases adopted.

[chrysh]

All eight of my earlier threads are resolved and the code at d874e15 does what they asked for: the roles are split, AsyncTransaction owns the in-flight state and cancels on drop, cancel() hands the buffers back on the error path too, and the crate is target-only so Instant has one definition. The QEMU test now covers drop-cancel, double start and try_recv before the response.

The two comments I left today are not blockers: the send buffer signature, which we can do here or in a follow-up, and a doc nit on three trait methods.