MCTP IPC Implementation

Cargo.toml

@@ -164,7 +164,7 @@ tlvc = { git = "https://github.com/oxidecomputer/tlvc", default-features = false
 tlvc-text = { git = "https://github.com/oxidecomputer/tlvc", default-features = false, version = "0.3.0" }
 transceiver-messages = { git = "https://github.com/oxidecomputer/transceiver-control/", default-features = false }
 vsc7448-pac = { git = "https://github.com/oxidecomputer/vsc7448", default-features = false }
-mctp = { git = "https://github.com/9elements/mctp-rs.git", branch = "embassy-feature", default-features = false }
+mctp = { git = "https://github.com/9elements/mctp-rs.git", branch = "main", default-features = false }
 
 [workspace.lints.rust]
 elided_lifetimes_in_paths = "warn"

app/ast1060-mctp-echo/app.toml

@@ -34,7 +34,8 @@ task-slots = ["mctp_server"]
 [tasks.mctp_server]
 name = "mctp-server"
 priority = 1
-max-sizes = {flash = 8192, ram = 8192}
+max-sizes = {flash = 16384, ram = 16384}
 start = true
-stacksize = 4096
+stacksize = 12288
+notifications = ["timer"]
 

idl/mctp.idol

@@ -38,10 +38,11 @@ Interface(
                 err: CLike("ServerError"),
             ),
         ),
-        "recv": (   // TODO Is one recv sufficient for listener and requests, or do we need two?
-            doc: "Receive a single message matched by the `Listener`",
+        "recv": (
+            doc: "Receive a single message, either a request matching the listener or a response to a request",
             args: {
                 "handle": "GenericHandle",
+                "timeout_millis": "u32",
             },
             leases: {
                 "buf": (type: "[u8]", write: true),
@@ -57,6 +58,7 @@ Interface(
             args: {
                 "handle": "GenericHandle",
                 "typ": "u8",
+                "eid": "Option<u8>",
                 "tag": "Option<u8>",
                 "ic": "bool",
             },

task/mctp-api/src/lib.rs

@@ -32,20 +32,30 @@ impl From<TaskId> for Stack {
 }
 
 impl Stack {
-    pub fn req(&self, eid: Eid) -> mctp::Result<MctpReqChannel<'_>> {
+    pub fn req(
+        &self,
+        eid: Eid,
+        timeout_millis: Option<u32>,
+    ) -> mctp::Result<MctpReqChannel<'_>> {
         let handle = self.ipc.req(eid.0)?;
         Ok(MctpReqChannel {
             stack: self,
             handle,
             eid,
             sent_tag: None,
+            timeout: timeout_millis.unwrap_or(0),
         })
     }
-    pub fn listener(&self, typ: MsgType) -> mctp::Result<MctpListener<'_>> {
+    pub fn listener(
+        &self,
+        typ: MsgType,
+        timeout_millis: Option<u32>,
+    ) -> mctp::Result<MctpListener<'_>> {
         let handle = self.ipc.listener(typ.0)?;
         Ok(MctpListener {
             stack: self,
             handle,
+            timeout: timeout_millis.unwrap_or(0),
         })
     }
     pub fn get_eid(&self) -> Eid {
@@ -63,6 +73,10 @@ pub struct MctpReqChannel<'r> {
     handle: ipc::GenericHandle,
     eid: Eid,
     sent_tag: Option<Tag>,
+    /// Timeout in milliseconds
+    ///
+    /// 0 means no timeout.
+    timeout: u32,
 }
 impl ReqChannel for MctpReqChannel<'_> {
     fn send_vectored(
@@ -80,7 +94,7 @@ impl ReqChannel for MctpReqChannel<'_> {
         let _ = typ;
         let _ = integrity_check;
         let _ = bufs;
-        todo!("Vectored messages are not supported jet!")
+        Err(Error::Unsupported)
     }
 
     fn recv<'f>(
@@ -96,8 +110,7 @@ impl ReqChannel for MctpReqChannel<'_> {
             msg_tag,
             remote_eid,
             size,
-            resp_handle: _,
-        } = self.stack.ipc.recv(self.handle, buf)?;
+        } = self.stack.ipc.recv(self.handle, self.timeout, buf)?;
         debug_assert_eq!(tv.0, msg_tag);
         debug_assert_eq!(self.eid.0, remote_eid);
         let ic = mctp::MsgIC(msg_ic);
@@ -112,7 +125,10 @@ impl ReqChannel for MctpReqChannel<'_> {
         if self.sent_tag.is_some() {
             return Err(Error::BadArgument);
         }
-        let tv = self.stack.ipc.send(self.handle, typ.0, None, false, buf)?;
+        let tv =
+            self.stack
+                .ipc
+                .send(self.handle, typ.0, None, None, false, buf)?;
         let tag = Tag::Owned(mctp::TagValue(tv));
         self.sent_tag = Some(tag);
         Ok(())
@@ -124,6 +140,10 @@ impl ReqChannel for MctpReqChannel<'_> {
 pub struct MctpListener<'r> {
     stack: &'r Stack,
     handle: ipc::GenericHandle,
+    /// Timeout in milliseconds
+    ///
+    /// 0 means no timeout.
+    timeout: u32,
 }
 impl Listener for MctpListener<'_> {
     type RespChannel<'a>
