Repository: probe-rs/probe-rs
Pull request #3265 (2025-04-27 14:41:00)

RPC: Split up RTT and Semihosting topics
This PR separates the MonitorEvent topic into two, adding mechanisms to bundle multiple topics together for more ergonimic use in the CLI. The point is to reduce breaking changes - while modifying a Topic (i.e. by adding more Semihosting events) is still breaking, adding a whole new stream (like Serial, or tracing output) will not be.


Patch:
diff --git a/changelog/added-debug-info-ref.md b/changelog/added-debug-info-ref.md
new file mode 100644
index 00000000..804f5f00
--- /dev/null
+++ b/changelog/added-debug-info-ref.md
@@ -0,0 +1,1 @@
+Debugger now handles DW_AT_type DebugInfoRef
\ No newline at end of file
diff --git a/probe-rs-debug/src/debug_info.rs b/probe-rs-debug/src/debug_info.rs
index 35b52957..d7e29b1a 100644
--- a/probe-rs-debug/src/debug_info.rs
+++ b/probe-rs-debug/src/debug_info.rs
@@ -10,8 +10,7 @@ use crate::{
     SourceLocation, VerifiedBreakpoint, registers, stack_frame::StackFrameInfo, unit_info::RangeExt,
 };
 use gimli::{
-    BaseAddresses, DebugFrame, DebugInfoOffset, RunTimeEndian, UnwindContext, UnwindSection,
-    UnwindTableRow,
+    BaseAddresses, DebugFrame, RunTimeEndian, UnwindContext, UnwindSection, UnwindTableRow,
 };
 use object::read::{Object, ObjectSection};
 use probe_rs::{Error, MemoryInterface, RegisterDataType, RegisterRole, RegisterValue, UnwindRule};