@@ -146,16 +166,11 @@ impl Listener for MctpListener<'_> {
             msg_tag,
             remote_eid,
             size,
-            resp_handle,
-        } = self.stack.ipc.recv(self.handle, buf)?;
-
-        let Some(resp_handle) = resp_handle else {
-            return Err(Error::InternalError);
-        };
+        } = self.stack.ipc.recv(self.handle, self.timeout, buf)?;
 
         let resp_channel = MctpRespChannel {
             stack: self.stack,
-            handle: resp_handle,
+            handle: self.handle.clone(),
             eid: Eid(remote_eid),
             typ: MsgType(msg_typ),
             tv: TagValue(msg_tag),
@@ -192,22 +207,29 @@ impl<'r> RespChannel for MctpRespChannel<'r> {
         //      This it not ideal but might be a sufficient for now.
         let _ = integrity_check;
         let _ = bufs;
-        todo!("Vectored messages are not supported jet!")
+        Err(Error::Unsupported)
     }
 
     fn remote_eid(&self) -> Eid {
         self.eid
     }
 
     fn req_channel(&self) -> mctp::Result<Self::ReqChannel> {
-        self.stack.req(self.eid)
+        self.stack.req(self.eid, None)
     }
 
     fn send(&mut self, buf: &[u8]) -> mctp::Result<()> {
         Ok(self
             .stack
             .ipc
-            .send(self.handle, self.typ.0, Some(self.tv.0), false, buf)
+            .send(
+                self.handle,
+                self.typ.0,
+                Some(self.eid.0),
+                Some(self.tv.0),
+                false,
+                buf,
+            )
             .map(|_| ())?)
     }
 }
@@ -232,7 +254,6 @@ pub mod ipc {
         pub msg_tag: u8,
         pub remote_eid: u8,
         pub size: u64,
-        pub resp_handle: Option<GenericHandle>,
     }
 
     /// Errors reported by the MCTP server
@@ -288,9 +309,11 @@ pub mod ipc {
         Serialize,
         Deserialize,
         SerializedSize,
+        PartialEq,
+        Eq,
     )]
     #[repr(transparent)]
-    pub struct GenericHandle(pub u8);
+    pub struct GenericHandle(pub u32);
 
     pub mod client {
         use super::*;

task/mctp-echo/src/main.rs

@@ -21,7 +21,7 @@ fn main() -> ! {
     let stack = mctp_api::Stack::from(MCTP.get_task_id());
 
     stack.set_eid(Eid(8)).unwrap_lite();
-    let mut listener = stack.listener(MsgType(1)).unwrap_lite();
+    let mut listener = stack.listener(MsgType(1), None).unwrap_lite();
     let mut recv_buf = [0; 255];
 
     loop {

task/mctp-server/Cargo.toml

@@ -13,9 +13,10 @@ hubpack = { workspace = true }
 zerocopy = { workspace = true }
 zerocopy-derive = { workspace = true }
 mctp-api = { path = "../mctp-api" }
-mctp-estack = { git = "https://github.com/9elements/mctp-rs.git", branch = "leongross/serial-sync" }
-mctp-stack = { path = "../../../mctp-lib", package = "mctp-baremetal" }
+mctp-stack = { git = "https://github.com/9elements/mctp-lib.git", package = "mctp-baremetal", branch = "buildup" }
 mctp = { workspace = true }
+heapless = { workspace = true }
+
 
 [build-dependencies]
 idol = { workspace = true }

task/mctp-server/src/main.rs

@@ -10,14 +10,30 @@ use userlib::*;
 mod serial;
 mod server;
 
+use server::Server;
+
+pub const TIMER_NOTIFICATION: u32 = 1;
+
+/// Maximum number of concurrent requests the server can handle.
+pub const MAX_REQUESTS: usize = 8;
+/// Maximum number of listeners that can be registered concurrently.
+pub const MAX_LISTENERS: usize = 8;
+/// Maximum number of concurrent outstanding receive calls.
+pub const MAX_OUTSTANDING: usize = 16;
+
 #[export_name = "main"]
 fn main() -> ! {
     let mut msg_buf = [0; ipc::INCOMING_SIZE];
-    let mut server =
-        server::Server::new(mctp::Eid(42), 0, serial::SerialSender {});
+    let mut server: Server<_, MAX_OUTSTANDING> =
+        Server::new(mctp::Eid(42), 0, serial::SerialSender {});
 
     loop {
-        let msg = sys_recv_open(&mut msg_buf, 0);
+        let msg = sys_recv_open(&mut msg_buf, TIMER_NOTIFICATION);
+        if msg.sender == TaskId::KERNEL && msg.operation == TIMER_NOTIFICATION {
+            let state = sys_get_timer();
+            server.update(state.now);
+            continue;
+        }
         handle_mctp_msg(&msg_buf, msg, &mut server);
     }
 }
@@ -29,10 +45,10 @@ mod ipc {
     include!(concat!(env!("OUT_DIR"), "/server_stub.rs"));
 }
 
-fn handle_mctp_msg<S: mctp_stack::Sender>(
+fn handle_mctp_msg<S: mctp_stack::Sender, const OUTSTANDING: usize>(
     msg_buf: &[u8],
     recv_msg: RecvMessage,
-    server: &mut server::Server<S>,
+    server: &mut server::Server<S, OUTSTANDING>,
 ) {
     use hubpack::deserialize;
     use idol_runtime::Leased;
@@ -68,7 +84,12 @@ fn handle_mctp_msg<S: mctp_stack::Sender>(
                 deserialize(msg_buf).unwrap_lite();
             let lease = Leased::write_only_slice(recv_msg.sender, 0, None)
                 .unwrap_lite();
-            server.recv(&recv_msg, recv_args.handle, lease);
+            server.recv(
+                recv_msg,
+                recv_args.handle,
+                recv_args.timeout_millis,
+                lease,
+            );
         }
         ipc::MCTPOperation::send => {
             let (send_args, _): (ipc::MCTP_send_ARGS, _) =
@@ -80,6 +101,7 @@ fn handle_mctp_msg<S: mctp_stack::Sender>(
                 &recv_msg,
                 send_args.handle,
                 send_args.typ,
+                send_args.eid,
                 send_args.tag,
                 ic,
                 lease,

task/mctp-server/src/serial.rs

@@ -5,7 +5,7 @@ pub struct SerialSender;
 impl mctp_stack::Sender for SerialSender {
     fn send(
         &mut self,
-        fragmenter: mctp_stack::Fragmenter,
+        fragmenter: mctp_stack::fragment::Fragmenter,
         payload: &[u8],
     ) -> Result<mctp::Tag> {
         todo!()