@@ -253,30 +252,20 @@ impl DebugInfo {
         }
 
         match parent_variable.variable_node_type {
-            VariableNodeType::TypeOffset(header_offset, type_offset) => {
-                let unit_header = self.dwarf.debug_info.header_from_offset(header_offset)?;
-                let unit_info = UnitInfo::new(gimli::Unit::new(&self.dwarf, unit_header)?);
-
-                // Find the parent node
-                let mut type_tree = unit_info.unit.entries_tree(Some(type_offset))?;
-                let parent_node = type_tree.root()?;
-
-                unit_info.process_tree(
-                    self,
-                    parent_node,
-                    parent_variable,
-                    memory,
-                    cache,
-                    frame_info,
-                )?;
-            }
-            VariableNodeType::DirectLookup(header_offset, unit_offset) => {
-                let unit_header = self.dwarf.debug_info.header_from_offset(header_offset)?;
-                let unit_info = UnitInfo::new(gimli::Unit::new(&self.dwarf, unit_header)?);
+            VariableNodeType::TypeOffset(header_offset, unit_offset)
+            | VariableNodeType::DirectLookup(header_offset, unit_offset) => {
+                let Some(unit_info) = self
+                    .unit_infos
+                    .iter()
+                    .find(|unit_info| unit_info.unit.header.offset() == header_offset.into())
+                else {
+                    return Err(DebugError::Other(
+                        "Failed to find unit info for offset lookup.".to_string(),
+                    ));
+                };
 
                 // Find the parent node
                 let mut type_tree = unit_info.unit.entries_tree(Some(unit_offset))?;
-
                 let parent_node = type_tree.root()?;
 
                 unit_info.process_tree(
@@ -289,43 +278,26 @@ impl DebugInfo {
                 )?;
             }
             VariableNodeType::UnitsLookup => {
-                // Look up static variables from all units
-                let mut unit_infos = self.unit_infos.iter();
-
-                let Some(unit_info) = unit_infos.next() else {
+                if self.unit_infos.is_empty() {
                     // No unit infos
                     return Err(DebugError::Other("Missing unit infos".to_string()));
-                };
-
-                let mut entries = unit_info.unit.entries();
-
-                // Only process statics for this unit header.
-                // Navigate the current unit from the header down.
-                let (_, unit_node) = entries.next_dfs()?.unwrap();
-
-                let mut tree = unit_info.unit.entries_tree(Some(unit_node.offset()))?;
-
-                unit_info.process_tree(
-                    self,
-                    tree.root()?,
-                    parent_variable,
-                    memory,
-                    cache,
-                    frame_info,
-                )?;
+                }
 
-                for unit in unit_infos {
-                    let mut entries = unit.unit.entries();
+                // Look up static variables from all units
+                for unit_info in self.unit_infos.iter() {
+                    let mut entries = unit_info.unit.entries();
 
                     // Only process statics for this unit header.
                     // Navigate the current unit from the header down.
                     let (_, unit_node) = entries.next_dfs()?.unwrap();
+                    let unit_offset = unit_node.offset();
 
-                    let mut tree = unit.unit.entries_tree(Some(unit_node.offset()))?;
+                    let mut type_tree = unit_info.unit.entries_tree(Some(unit_offset))?;
+                    let parent_node = type_tree.root()?;
 
-                    unit.process_tree(
+                    unit_info.process_tree(
                         self,
-                        tree.root()?,
+                        parent_node,
                         parent_variable,
                         memory,
                         cache,
@@ -917,25 +889,6 @@ impl DebugInfo {
         )))
     }
 
-    /// Get the DIE at the given offset into the debug info section.
-    pub(crate) fn get_die_at_offset(&self, offset: DebugInfoOffset) -> Result<Die, DebugError> {
-        for unit_info in &self.unit_infos {
-            if let Some(unit_offset) = offset.to_unit_offset(&unit_info.unit.header) {
-                return unit_info.unit.entry(unit_offset).map_err(|error| {
-                    DebugError::Other(format!(
-                        "Error reading DIE at debug info offset {:#x} : {}",
-                        offset.0, error
-                    ))
-                });
-            }
-        }
-
-        Err(DebugError::Other(format!(
-            "DIE at debug info offset {:#010x} not found",
-            offset.0
-        )))
-    }
-
     /// Look up the DIE reference for the given attribute, if it exists.
     pub(crate) fn resolve_die_reference<'debug_info, 'unit_info>(
         &'debug_info self,
@@ -946,27 +899,56 @@ impl DebugInfo {
     where
         'unit_info: 'debug_info,
     {
-        let value = die.attr_value(attribute).ok().flatten()?;
+        let attr = die.attr(attribute).ok().flatten()?;
 
-        match value {
-            gimli::AttributeValue::UnitRef(unit_ref) => unit_info.unit.entry(unit_ref).ok(),
-            gimli::AttributeValue::DebugInfoRef(debug_info_ref) => {
-                self.get_die_at_offset(debug_info_ref).ok()
-            }
-            other_value => {
-                tracing::warn!(
-                    "Unsupported {:?} value: {other_value:?}",
-                    attribute.static_string(),
-                );
-                None
-            }
-        }
+        self.resolve_die_reference_with_unit(&attr, unit_info)
+            .ok()
+            .map(|(_, die)| die)
     }
 
     /// The program binary's (and core's) endianness.
     pub fn endianness(&self) -> RunTimeEndian {
         self.endianness
     }
+
+    /// Returns the UnitInfo and DIE for the given attribute.
+    pub(crate) fn resolve_die_reference_with_unit<'debug_info, 'unit_info>(
+        &'debug_info self,
+        attr: &gimli::Attribute<GimliReader>,
+        unit_info: &'unit_info UnitInfo,
+    ) -> Result<(&'debug_info UnitInfo, Die<'debug_info, 'debug_info>), DebugError>
+    where
+        'unit_info: 'debug_info,
+    {
+        match attr.value() {
+            gimli::AttributeValue::UnitRef(unit_ref) => {
+                Ok((unit_info, unit_info.unit.entry(unit_ref)?))
+            }
+            gimli::AttributeValue::DebugInfoRef(offset) => {
+                for unit_info in &self.unit_infos {
+                    let Some(unit_offset) = offset.to_unit_offset(&unit_info.unit.header) else {
+                        continue;
+                    };
+
+                    let entry = unit_info.unit.entry(unit_offset).map_err(|error| {
+                        DebugError::Other(format!(
+                            "Error reading DIE at debug info offset {:#x} : {}",
+                            offset.0, error
+                        ))
+                    })?;
+                    return Ok((unit_info, entry));
+                }
+
+                Err(DebugError::Other(format!(
+                    "Unable to find unit info for debug info offset {:#x}",
+                    offset.0
+                )))
+            }
+            other_attribute_value => Err(DebugError::Other(format!(
+                "Unimplemented attribute value {other_attribute_value:?}"
+            ))),
+        }
+    }
 }
 
 /// Uses the [`TypedPathBuf::normalize`] function to normalize both paths before comparing them
diff --git a/probe-rs-debug/src/unit_info.rs b/probe-rs-debug/src/unit_info.rs
index 31c60287..17434319 100644
--- a/probe-rs-debug/src/unit_info.rs
+++ b/probe-rs-debug/src/unit_info.rs
@@ -429,23 +429,21 @@ impl UnitInfo {
         frame_info: StackFrameInfo<'_>,
         cache: &mut VariableCache,
     ) -> Result<(), DebugError> {
-        match attr.value() {
-            gimli::AttributeValue::UnitRef(unit_ref) => {
-                // Reference to a type, or an entry to another type or a type modifier which will point to another type.
-                // Before we resolve that type tree, we need to resolve the current node's memory location.
-                // This is because the memory location of the type nodes and child variables often inherit this value.
-                self.process_memory_location(
-                    debug_info,
-                    attributes_entry,
-                    parent_variable,
-                    child_variable,
-                    memory,
-                    frame_info,
-                )?;
+        // Reference to a type, or an entry to another type or a type modifier which will point to another type.
+        // Before we resolve that type tree, we need to resolve the current node's memory location.
+        // This is because the memory location of the type nodes and child variables often inherit this value.
+        self.process_memory_location(
+            debug_info,
+            attributes_entry,
+            parent_variable,
+            child_variable,
+            memory,
+            frame_info,
+        )?;
 
-                // Now resolve the referenced tree node for the type.
-                let referenced_type_tree_node = self.unit.entry(unit_ref)?;
-                self.extract_type(
+        match debug_info.resolve_die_reference_with_unit(attr, self) {
+            Ok((unit_info, referenced_type_tree_node)) => {
+                unit_info.extract_type(
                     debug_info,
                     &referenced_type_tree_node,
                     parent_variable,
@@ -455,9 +453,9 @@ impl UnitInfo {
                     frame_info,
                 )?;
             }
-            other_attribute_value => {
+            Err(error) => {
                 child_variable.set_value(VariableValue::Error(format!(
-                    "Unimplemented: Attribute Value for DW_AT_type {other_attribute_value:?}"
+                    "Failed to process DW_AT_type: {error:?}"
                 )));
             }
         }
diff --git a/probe-rs-debug/src/variable.rs b/probe-rs-debug/src/variable.rs
index 2b9b0d0d..87bc9bcd 100644
--- a/probe-rs-debug/src/variable.rs
+++ b/probe-rs-debug/src/variable.rs
@@ -635,8 +635,7 @@ impl Variable {
             if self.variable_node_type.is_deferred() {
                 // When we will do a lazy-load of variable children, and they have not yet been
                 // requested by the user, just display the type_name as the value
-                self.type_name
-                    .display_name(language::from_dwarf(self.language).as_ref())
+                self.type_name()
             } else if let VariableLocation::Error(ref error) = self.memory_location {
                 error.clone()
             } else {
@@ -686,12 +685,8 @@ impl Variable {
         {
             // And we have not previously assigned the value, then assign the type and address as
             // the value.
-            self.value = VariableValue::Valid(format!(
-                "{} @ {}",
-                self.type_name
-                    .display_name(language::from_dwarf(self.language).as_ref()),
-                self.memory_location
-            ));
+            self.value =
+                VariableValue::Valid(format!("{} @ {}", self.type_name(), self.memory_location));
             return;
         }
 
diff --git a/probe-rs-tools/src/bin/probe-rs/rpc/client.rs b/probe-rs-tools/src/bin/probe-rs/rpc/client.rs
index 7296938a..054e094d 100644
--- a/probe-rs-tools/src/bin/probe-rs/rpc/client.rs
+++ b/probe-rs-tools/src/bin/probe-rs/rpc/client.rs
@@ -9,7 +9,7 @@
 use postcard_rpc::{
     Topic,
     header::{VarSeq, VarSeqKind},
-    host_client::{HostClient, HostClientConfig, HostErr, IoClosed, Subscription},
+    host_client::{HostClient, HostClientConfig, HostErr, IoClosed, SubscribeError, Subscription},
 };
 use postcard_schema::Schema;
 use probe_rs::{Session, config::Registry, flashing::FlashLoader};
@@ -31,11 +31,11 @@ use crate::{
             AttachEndpoint, BuildEndpoint, ChipInfoEndpoint, CreateRttClientEndpoint,
             CreateTempFileEndpoint, EraseEndpoint, FlashEndpoint, ListChipFamiliesEndpoint,
             ListProbesEndpoint, ListTestsEndpoint, LoadChipFamilyEndpoint, MonitorEndpoint,
-            MonitorTopic, ProgressEventTopic, ReadMemory8Endpoint, ReadMemory16Endpoint,
-            ReadMemory32Endpoint, ReadMemory64Endpoint, ResetCoreEndpoint, ResumeAllCoresEndpoint,
-            RpcResult, RunTestEndpoint, SelectProbeEndpoint, TakeStackTraceEndpoint,
-            TargetInfoDataTopic, TargetInfoEndpoint, TempFileDataEndpoint, TokioSpawner,
-            VerifyEndpoint, WriteMemory8Endpoint, WriteMemory16Endpoint, WriteMemory32Endpoint,
+            ProgressEventTopic, ReadMemory8Endpoint, ReadMemory16Endpoint, ReadMemory32Endpoint,
+            ReadMemory64Endpoint, ResetCoreEndpoint, ResumeAllCoresEndpoint, RpcResult,
+            RunTestEndpoint, SelectProbeEndpoint, TakeStackTraceEndpoint, TargetInfoDataTopic,
+            TargetInfoEndpoint, TempFileDataEndpoint, TokioSpawner, VerifyEndpoint,
+            WriteMemory8Endpoint, WriteMemory16Endpoint, WriteMemory32Endpoint,
             WriteMemory64Endpoint,
             chip::{ChipData, ChipFamily, ChipInfoRequest, LoadChipFamilyRequest},
             file::{AppendFileRequest, TempFile},
@@ -45,7 +45,7 @@ use crate::{
             },
             info::{InfoEvent, TargetInfoRequest},
             memory::{ReadMemoryRequest, WriteMemoryRequest},
-            monitor::{MonitorEvent, MonitorMode, MonitorOptions, MonitorRequest},
+            monitor::{MonitorMode, MonitorOptions, MonitorRequest},
             probe::{
                 AttachRequest, AttachResult, DebugProbeEntry, DebugProbeSelector,
                 ListProbesRequest, SelectProbeRequest, SelectProbeResult,
@@ -58,7 +58,10 @@ use crate::{
         },
         transport::memory::{PostcardReceiver, PostcardSender, WireRx, WireTx},
     },
-    util::rtt::{RttChannelConfig, client::RttClient},
+    util::{
+        cli::MonitorEvent,
+        rtt::{RttChannelConfig, client::RttClient},
+    },
 };
 
 #[cfg(feature = "remote")]
@@ -298,17 +301,16 @@ impl RpcClient {
         E: postcard_rpc::Endpoint<Response = RpcResult<R>>,
         E::Request: Serialize + Schema,
         E::Response: DeserializeOwned + Schema,
-        T: Topic,
-        T::Message: DeserializeOwned,
+        T: MultiTopic,
     {
-        let mut stream = match self.client.subscribe_exclusive::<T>(64).await {
+        let mut stream = match T::subscribe(&self.client, 64).await {
             Ok(stream) => stream,
-            Err(err) => anyhow::bail!("Failed to subscribe to '{}': {err:?}", T::PATH),
+            Err(err) => anyhow::bail!("Failed to subscribe to '{}': {:?}", err.topic, err.error),
         };
 
         tokio::select! {
             biased;
-            _ = read_stream(&mut stream, on_msg) => anyhow::bail!("Topic reader returned unexpectedly"),
+            _ = stream.stream(on_msg) => anyhow::bail!("Topic reader returned unexpectedly"),
             r = self.send_resp::<E, R>(req) => r,
         }
     }
@@ -501,7 +503,7 @@ impl SessionInterface {
         on_msg: impl AsyncFnMut(MonitorEvent),
     ) -> anyhow::Result<()> {
         self.client
-            .send_and_read_stream::<MonitorEndpoint, MonitorTopic, _>(
+            .send_and_read_stream::<MonitorEndpoint, MonitorEvent, _>(
                 &MonitorRequest {
                     sessid: self.sessid,
                     mode,
@@ -519,7 +521,7 @@ impl SessionInterface {
         on_msg: impl AsyncFnMut(MonitorEvent),
     ) -> anyhow::Result<Tests> {
         self.client
-            .send_and_read_stream::<ListTestsEndpoint, MonitorTopic, _>(
+            .send_and_read_stream::<ListTestsEndpoint, MonitorEvent, _>(
                 &ListTestsRequest {
                     sessid: self.sessid,
                     boot_info,
@@ -537,7 +539,7 @@ impl SessionInterface {
         on_msg: impl AsyncFnMut(MonitorEvent),
     ) -> anyhow::Result<TestResult> {
         self.client
-            .send_and_read_stream::<RunTestEndpoint, MonitorTopic, _>(
+            .send_and_read_stream::<RunTestEndpoint, MonitorEvent, _>(
                 &RunTestRequest {
                     sessid: self.sessid,
                     test,
@@ -690,14 +692,70 @@ impl CoreInterface {
     }
 }
 
-async fn read_stream<T>(stream: &mut Subscription<T>, mut on_msg: impl AsyncFnMut(T))
+#[derive(Debug)]
+pub(crate) struct MultiSubscribeError {
+    topic: &'static str,
+    error: SubscribeError,
+}
+
+pub(crate) trait MultiTopic {
+    type Message;
+    type Subscription: MultiSubscription<Message = Self::Message>;
+
+    async fn subscribe<E>(
+        client: &HostClient<E>,
+        depth: usize,
+    ) -> Result<Self::Subscription, MultiSubscribeError>
+    where
+        E: DeserializeOwned + Schema;
+}
+
+impl<T> MultiTopic for T
 where
-    T: DeserializeOwned,
+    T: Topic,
+    T::Message: DeserializeOwned,
 {
-    while let Some(message) = stream.recv().await {
-        on_msg(message).await;
+    type Message = T::Message;
+    type Subscription = Subscription<T::Message>;
+
+    async fn subscribe<E>(
+        client: &HostClient<E>,
+        depth: usize,
+    ) -> Result<Self::Subscription, MultiSubscribeError>
+    where
+        E: DeserializeOwned + Schema,
+    {
+        match client.subscribe_exclusive::<Self>(depth).await {
+            Ok(subscription) => Ok(subscription),
+            Err(error) => Err(MultiSubscribeError {
+                topic: T::PATH,
+                error,
+            }),
+        }
     }
+}
+
+pub(crate) trait MultiSubscription {
+    type Message;
+
+    async fn next(&mut self) -> Option<Self::Message>;
+    async fn stream(&mut self, mut on_msg: impl AsyncFnMut(Self::Message)) -> anyhow::Result<()> {
+        while let Some(message) = self.next().await {
+            on_msg(message).await;
+        }
+
+        tracing::warn!("Failed to read topic");
+        futures_util::future::pending().await
+    }
+}
+
+impl<T> MultiSubscription for Subscription<T>
+where
+    T: DeserializeOwned,
+{
+    type Message = T;
 
-    tracing::warn!("Failed to read topic");
-    futures_util::future::pending().await
+    async fn next(&mut self) -> Option<Self::Message> {
+        self.recv().await
+    }
 }
diff --git a/probe-rs-tools/src/bin/probe-rs/rpc/functions.rs b/probe-rs-tools/src/bin/probe-rs/rpc/functions.rs
index 65a57987..2a911ffd 100644
--- a/probe-rs-tools/src/bin/probe-rs/rpc/functions.rs
+++ b/probe-rs-tools/src/bin/probe-rs/rpc/functions.rs
@@ -20,7 +20,7 @@ use crate::{
             },
             info::{InfoEvent, TargetInfoRequest, target_info},
             memory::{ReadMemoryRequest, WriteMemoryRequest, read_memory, write_memory},
-            monitor::{MonitorEvent, MonitorRequest, monitor},
+            monitor::{MonitorRequest, RttEvent, SemihostingEvent, monitor},
             probe::{
                 AttachRequest, AttachResponse, ListProbesRequest, ListProbesResponse, attach,
                 list_probes,
@@ -150,6 +150,65 @@ pub struct RpcSpawnContext {
     sender: PostcardSender<WireTxImpl>,
 }
 
+pub(crate) trait MultiTopicWriter {
+    type Sender: Send + 'static;
+    type Publisher: MultiTopicPublisher;
+
+    fn create(token: CancellationToken) -> (Self::Sender, Self::Publisher);
+}
+
+impl<T> MultiTopicWriter for T
+where
+    T: Topic,
+    T::Message: Serialize + Sized + Send + 'static,
+{
+    type Sender = Sender<T::Message>;
+    type Publisher = TopicPublisher<T>;
+
+    fn create(token: CancellationToken) -> (Self::Sender, Self::Publisher) {
+        let (tx, rx) = channel::<T::Message>(256);
+        (tx, TopicPublisher { rx, token })
+    }
+}
+
+pub(crate) trait MultiTopicPublisher {
+    async fn publish(self, sender: &PostcardSender<WireTxImpl>);
+}
+
+pub(crate) struct TopicPublisher<T>
+where
+    T: Topic,
+    T::Message: Serialize + Sized + Send + 'static,
+{
+    rx: Receiver<T::Message>,
+    token: CancellationToken,
+}
+
+impl<T> MultiTopicPublisher for TopicPublisher<T>
+where
+    T: Topic,
+    T::Message: Serialize + Sized + Send + 'static,
+{
+    async fn publish(mut self, sender: &PostcardSender<WireTxImpl>) {
+        loop {
+            tokio::select! {
+                biased;
+
+                _ = self.token.cancelled() => break,
+                Some(event) = self.rx.recv() => {
+                    sender
+                        .publish::<T>(VarSeq::Seq2(0), &event)
+                        .await
+                        .unwrap();
+                }
+            }
+        }
+        std::mem::drop(self.rx);
+
+        futures_util::future::pending().await
+    }
+}
+
 impl RpcSpawnContext {
     fn dry_run(&self, sessid: Key<Session>) -> bool {
         self.state.dry_run(sessid)
@@ -172,41 +231,20 @@ impl RpcSpawnContext {
 
     pub async fn run_blocking<T, F, REQ, RESP>(&mut self, request: REQ, task: F) -> RESP
     where
-        T: Topic,
-        T::Message: Serialize + Schema + Sized + Send + 'static,
-        F: FnOnce(RpcSpawnContext, REQ, Sender<T::Message>) -> RESP,
+        T: MultiTopicWriter,
+        F: FnOnce(RpcSpawnContext, REQ, T::Sender) -> RESP,
         F: Send + 'static,
         REQ: Send + 'static,
         RESP: Send + 'static,
     {
-        let (channel_sender, mut channel_receiver) = tokio::sync::mpsc::channel::<T::Message>(256);
-
-        let sender = self.sender.clone();
         let token = self.cancellation_token();
-        let sender = async move {
-            loop {
-                tokio::select! {
-                    biased;
-
-                    _ = token.cancelled() => break,
-                    Some(event) = channel_receiver.recv() => {
-                        sender
-                            .publish::<T>(VarSeq::Seq2(0), &event)
-                            .await
-                            .unwrap();
-                    }
-                }
-            }
-            std::mem::drop(channel_receiver);
-
-            futures_util::future::pending().await
-        };
+        let (sender, publisher) = T::create(token);
 
         let ctx = self.clone();
-        let blocking = tokio::task::spawn_blocking(move || task(ctx, request, channel_sender));
+        let blocking = tokio::task::spawn_blocking(move || task(ctx, request, sender));
 
         tokio::select! {
-            _ = sender => unreachable!(),
+            _ =  publisher.publish(&self.sender) => unreachable!(),
             response = blocking => {
                 response.unwrap()
             }
@@ -466,11 +504,12 @@ topics! {
 topics! {
     list = TOPICS_OUT_LIST;
     direction = TopicDirection::ToClient;
-    | TopicTy             | MessageTy     | Path              | Cfg |
-    | -------             | ---------     | ----              | --- |
-    | TargetInfoDataTopic | InfoEvent     | "info/data"       |     |
-    | ProgressEventTopic  | ProgressEvent | "flash/progress"  |     |
-    | MonitorTopic        | MonitorEvent  | "monitor"         |     |
+    | TopicTy             | MessageTy        | Path             | Cfg |
+    | -------             | ---------        | ----             | --- |
+    | TargetInfoDataTopic | InfoEvent        | "info/data"      |     |
+    | ProgressEventTopic  | ProgressEvent    | "flash/progress" |     |
+    | RttTopic            | RttEvent         | "rtt"            |     |
+    | SemihostingTopic    | SemihostingEvent | "semihosting"    |     |
 }
 
 postcard_rpc::define_dispatch! {
diff --git a/probe-rs-tools/src/bin/probe-rs/rpc/functions/monitor.rs b/probe-rs-tools/src/bin/probe-rs/rpc/functions/monitor.rs
index 2f5c375e..c0ffeb1d 100644
--- a/probe-rs-tools/src/bin/probe-rs/rpc/functions/monitor.rs
+++ b/probe-rs-tools/src/bin/probe-rs/rpc/functions/monitor.rs
@@ -3,7 +3,10 @@ use std::{num::NonZeroU32, time::Duration};
 use crate::{
     rpc::{
         Key,
-        functions::{MonitorEndpoint, MonitorTopic, RpcSpawnContext, WireTxImpl, flash::BootInfo},
+        functions::{
+            MonitorEndpoint, MultiTopicPublisher, MultiTopicWriter, RpcSpawnContext, RttTopic,
+            SemihostingTopic, WireTxImpl, flash::BootInfo,
+        },
         utils::run_loop::{RunLoop, RunLoopPoller},
     },
     util::rtt::client::RttClient,
@@ -16,7 +19,8 @@ use probe_rs::{
     semihosting::{CloseRequest, OpenRequest, SemihostingCommand, WriteRequest},
 };
 use serde::{Deserialize, Serialize};
-use tokio::sync::mpsc;
+use tokio::sync::mpsc::{self, error::SendError};
+use tokio_util::sync::CancellationToken;
 
 #[derive(Serialize, Deserialize, Schema)]
 pub enum MonitorMode {
@@ -41,23 +45,6 @@ impl MonitorMode {
     }
 }
 
-#[derive(Serialize, Deserialize, Schema)]
-// TODO: maybe this should be an enum of RttEvent | SemihostingEvent? Communicate semihosting stream without strings, by adding open/close events?
-pub enum MonitorEvent {
-    RttDiscovered {
-        up_channels: Vec<String>,
-        down_channels: Vec<String>,
-    },
-    RttOutput {
-        channel: u32,
-        bytes: Vec<u8>,
-    },
-    SemihostingOutput {
-        stream: String,
-        data: String,
-    },
-}
-
 #[derive(Serialize, Deserialize, Schema)]
 pub struct MonitorOptions {
     /// Enable reset vector catch if its supported on the target.
@@ -83,7 +70,7 @@ pub async fn monitor(
     sender: Sender<WireTxImpl>,
 ) {
     let resp = ctx
-        .run_blocking::<MonitorTopic, _, _, _>(request, monitor_impl)
+        .run_blocking::<MonitorSender, _, _, _>(request, monitor_impl)
         .await
         .map_err(Into::into);
 
@@ -93,15 +80,84 @@ pub async fn monitor(
         .unwrap();
 }
 
+#[derive(Serialize, Deserialize, Schema)]
+pub enum RttEvent {
+    Discovered {
+        up_channels: Vec<String>,
+        down_channels: Vec<String>,
+    },
+    Output {
+        channel: u32,
+        bytes: Vec<u8>,
+    },
+}
+
+#[derive(Serialize, Deserialize, Schema)]
+pub enum SemihostingEvent {
+    Output { stream: String, data: String },
+}
+
+pub(crate) struct MonitorSender {
+    rtt: mpsc::Sender<RttEvent>,
+    semihosting_output: mpsc::Sender<SemihostingEvent>,
+}
+impl MonitorSender {
+    pub(crate) fn send_semihosting_event(
+        &self,
+        event: SemihostingEvent,
+    ) -> Result<(), SendError<SemihostingEvent>> {
+        self.semihosting_output.blocking_send(event)
+    }
+
+    pub(crate) fn send_rtt_event(&self, event: RttEvent) -> Result<(), SendError<RttEvent>> {
+        self.rtt.blocking_send(event)
+    }
+}
+
+pub(crate) struct MonitorPublisher {
+    rtt: <RttTopic as MultiTopicWriter>::Publisher,
+    semihosting_output: <SemihostingTopic as MultiTopicWriter>::Publisher,
+}
+
+impl MultiTopicWriter for MonitorSender {
+    type Sender = Self;
+    type Publisher = MonitorPublisher;
+
+    fn create(token: CancellationToken) -> (Self::Sender, Self::Publisher) {
+        let (rtt_sender, rtt_publisher) = RttTopic::create(token.clone());
+        let (semihosting_sender, semihosting_publisher) = SemihostingTopic::create(token);
+
+        (
+            Self {
+                rtt: rtt_sender,
+                semihosting_output: semihosting_sender,
+            },
+            MonitorPublisher {
+                rtt: rtt_publisher,
+                semihosting_output: semihosting_publisher,
+            },
+        )
+    }
+}
+
+impl MultiTopicPublisher for MonitorPublisher {
+    async fn publish(self, sender: &Sender<WireTxImpl>) {
+        tokio::join!(
+            self.rtt.publish(sender),
+            self.semihosting_output.publish(sender)
+        );
+    }
+}
+
 fn monitor_impl(
     ctx: RpcSpawnContext,
     request: MonitorRequest,
-    sender: mpsc::Sender<MonitorEvent>,
+    sender: MonitorSender,
 ) -> anyhow::Result<()> {
     let mut session = ctx.session_blocking(request.sessid);
 
     let mut semihosting_sink =
-        MonitorEventHandler::new(|event| sender.blocking_send(event).unwrap());
+        MonitorEventHandler::new(|event| sender.send_semihosting_event(event).unwrap());
 
     let mut rtt_client = request
         .options
@@ -126,7 +182,11 @@ fn monitor_impl(
 
     let poller = rtt_client.as_deref_mut().map(|client| RttPoller {
         rtt_client: client,
-        sender: sender.clone(),
+        sender: |message| {
+            sender
+                .send_rtt_event(message)
+                .context("Failed to send RTT event")
+        },
     });
 
     run_loop.run_until(
@@ -141,12 +201,20 @@ fn monitor_impl(
     Ok(())
 }
 
-pub struct RttPoller<'c> {
+pub struct RttPoller<'c, S>
+where
+    S: FnMut(RttEvent) -> anyhow::Result<()>,
+    S: 'c,
+{
     pub rtt_client: &'c mut RttClient,
-    pub sender: mpsc::Sender<MonitorEvent>,
+    pub sender: S,
 }
 
-impl RunLoopPoller for RttPoller<'_> {
+impl<'c, S> RunLoopPoller for RttPoller<'c, S>
+where
+    S: FnMut(RttEvent) -> anyhow::Result<()>,
+    S: 'c,
+{
     fn start(&mut self, _core: &mut Core<'_>) -> anyhow::Result<()> {
         Ok(())
     }
@@ -166,12 +234,11 @@ impl RunLoopPoller for RttPoller<'_> {
                 .iter()
                 .map(|c| c.channel_name())
                 .collect::<Vec<_>>();
-            self.sender
-                .blocking_send(MonitorEvent::RttDiscovered {
-                    up_channels,
-                    down_channels,
-                })
-                .with_context(|| "Failed to send RTT discovery")?;
+            (self.sender)(RttEvent::Discovered {
+                up_channels,
+                down_channels,
+            })
+            .with_context(|| "Failed to send RTT discovery")?;
         }
 
         let mut next_poll = Duration::from_millis(100);
@@ -182,12 +249,11 @@ impl RunLoopPoller for RttPoller<'_> {
                 // Once we receive new data, we bump the frequency to 1kHz.
                 next_poll = Duration::from_millis(1);
 
-                self.sender
-                    .blocking_send(MonitorEvent::RttOutput {
-                        channel: channel as u32,
-                        bytes: bytes.to_vec(),
-                    })
-                    .with_context(|| "Failed to send RTT output")?;
+                (self.sender)(RttEvent::Output {
+                    channel: channel as u32,
+                    bytes: bytes.to_vec(),
+                })
+                .with_context(|| "Failed to send RTT output")?;
             }
         }
 
@@ -200,12 +266,12 @@ impl RunLoopPoller for RttPoller<'_> {
     }
 }
 
-struct MonitorEventHandler<F: FnMut(MonitorEvent)> {
+struct MonitorEventHandler<F: FnMut(SemihostingEvent)> {
     sender: F,
     semihosting_reader: SemihostingReader,
 }
 
-impl<F: FnMut(MonitorEvent)> MonitorEventHandler<F> {
+impl<F: FnMut(SemihostingEvent)> MonitorEventHandler<F> {
     pub fn new(sender: F) -> Self {
         Self {
             sender,
@@ -245,7 +311,7 @@ impl<F: FnMut(MonitorEvent)> MonitorEventHandler<F> {
             SemihostingCommand::Errno(_) => Ok(None),
             other if SemihostingReader::is_io(other) => {
                 if let Some((stream, data)) = self.semihosting_reader.handle(other, core)? {
-                    (self.sender)(MonitorEvent::SemihostingOutput { stream, data });
+                    (self.sender)(SemihostingEvent::Output { stream, data });
                 }
                 Ok(None)
             }
diff --git a/probe-rs-tools/src/bin/probe-rs/util/cli.rs b/probe-rs-tools/src/bin/probe-rs/util/cli.rs
index 040662ed..7bb3786a 100644
--- a/probe-rs-tools/src/bin/probe-rs/util/cli.rs
+++ b/probe-rs-tools/src/bin/probe-rs/util/cli.rs
@@ -5,6 +5,9 @@ use std::{future::Future, ops::DerefMut, path::Path, time::Instant};
 use anyhow::Context;
 use colored::Colorize;
 use libtest_mimic::{Failed, Trial};
+use postcard_rpc::host_client::HostClient;
+use postcard_schema::Schema;
+use serde::de::DeserializeOwned;
 use time::UtcOffset;
 use tokio::io::AsyncWriteExt;
 use tokio::{runtime::Handle, sync::mpsc::UnboundedSender};
@@ -14,11 +17,11 @@ use crate::{
     FormatOptions,
     rpc::{
         Key,
-        client::{RpcClient, SessionInterface},
+        client::{MultiSubscribeError, MultiSubscription, MultiTopic, RpcClient, SessionInterface},
         functions::{
-            CancelTopic,
+            CancelTopic, RttTopic, SemihostingTopic,
             flash::{BootInfo, DownloadOptions, FlashLayout, ProgressEvent, VerifyResult},
-            monitor::{MonitorEvent, MonitorMode, MonitorOptions},
+            monitor::{MonitorMode, MonitorOptions, RttEvent, SemihostingEvent},
             probe::{
                 AttachRequest, AttachResult, DebugProbeEntry, DebugProbeSelector, SelectProbeResult,
             },
@@ -383,6 +386,45 @@ pub async fn flash(
     Ok(loader.boot_info)
 }
 
+pub enum MonitorEvent {
+    Rtt(RttEvent),
+    Semihosting(SemihostingEvent),
+}
+
+impl MultiTopic for MonitorEvent {
+    type Message = Self;
+    type Subscription = MonitorSubscription;
+
+    async fn subscribe<E>(
+        client: &HostClient<E>,
+        depth: usize,
+    ) -> Result<Self::Subscription, MultiSubscribeError>
+    where
+        E: DeserializeOwned + Schema,
+    {
+        // TODO: remove MonitorEvent from the RPC interface, split this subscribe into two:
+        // one for RTT, one for semihosting, then introduce a MultiSubscription impl for them
+        let rtt = RttTopic::subscribe(client, depth).await?;
+        let semihosting = SemihostingTopic::subscribe(client, depth).await?;
+        Ok(MonitorSubscription { rtt, semihosting })
+    }
+}
+
+pub struct MonitorSubscription {
+    rtt: <RttTopic as MultiTopic>::Subscription,
+    semihosting: <SemihostingTopic as MultiTopic>::Subscription,
+}
+impl MultiSubscription for MonitorSubscription {
+    type Message = MonitorEvent;
+
+    async fn next(&mut self) -> Option<Self::Message> {
+        tokio::select! {
+            message = self.rtt.recv() => message.map(MonitorEvent::Rtt),
+            message = self.semihosting.recv() => message.map(MonitorEvent::Semihosting),
+        }
+    }
+}
+
 pub async fn monitor(
     session: &SessionInterface,
     mode: MonitorMode,
@@ -625,14 +667,14 @@ async fn print_monitor_event(
     target_output_files: &mut TargetOutputFiles,
 ) {
     match event {
-        MonitorEvent::RttDiscovered { up_channels, .. } => {
+        MonitorEvent::Rtt(RttEvent::Discovered { up_channels, .. }) => {
             let Some(client) = rtt_client else {
                 return;
             };
 
             client.on_channels_discovered(&up_channels);
         }
-        MonitorEvent::RttOutput { channel, bytes } => {
+        MonitorEvent::Rtt(RttEvent::Output { channel, bytes }) => {
             let Some(client) = rtt_client else {
                 return;
             };
@@ -653,7 +695,7 @@ async fn print_monitor_event(
                 )
                 .await;
         }
-        MonitorEvent::SemihostingOutput { stream, data } => {
+        MonitorEvent::Semihosting(SemihostingEvent::Output { stream, data }) => {
             match stream.as_str() {
                 "stdout" => print!("{data}"),
                 "stderr" => eprint!("{data}"),


Test patch:
diff --git a/probe-rs-debug/src/snapshots/probe_rs_debug__debug_info__test__esp32s3_esp_hal_panic.snap b/probe-rs-debug/src/snapshots/probe_rs_debug__debug_info__test__esp32s3_esp_hal_panic.snap
index b2c0a959..02cd5c0d 100644
--- a/probe-rs-debug/src/snapshots/probe_rs_debug__debug_info__test__esp32s3_esp_hal_panic.snap
+++ b/probe-rs-debug/src/snapshots/probe_rs_debug__debug_info__test__esp32s3_esp_hal_panic.snap
@@ -5975,12 +5975,56 @@ expression: stack_frames
                   children:
                     - name:
                         Named: inner
-                      type_name: Unknown
-                      value: "< Unimplemented: Attribute Value for DW_AT_type DebugInfoRef(DebugInfoOffset(5879)) >"
-                - name:
-                    Named: not_send
-                  type_name: Unknown
-                  value: "< Unimplemented: Attribute Value for DW_AT_type DebugInfoRef(DebugInfoOffset(6875)) >"
+                      type_name:
+                        Struct: Executor
+                      value: Executor @ 0x00000000
+                      children:
+                        - name:
+                            Named: inner
+                          type_name:
+                            Struct: SyncExecutor
+                          value: SyncExecutor @ 0x00000000
+                          children:
+                            - name:
+                                Named: run_queue
+                              type_name:
+                                Struct: RunQueue
+                              value: RunQueue @ 0x00000000
+                              children:
+                                - name:
+                                    Named: head
+                                  type_name:
+                                    Struct: "AtomicPtr<embassy_executor::raw::TaskHeader>"
+                                  value: "AtomicPtr<embassy_executor::raw::TaskHeader> @ 0x00000000"
+                                  children:
+                                    - name:
+                                        Named: p
+                                      type_name:
+                                        Struct: "UnsafeCell<*mut embassy_executor::raw::TaskHeader>"
+                                      value: "UnsafeCell<*mut embassy_executor::raw::TaskHeader> @ 0x00000000"
+                                      children:
+                                        - name:
+                                            Named: value
+                                          type_name:
+                                            Pointer: "*mut embassy_executor::raw::TaskHeader"
+                                          value: "*mut embassy_executor::raw::TaskHeader @ 0x00000000"
+                                          children:
+                                            - name:
+                                                Named: "*value"
+                                              type_name:
+                                                Struct: TaskHeader
+                                              value: TaskHeader
+                            - name:
+                                Named: pender
+                              type_name:
+                                Struct: Pender
+                              value: Pender @ 0x00000004
+                              children:
+                                - name:
+                                    Named: __0
+                                  type_name:
+                                    Pointer: "*mut ()"
+                                  value: "*mut () @ 0x00000004"
         - name:
             Named: init
           type_name:
diff --git a/probe-rs-tools/src/bin/probe-rs/rpc/functions/test.rs b/probe-rs-tools/src/bin/probe-rs/rpc/functions/test.rs
index 4d0b6629..44fd5aa7 100644
--- a/probe-rs-tools/src/bin/probe-rs/rpc/functions/test.rs
+++ b/probe-rs-tools/src/bin/probe-rs/rpc/functions/test.rs
@@ -1,19 +1,18 @@
 use std::time::Duration;
 
+use anyhow::Context;
 use postcard_rpc::{header::VarHeader, server::Sender};
 use postcard_schema::Schema;
 use probe_rs::{BreakpointCause, Core, HaltReason, Session, semihosting::SemihostingCommand};
 use serde::{Deserialize, Serialize};
-use tokio::sync::mpsc;
 
 use crate::{
     rpc::{
         Key,
         functions::{
-            ListTestsEndpoint, MonitorTopic, RpcResult, RpcSpawnContext, RunTestEndpoint,
-            WireTxImpl,
+            ListTestsEndpoint, RpcResult, RpcSpawnContext, RunTestEndpoint, WireTxImpl,
             flash::BootInfo,
-            monitor::{MonitorEvent, RttPoller, SemihostingReader},
+            monitor::{MonitorSender, RttPoller, SemihostingEvent, SemihostingReader},
         },
         utils::run_loop::{ReturnReason, RunLoop},
     },
@@ -114,7 +113,7 @@ pub async fn list_tests(
     sender: Sender<WireTxImpl>,
 ) {
     let resp = ctx
-        .run_blocking::<MonitorTopic, _, _, _>(request, list_tests_impl)
+        .run_blocking::<MonitorSender, _, _, _>(request, list_tests_impl)
         .await
         .map_err(Into::into);
 
@@ -127,10 +126,11 @@ pub async fn list_tests(
 fn list_tests_impl(
     ctx: RpcSpawnContext,
     request: ListTestsRequest,
-    sender: mpsc::Sender<MonitorEvent>,
+    sender: MonitorSender,
 ) -> anyhow::Result<Tests> {
     let mut session = ctx.session_blocking(request.sessid);
-    let mut list_handler = ListEventHandler::new(|event| sender.blocking_send(event).unwrap());
+    let mut list_handler =
+        ListEventHandler::new(|event| sender.send_semihosting_event(event).unwrap());
 
     let mut rtt_client = request
         .rtt_client
@@ -152,7 +152,11 @@ fn list_tests_impl(
 
     let poller = rtt_client.as_deref_mut().map(|client| RttPoller {
         rtt_client: client,
-        sender: sender.clone(),
+        sender: |message| {
+            sender
+                .send_rtt_event(message)
+                .context("Failed to send RTT event")
+        },
     });
 
     match run_loop.run_until(
@@ -191,7 +195,7 @@ pub async fn run_test(
     sender: Sender<WireTxImpl>,
 ) {
     let resp = ctx
-        .run_blocking::<MonitorTopic, _, _, _>(request, run_test_impl)
+        .run_blocking::<MonitorSender, _, _, _>(request, run_test_impl)
         .await
         .map_err(Into::into);
 
@@ -204,7 +208,7 @@ pub async fn run_test(
 fn run_test_impl(
     ctx: RpcSpawnContext,
     request: RunTestRequest,
-    sender: mpsc::Sender<MonitorEvent>,
+    sender: MonitorSender,
 ) -> anyhow::Result<TestResult> {
     tracing::info!("Running test {}", request.test.name);
 
@@ -226,8 +230,9 @@ fn run_test_impl(
     }
 
     let expected_outcome = request.test.expected_outcome;
-    let mut run_handler =
-        RunEventHandler::new(request.test, |event| sender.blocking_send(event).unwrap());
+    let mut run_handler = RunEventHandler::new(request.test, |event| {
+        sender.send_semihosting_event(event).unwrap()
+    });
 
     let mut run_loop = RunLoop {
         core_id,
@@ -236,7 +241,11 @@ fn run_test_impl(
 
     let poller = rtt_client.as_deref_mut().map(|client| RttPoller {
         rtt_client: client,
-        sender: sender.clone(),
+        sender: |message| {
+            sender
+                .send_rtt_event(message)
+                .context("Failed to send RTT event")
+        },
     });
 
     match run_loop.run_until(
@@ -260,13 +269,13 @@ fn run_test_impl(
     }
 }
 
-struct ListEventHandler<F: FnMut(MonitorEvent)> {
+struct ListEventHandler<F: FnMut(SemihostingEvent)> {
     semihosting_reader: SemihostingReader,
     cmdline_requested: bool,
     sender: F,
 }
 
-impl<F: FnMut(MonitorEvent)> ListEventHandler<F> {
+impl<F: FnMut(SemihostingEvent)> ListEventHandler<F> {
     const SEMIHOSTING_USER_LIST: u32 = 0x100;
 
     fn new(sender: F) -> Self {
@@ -309,7 +318,7 @@ impl<F: FnMut(MonitorEvent)> ListEventHandler<F> {
             }
             other if SemihostingReader::is_io(other) => {
                 if let Some((stream, data)) = self.semihosting_reader.handle(other, core)? {
-                    (self.sender)(MonitorEvent::SemihostingOutput { stream, data });
+                    (self.sender)(SemihostingEvent::Output { stream, data });
                 }
                 Ok(None)
             }
@@ -337,14 +346,14 @@ fn read_test_list(
     Ok(list)
 }
 
-struct RunEventHandler<F: FnMut(MonitorEvent)> {
+struct RunEventHandler<F: FnMut(SemihostingEvent)> {
     semihosting_reader: SemihostingReader,
     cmdline_requested: bool,
     test: Test,
     sender: F,
 }
 
-impl<F: FnMut(MonitorEvent)> RunEventHandler<F> {
+impl<F: FnMut(SemihostingEvent)> RunEventHandler<F> {
     fn new(test: Test, sender: F) -> Self {
         Self {
             test,
@@ -384,7 +393,7 @@ impl<F: FnMut(MonitorEvent)> RunEventHandler<F> {
             }
             other if SemihostingReader::is_io(other) => {
                 if let Some((stream, data)) = self.semihosting_reader.handle(other, core)? {
-                    (self.sender)(MonitorEvent::SemihostingOutput { stream, data });
+                    (self.sender)(SemihostingEvent::Output { stream, data });
                 }
                 Ok(None)
             }