Repository: probe-rs/probe-rs
Pull request #3049 (2025-01-24 22:05:58)

Rework the RTT processing code
This PR builds on top of the RPC PR and moves RTT processing from the server to the client. This mainly means the firmware doesn't need to be uploaded for monitoring operations. I'm also fixing a bug where the defmt processor couldn't deal with data arriving split into multiple reads, and I'm also changing the default log format to "oneline" because it's just so much more readable.

Closes #2981

Murmele: I tried this pr because I had some lost messages, but with this pr the defmt-rtt messages are sometimes out of order

![grafik](https://github.com/user-attachments/assets/065e2ebb-bbda-4bda-82a2-4b9472b5a54a)


```
let mut counter = 0;
loop {
    error!("Counter: {}", counter);
    counter += 1;
}
```
bugadani: Thanks for letting me know!

Patch:
diff --git a/changelog/changed-defmt-default.md b/changelog/changed-defmt-default.md
new file mode 100644
index 00000000..a7eb92c0
--- /dev/null
+++ b/changelog/changed-defmt-default.md
@@ -0,0 +1,1 @@
+The default defmt output format is now single line. The previous default can be activated by using the "twoline" log format.
\ No newline at end of file
diff --git a/changelog/fixed-defmt.md b/changelog/fixed-defmt.md
new file mode 100644
index 00000000..8a1da8ae
--- /dev/null
+++ b/changelog/fixed-defmt.md
@@ -0,0 +1,1 @@
+Fixed defmt not being able to read messages larger than the RTT buffer.
\ No newline at end of file
diff --git a/probe-rs-tools/src/bin/probe-rs/cmd/attach.rs b/probe-rs-tools/src/bin/probe-rs/cmd/attach.rs
index 94d70434..59eec65d 100644
--- a/probe-rs-tools/src/bin/probe-rs/cmd/attach.rs
+++ b/probe-rs-tools/src/bin/probe-rs/cmd/attach.rs
@@ -1,7 +1,8 @@
+use time::UtcOffset;
+
 use crate::rpc::client::RpcClient;
 use crate::rpc::functions::monitor::{MonitorMode, MonitorOptions};
-use crate::rpc::functions::rtt_client::LogOptions;
-use crate::util::cli;
+use crate::util::cli::{self, rtt_client};
 
 #[derive(clap::Parser)]
 #[group(skip)]
@@ -11,29 +12,34 @@ pub struct Cmd {
 }
 
 impl Cmd {
-    pub async fn run(self, client: RpcClient) -> anyhow::Result<()> {
+    pub async fn run(self, client: RpcClient, utc_offset: UtcOffset) -> anyhow::Result<()> {
         let session =
             cli::attach_probe(&client, self.run.shared_options.probe_options, true).await?;
 
-        let rtt = session
-            .create_rtt_client(
-                Some(self.run.shared_options.path.clone()),
-                LogOptions {
-                    no_location: self.run.shared_options.no_location,
-                    log_format: self.run.shared_options.log_format,
-                    rtt_scan_memory: self.run.shared_options.rtt_scan_memory,
-                },
-            )
-            .await?;
+        let rtt_client = rtt_client(
+            &session,
+            &self.run.shared_options.path,
+            match self.run.shared_options.rtt_scan_memory {
+                true => crate::rpc::functions::rtt_client::ScanRegion::TargetDefault,
+                false => crate::rpc::functions::rtt_client::ScanRegion::Ranges(vec![]),
+            },
+            self.run.shared_options.log_format,
+            !self.run.shared_options.no_location,
+            Some(utc_offset),
+        )
+        .await?;
+
+        let client_handle = rtt_client.handle();
 
         cli::monitor(
             &session,
             MonitorMode::AttachToRunning,
             &self.run.shared_options.path,
+            Some(rtt_client),
             MonitorOptions {
                 catch_reset: self.run.run_options.catch_reset,
                 catch_hardfault: self.run.run_options.catch_hardfault,
-                rtt_client: Some(rtt),
+                rtt_client: Some(client_handle),
             },
             self.run.shared_options.always_print_stacktrace,
         )
diff --git a/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/config/mod.rs b/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/config/mod.rs
index a2a9c36e..ed121377 100644
--- a/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/config/mod.rs
+++ b/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/config/mod.rs
@@ -4,11 +4,13 @@ use figment::{
     Figment,
 };
 use probe_rs::probe::WireProtocol;
-use probe_rs::rtt::ChannelMode;
 use serde::{Deserialize, Serialize};
 use std::{net::SocketAddr, path::PathBuf, time::Duration};
 
-use crate::util::{logging::LevelFilter, rtt::DataFormat};
+use crate::util::{
+    logging::LevelFilter,
+    rtt::{ChannelMode, DataFormat},
+};
 
 use super::rttui::tab::TabConfig;
 
@@ -74,10 +76,9 @@ pub struct General {
     pub connect_under_reset: bool,
 }
 
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-// Note: default values are defined in `RttChannelConfig`.
+#[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize)]
 pub struct UpChannelConfig {
-    pub channel: usize,
+    pub channel: u32,
     #[serde(default)]
     pub mode: Option<ChannelMode>,
     #[serde(default)]
@@ -97,7 +98,7 @@ pub struct UpChannelConfig {
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct DownChannelConfig {
-    pub channel: usize,
+    pub channel: u32,
     #[serde(default)]
     pub mode: Option<ChannelMode>,
 }
@@ -126,7 +127,7 @@ pub struct Rtt {
 
 impl Rtt {
     /// Returns the configuration for the specified up channel number, if it exists.
-    pub fn up_channel_config(&self, channel_number: usize) -> Option<&UpChannelConfig> {
+    pub fn up_channel_config(&self, channel_number: u32) -> Option<&UpChannelConfig> {
         self.up_channels
             .iter()
             .find(|ch| ch.channel == channel_number)
diff --git a/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/mod.rs b/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/mod.rs
index 3c881e67..192e029e 100644
--- a/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/mod.rs
+++ b/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/mod.rs
@@ -30,7 +30,7 @@ use crate::util::common_options::{BinaryDownloadOptions, OperationError, ProbeOp
 use crate::util::flash::{build_loader, run_flash_download};
 use crate::util::logging::setup_logging;
 use crate::util::rtt::client::RttClient;
-use crate::util::rtt::{RttChannelConfig, RttConfig};
+use crate::util::rtt::{self, RttChannelConfig, RttConfig};
 use crate::util::{cargo::build_artifact, common_options::CargoOptions, logging};
 use crate::FormatOptions;
 
@@ -250,12 +250,18 @@ fn main_try(args: &[OsString], offset: UtcOffset) -> Result<()> {
     } else {
         None
     };
-    let rtt_client = RttClient::new(
-        elf.as_deref(),
-        session.target(),
-        create_rtt_config(&config).clone(),
-        ScanRegion::Ram,
-    )?;
+
+    let scan = if let Some(ref elf) = elf {
+        match rtt::get_rtt_symbol_from_bytes(elf) {
+            Ok(address) => ScanRegion::Exact(address),
+            // Do not scan the memory for the control block.
+            _ => ScanRegion::Ranges(vec![]),
+        }
+    } else {
+        ScanRegion::Ram
+    };
+
+    let rtt_client = RttClient::new(create_rtt_config(&config).clone(), scan);
 
     // FIXME: we should probably figure out in a different way which core we can work with.
     // It seems arbitrary that we reset the target using the same core we use for polling RTT.
@@ -348,6 +354,7 @@ fn main_try(args: &[OsString], offset: UtcOffset) -> Result<()> {
         // GDB is also using the session, so we do not lock on the outside.
         run_rttui_app(
             name,
+            elf,
             &session,
             config,
             offset,
@@ -387,6 +394,7 @@ fn should_resume_core(config: &config::Config) -> bool {
 
 fn run_rttui_app(
     name: &str,
+    elf: Option<Vec<u8>>,
     session: &FairMutex<Session>,
     config: config::Config,
     timezone_offset: UtcOffset,
@@ -453,7 +461,7 @@ fn run_rttui_app(
         / 1_000_000;
 
     let logname = format!("{name}_{chip_name}_{timestamp_millis}");
-    let mut app = rttui::app::App::new(rtt, config, logname)?;
+    let mut app = rttui::app::App::new(rtt, elf, config, timezone_offset, logname)?;
     loop {
         app.render();
 
diff --git a/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/rttui/app.rs b/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/rttui/app.rs
index 54774043..d2f812c0 100644
--- a/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/rttui/app.rs
+++ b/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/rttui/app.rs
@@ -14,10 +14,11 @@ use ratatui::{
 };
 use std::{cell::RefCell, io::Write, rc::Rc};
 use std::{path::PathBuf, sync::mpsc::TryRecvError};
+use time::UtcOffset;
 
 use crate::{
     cmd::cargo_embed::rttui::{channel::ChannelData, tab::TabConfig},
-    util::rtt::client::RttClient,
+    util::rtt::{client::RttClient, DataFormat, DefmtProcessor, DefmtState, RttDecoder},
 };
 
 use super::super::config;
@@ -44,7 +45,19 @@ pub struct App {
 }
 
 impl App {
-    pub fn new(client: RttClient, config: config::Config, logname: String) -> Result<Self> {
+    pub fn new(
+        client: RttClient,
+        elf: Option<Vec<u8>>,
+        config: config::Config,
+        timestamp_offset: UtcOffset,
+        logname: String,
+    ) -> Result<Self> {
+        let defmt_data = if let Some(elf) = elf {
+            DefmtState::try_from_bytes(&elf)?
+        } else {
+            None
+        };
+
         let mut tab_config = config.rtt.tabs;
 
         // Create channel states
@@ -65,6 +78,17 @@ impl App {
                 });
             }
 
+            let mut channel_config = client
+                .config
+                .channel_config(number)
+                .cloned()
+                .unwrap_or_default();
+
+            if up.channel_name() == "defmt" {
+                channel_config.data_format = DataFormat::Defmt;
+            }
+
+            // Where `channel_config` is unspecified, apply default from `default_channel_config`.
             // Is a TCP publish address configured?
             let stream = config
                 .rtt
@@ -73,7 +97,31 @@ impl App {
                 .find(|up_config| up_config.channel == number)
                 .and_then(|up_config| up_config.socket);
 
-            up_channels.push(Rc::new(RefCell::new(UpChannel::new(up, stream))));
+            let data_format = match channel_config.data_format {
+                DataFormat::String => RttDecoder::String {
+                    timestamp_offset: Some(timestamp_offset),
+                    last_line_done: false,
+                },
+                DataFormat::BinaryLE => RttDecoder::BinaryLE,
+                DataFormat::Defmt if defmt_data.is_none() => {
+                    tracing::warn!("Defmt data not found in ELF file");
+                    continue;
+                }
+                DataFormat::Defmt => RttDecoder::Defmt {
+                    processor: DefmtProcessor::new(
+                        defmt_data.clone().unwrap(),
+                        channel_config.show_timestamps,
+                        channel_config.show_location,
+                        channel_config.log_format.as_deref(),
+                    ),
+                },
+            };
+
+            up_channels.push(Rc::new(RefCell::new(UpChannel::new(
+                up,
+                data_format,
+                stream,
+            ))));
         }
 
         for down in client.down_channels() {
@@ -83,7 +131,7 @@ impl App {
                 .any(|tab| tab.down_channel == Some(number))
             {
                 tab_config.push(TabConfig {
-                    up_channel: if up_channels.len() > number {
+                    up_channel: if up_channels.len() as u32 > number {
                         number
                     } else {
                         0
@@ -103,7 +151,7 @@ impl App {
             if tab.hide {
                 continue;
             }
-            let Some(up_channel) = up_channels.get(tab.up_channel) else {
+            let Some(up_channel) = up_channels.get(tab.up_channel as usize) else {
                 tracing::warn!(
                     "Configured up channel {} does not exist, skipping tab",
                     tab.up_channel
diff --git a/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/rttui/channel.rs b/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/rttui/channel.rs
index 0e189654..c4959bdc 100644
--- a/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/rttui/channel.rs
+++ b/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/rttui/channel.rs
@@ -4,7 +4,7 @@ use probe_rs::{rtt::Error, Core};
 
 use crate::{
     cmd::cargo_embed::rttui::tcp::TcpPublisher,
-    util::rtt::{client::RttClient, ChannelDataCallbacks, RttActiveUpChannel},
+    util::rtt::{client::RttClient, RttActiveUpChannel, RttDataHandler, RttDecoder},
 };
 
 pub enum ChannelData {
@@ -12,8 +12,8 @@ pub enum ChannelData {
     Binary { data: Vec<u8> },
 }
 
-impl ChannelDataCallbacks for (&mut Option<TcpPublisher>, &mut ChannelData) {
-    fn on_string_data(&mut self, _channel: usize, data: String) -> Result<(), Error> {
+impl RttDataHandler for (&mut Option<TcpPublisher>, &mut ChannelData) {
+    fn on_string_data(&mut self, _channel: u32, data: String) -> Result<(), Error> {
         if let Some(ref mut stream) = self.0 {
             stream.send(data.as_bytes());
         }
@@ -26,7 +26,7 @@ impl ChannelDataCallbacks for (&mut Option<TcpPublisher>, &mut ChannelData) {
         Ok(())
     }
 
-    fn on_binary_data(&mut self, _channel: usize, incoming: &[u8]) -> Result<(), Error> {
+    fn on_binary_data(&mut self, _channel: u32, incoming: &[u8]) -> Result<(), Error> {
         if let Some(ref mut stream) = self.0 {
             stream.send(incoming);
         }
@@ -41,22 +41,28 @@ impl ChannelDataCallbacks for (&mut Option<TcpPublisher>, &mut ChannelData) {
 }
 
 pub struct UpChannel {
-    channel_number: usize,
+    channel_number: u32,
     tcp_stream: Option<TcpPublisher>,
     channel_name: String,
+    data_format: RttDecoder,
     pub data: ChannelData,
 }
 
 impl UpChannel {
-    pub fn new(rtt_channel: &RttActiveUpChannel, tcp_stream: Option<SocketAddr>) -> Self {
+    pub fn new(
+        rtt_channel: &RttActiveUpChannel,
+        data_format: RttDecoder,
+        tcp_stream: Option<SocketAddr>,
+    ) -> Self {
         Self {
-            data: if rtt_channel.data_format.is_binary() {
+            data: if data_format.is_binary() {
                 ChannelData::Binary { data: Vec::new() }
             } else {
                 ChannelData::Strings {
                     messages: Vec::new(),
                 }
             },
+            data_format,
             tcp_stream: tcp_stream.map(TcpPublisher::new),
             channel_number: rtt_channel.number(),
             channel_name: rtt_channel.channel_name().to_string(),
@@ -64,11 +70,15 @@ impl UpChannel {
     }
 
     pub fn poll_rtt(&mut self, core: &mut Core<'_>, client: &mut RttClient) -> Result<(), Error> {
-        client.poll_channel(
-            core,
+        let bytes = client.poll_channel(core, self.channel_number)?;
+
+        self.data_format.process(
             self.channel_number,
+            bytes,
             &mut (&mut self.tcp_stream, &mut self.data),
-        )
+        )?;
+
+        Ok(())
     }
 
     pub(crate) fn channel_name(&self) -> &str {
diff --git a/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/rttui/tab.rs b/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/rttui/tab.rs
index c4f31a55..44747c35 100644
--- a/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/rttui/tab.rs
+++ b/probe-rs-tools/src/bin/probe-rs/cmd/cargo_embed/rttui/tab.rs
@@ -13,11 +13,11 @@ use super::channel::UpChannel;
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct TabConfig {
     /// Which up channel to use.
-    pub up_channel: usize,
+    pub up_channel: u32,
 
     /// Which down channel to use, if any.
     #[serde(default)]
-    pub down_channel: Option<usize>,
+    pub down_channel: Option<u32>,
 
     /// The name of the tab. If not set, the name of the up channel is used.
     #[serde(default)]
@@ -30,7 +30,7 @@ pub struct TabConfig {
 
 pub struct Tab {
     up_channel: Rc<RefCell<UpChannel>>,
-    down_channel: Option<(usize, String)>,
+    down_channel: Option<(u32, String)>,
     name: String,
     scroll_offset: usize,
     messages: Vec<String>,
@@ -41,7 +41,7 @@ pub struct Tab {
 impl Tab {
     pub fn new(
         up_channel: Rc<RefCell<UpChannel>>,
-        down_channel: Option<usize>,
+        down_channel: Option<u32>,
         name: Option<String>,
     ) -> Self {
         Self {
diff --git a/probe-rs-tools/src/bin/probe-rs/cmd/dap_server/debug_adapter/dap/adapter.rs b/probe-rs-tools/src/bin/probe-rs/cmd/dap_server/debug_adapter/dap/adapter.rs
index c4c885df..08816046 100644
--- a/probe-rs-tools/src/bin/probe-rs/cmd/dap_server/debug_adapter/dap/adapter.rs
+++ b/probe-rs-tools/src/bin/probe-rs/cmd/dap_server/debug_adapter/dap/adapter.rs
@@ -1773,7 +1773,7 @@ impl<P: ProtocolAdapter> DebugAdapter<P> {
     /// Send a custom `probe-rs-rtt-channel-config` event to the MS DAP Client, to create a window for a specific RTT channel.
     pub fn rtt_window(
         &mut self,
-        channel_number: usize,
+        channel_number: u32,
         channel_name: String,
         data_format: rtt::DataFormat,
     ) -> bool {
@@ -1790,7 +1790,7 @@ impl<P: ProtocolAdapter> DebugAdapter<P> {
     }
 
     /// Send a custom `probe-rs-rtt-data` event to the MS DAP Client, to
-    pub fn rtt_output(&mut self, channel_number: usize, rtt_data: String) -> bool {
+    pub fn rtt_output(&mut self, channel_number: u32, rtt_data: String) -> bool {
         let Ok(event_body) = serde_json::to_value(RttDataEventBody {
             channel_number,
             data: rtt_data,
diff --git a/probe-rs-tools/src/bin/probe-rs/cmd/dap_server/debug_adapter/dap/dap_types.rs b/probe-rs-tools/src/bin/probe-rs/cmd/dap_server/debug_adapter/dap/dap_types.rs
index 56e87d4e..9289872b 100644
--- a/probe-rs-tools/src/bin/probe-rs/cmd/dap_server/debug_adapter/dap/dap_types.rs
+++ b/probe-rs-tools/src/bin/probe-rs/cmd/dap_server/debug_adapter/dap/dap_types.rs
@@ -68,14 +68,14 @@ pub struct RttWindowOpened {
 #[serde(rename_all = "camelCase")]
 pub struct RttWindowOpenedArguments {
     /// The RTT channel number.
-    pub channel_number: usize,
+    pub channel_number: u32,
     pub window_is_open: bool,
 }
 
 #[derive(Clone, PartialEq, Eq, Debug, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct RttChannelEventBody {
-    pub channel_number: usize,
+    pub channel_number: u32,
     pub channel_name: String,
     pub data_format: rtt::DataFormat,
 }
@@ -83,7 +83,7 @@ pub struct RttChannelEventBody {
 #[derive(Clone, PartialEq, Eq, Debug, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct RttDataEventBody {
-    pub channel_number: usize,
+    pub channel_number: u32,
     /// RTT output
     pub data: String,
 }
diff --git a/probe-rs-tools/src/bin/probe-rs/cmd/dap_server/server/core_data.rs b/probe-rs-tools/src/bin/probe-rs/cmd/dap_server/server/core_data.rs
index 9318573f..b2c3722d 100644
--- a/probe-rs-tools/src/bin/probe-rs/cmd/dap_server/server/core_data.rs
+++ b/probe-rs-tools/src/bin/probe-rs/cmd/dap_server/server/core_data.rs
@@ -1,21 +1,24 @@
 use std::{ops::Range, path::Path};
 
 use super::session_data::{self, ActiveBreakpoint, BreakpointType, SourceLocationScope};
-use crate::cmd::dap_server::{
-    debug_adapter::{
-        dap::{
-            adapter::DebugAdapter,
-            core_status::DapStatus,
-            dap_types::{ContinuedEventBody, MessageSeverity, Source, StoppedEventBody},
+use crate::util::rtt::client::RttClient;
+use crate::util::rtt::{self, DataFormat, DefmtProcessor, DefmtState};
+use crate::{
+    cmd::dap_server::{
+        debug_adapter::{
+            dap::{
+                adapter::DebugAdapter,
+                core_status::DapStatus,
+                dap_types::{ContinuedEventBody, MessageSeverity, Source, StoppedEventBody},
+            },
+            protocol::ProtocolAdapter,
         },
-        protocol::ProtocolAdapter,
+        peripherals::svd_variables::SvdCache,
+        server::debug_rtt,
+        DebuggerError,
     },
-    peripherals::svd_variables::SvdCache,
-    server::debug_rtt,
-    DebuggerError,
+    util::rtt::RttDecoder,
 };
-use crate::util::rtt::client::RttClient;
-use crate::util::rtt::{self, DataFormat};
 use anyhow::{anyhow, Result};
 use probe_rs::{rtt::ScanRegion, Core, CoreStatus, HaltReason};
 use probe_rs_debug::VerifiedBreakpoint;
@@ -179,24 +182,22 @@ impl<'p> CoreHandle<'p> {
         rtt_config: &rtt::RttConfig,
         timestamp_offset: UtcOffset,
     ) -> Result<()> {
+        // Create the RTT client using the RTT control block address from the ELF file.
+        let elf = std::fs::read(program_binary)
+            .map_err(|error| anyhow!("Error attempting to attach to RTT: {error}"))?;
+
         let client = if let Some(client) = self.core_data.rtt_client.as_mut() {
             client
         } else {
-            // Create the RTT client using the RTT control block address from the ELF file.
-            let elf = std::fs::read(program_binary)
-                .map_err(|error| anyhow!("Error attempting to attach to RTT: {error}"))?;
-
-            let mut client = RttClient::new(
-                Some(&elf),
-                self.core.target(),
-                rtt_config.clone(),
+            let scan = match rtt::get_rtt_symbol_from_bytes(&elf) {
+                Ok(address) => ScanRegion::Exact(address),
                 // Do not scan the memory for the control block.
-                ScanRegion::Ranges(vec![]),
-            )?;
+                _ => ScanRegion::Ranges(vec![]),
+            };
 
-            client.timezone_offset = timestamp_offset;
-
-            self.core_data.rtt_client.insert(client)
+            self.core_data
+                .rtt_client
+                .insert(RttClient::new(rtt_config.clone(), scan))
         };
 
         if !client.try_attach(&mut self.core)? {
@@ -210,16 +211,59 @@ impl<'p> CoreHandle<'p> {
 
         let mut debugger_rtt_channels = vec![];
 
+        let defmt_data = DefmtState::try_from_bytes(&elf)?;
+
         for up_channel in client.up_channels() {
+            let number = up_channel.number();
+
+            let mut channel_config = rtt_config
+                .channel_config(number)
+                .cloned()
+                .unwrap_or_default();
+
+            if up_channel.channel_name() == "defmt" {
+                channel_config.data_format = DataFormat::Defmt;
+            }
+
+            // Where `channel_config` is unspecified, apply default from `default_channel_config`.
+            let show_timestamps = channel_config.show_timestamps;
+            let show_location = channel_config.show_location;
+            let log_format = channel_config.log_format.clone();
+
+            let channel_data_format = match channel_config.data_format {
+                DataFormat::String => RttDecoder::String {
+                    timestamp_offset: Some(timestamp_offset),
+                    last_line_done: false,
+                },
+                DataFormat::BinaryLE => RttDecoder::BinaryLE,
+                DataFormat::Defmt => {
+                    let Some(defmt_data) = defmt_data.clone() else {
+                        tracing::warn!("Defmt data not found in ELF file");
+                        continue;
+                    };
+
+                    RttDecoder::Defmt {
+                        processor: DefmtProcessor::new(
+                            defmt_data,
+                            show_timestamps,
+                            show_location,
+                            log_format.as_deref(),
+                        ),
+                    }
+                }
+            };
+
             debugger_rtt_channels.push(debug_rtt::DebuggerRttChannel {
                 channel_number: up_channel.number(),
                 // This value will eventually be set to true by a VSCode client request "rttWindowOpened"
                 has_client_window: false,
+                channel_data_format,
             });
+
             debug_adapter.rtt_window(
                 up_channel.number(),
                 up_channel.channel_name(),
-                DataFormat::from(&up_channel.data_format),
+                channel_config.data_format,
             );
         }
 
diff --git a/probe-rs-tools/src/bin/probe-rs/cmd/dap_server/server/debug_rtt.rs b/probe-rs-tools/src/bin/probe-rs/cmd/dap_server/server/debug_rtt.rs
index b368c728..36a2e94c 100644
--- a/probe-rs-tools/src/bin/probe-rs/cmd/dap_server/server/debug_rtt.rs
+++ b/probe-rs-tools/src/bin/probe-rs/cmd/dap_server/server/debug_rtt.rs
@@ -1,13 +1,13 @@
-use crate::util::rtt::client::RttClient;
+use crate::util::rtt::{client::RttClient, RttDataHandler};
 use crate::{
     cmd::dap_server::{
         debug_adapter::{dap::adapter::*, protocol::ProtocolAdapter},
         DebuggerError,
     },
-    util::rtt::ChannelDataCallbacks,
+    util::rtt::RttDecoder,
 };
 use anyhow::anyhow;
-use probe_rs::{rtt::Error, Core};
+use probe_rs::{rtt, Core};
 
 /// Manage the active RTT target for a specific SessionData, as well as provide methods to reliably move RTT from target, through the debug_adapter, to the client.
 pub struct RttConnection {
@@ -43,9 +43,10 @@ impl RttConnection {
 }
 
 pub(crate) struct DebuggerRttChannel {
-    pub(crate) channel_number: usize,
+    pub(crate) channel_number: u32,
     // We will not poll target RTT channels until we have confirmation from the client that the output window has been opened.
     pub(crate) has_client_window: bool,
+    pub(crate) channel_data_format: RttDecoder,
 }
 
 impl DebuggerRttChannel {
@@ -64,11 +65,18 @@ impl DebuggerRttChannel {
 
         let mut out = StringCollector { data: None };
 
-        if let Err(e) = client.poll_channel(core, self.channel_number, &mut out) {
-            debug_adapter
-                .show_error_message(&DebuggerError::Other(anyhow!(e)))
-                .ok();
-            return false;
+        match client.poll_channel(core, self.channel_number) {
+            Ok(bytes) => {
+                self.channel_data_format
+                    .process(self.channel_number, bytes, &mut out)
+                    .ok();
+            }
+            Err(e) => {
+                debug_adapter
+                    .show_error_message(&DebuggerError::Other(anyhow!(e)))
+                    .ok();
+                return false;
+            }
         }
 
         match out.data {
@@ -82,8 +90,8 @@ struct StringCollector {
     data: Option<String>,
 }
 
-impl ChannelDataCallbacks for StringCollector {
-    fn on_string_data(&mut self, _channel: usize, data: String) -> Result<(), Error> {
+impl RttDataHandler for StringCollector {
+    fn on_string_data(&mut self, _channel: u32, data: String) -> Result<(), rtt::Error> {
         self.data = Some(data);
         Ok(())
     }
diff --git a/probe-rs-tools/src/bin/probe-rs/cmd/info.rs b/probe-rs-tools/src/bin/probe-rs/cmd/info.rs
index 71e7d8d4..8fcbd62b 100644
--- a/probe-rs-tools/src/bin/probe-rs/cmd/info.rs
+++ b/probe-rs-tools/src/bin/probe-rs/cmd/info.rs
@@ -76,7 +76,7 @@ impl Cmd {
                 dry_run: self.common.dry_run,
             };
 
-            client
+            let result = client
                 .info(req, |message| {
                     let is_success =
                         matches!(message, InfoEvent::Idcode { .. } | InfoEvent::ArmDp(_));
@@ -92,7 +92,11 @@ impl Cmd {
                         errors.push(message);
                     }
                 })
-                .await?;
+                .await;
+
+            if let Err(error) = result {
+                println!("Error while probing target: {}", error);
+            }
 
             if successes.is_empty() {
                 for message in errors {
diff --git a/probe-rs-tools/src/bin/probe-rs/cmd/run.rs b/probe-rs-tools/src/bin/probe-rs/cmd/run.rs
index b12ad142..a2dd206a 100644
--- a/probe-rs-tools/src/bin/probe-rs/cmd/run.rs
+++ b/probe-rs-tools/src/bin/probe-rs/cmd/run.rs
@@ -2,9 +2,8 @@ use std::path::{Path, PathBuf};
 
 use crate::rpc::client::RpcClient;
 use crate::rpc::functions::monitor::{MonitorMode, MonitorOptions};
-use crate::rpc::functions::rtt_client::LogOptions;
 
-use crate::util::cli;
+use crate::util::cli::{self, rtt_client};
 use crate::util::common_options::{BinaryDownloadOptions, ProbeOptions};
 use crate::FormatOptions;
 
@@ -12,6 +11,7 @@ use libtest_mimic::{Arguments, FormatSetting};
 use probe_rs::flashing::FileDownloadError;
 use std::fs::File;
 use std::io::Read;
+use time::UtcOffset;
 
 /// Options only used in normal run mode
 #[derive(Debug, clap::Parser, Clone)]
@@ -146,6 +146,8 @@ pub struct SharedOptions {
 
     /// The format string to use when printing defmt encoded log messages from the target.
     ///
+    /// You can also use one of two presets: oneline (default) and full.
+    ///
     /// See <https://defmt.ferrous-systems.com/custom-log-output>
     #[clap(long)]
     pub(crate) log_format: Option<String>,
@@ -156,22 +158,26 @@ pub struct SharedOptions {
 }
 
 impl Cmd {
-    pub async fn run(self, client: RpcClient) -> anyhow::Result<()> {
+    pub async fn run(self, client: RpcClient, utc_offset: UtcOffset) -> anyhow::Result<()> {
         // Detect run mode based on ELF file
         let run_mode = detect_run_mode(&self)?;
 
         let session = cli::attach_probe(&client, self.shared_options.probe_options, true).await?;
 
-        let rtt_client = session
-            .create_rtt_client(
-                Some(self.shared_options.path.clone()),
-                LogOptions {
-                    no_location: self.shared_options.no_location,
-                    log_format: self.shared_options.log_format,
-                    rtt_scan_memory: self.shared_options.rtt_scan_memory,
-                },
-            )
-            .await?;
+        let mut rtt_client = rtt_client(
+            &session,
+            &self.shared_options.path,
+            match self.shared_options.rtt_scan_memory {
+                true => crate::rpc::functions::rtt_client::ScanRegion::TargetDefault,
+                false => crate::rpc::functions::rtt_client::ScanRegion::Ranges(vec![]),
+            },
+            self.shared_options.log_format,
+            !self.shared_options.no_location,
+            Some(utc_offset),
+        )
+        .await?;
+
+        let client_handle = rtt_client.handle();
 
         // Flash firmware
         let boot_info = cli::flash(
@@ -180,7 +186,7 @@ impl Cmd {
             self.shared_options.chip_erase,
             self.shared_options.format_options,
             self.shared_options.download_options,
-            Some(rtt_client),
+            Some(&mut rtt_client),
         )
         .await?;
 
@@ -215,10 +221,11 @@ impl Cmd {
                 &session,
                 MonitorMode::Run(boot_info),
                 &self.shared_options.path,
+                Some(rtt_client),
                 MonitorOptions {
                     catch_reset: self.run_options.catch_reset,
                     catch_hardfault: self.run_options.catch_hardfault,
-                    rtt_client: Some(rtt_client),
+                    rtt_client: Some(client_handle),
                 },
                 self.shared_options.always_print_stacktrace,
             )
diff --git a/probe-rs-tools/src/bin/probe-rs/main.rs b/probe-rs-tools/src/bin/probe-rs/main.rs
index b929422a..6aa8e389 100644
--- a/probe-rs-tools/src/bin/probe-rs/main.rs
+++ b/probe-rs-tools/src/bin/probe-rs/main.rs
@@ -97,7 +97,7 @@ struct Cli {
 }
 
 impl Cli {
-    async fn run(self, client: RpcClient, _config: Config) -> Result<()> {
+    async fn run(self, client: RpcClient, _config: Config, utc_offset: UtcOffset) -> Result<()> {
         let lister = Lister::new();
         match self.subcommand {
             Subcommand::DapServer { .. } => unreachable!(),
@@ -109,8 +109,8 @@ impl Cli {
             Subcommand::Reset(cmd) => cmd.run(client).await,
             Subcommand::Debug(cmd) => cmd.run(&lister),
             Subcommand::Download(cmd) => cmd.run(client).await,
-            Subcommand::Run(cmd) => cmd.run(client).await,
-            Subcommand::Attach(cmd) => cmd.run(client).await,
+            Subcommand::Run(cmd) => cmd.run(client, utc_offset).await,
+            Subcommand::Attach(cmd) => cmd.run(client, utc_offset).await,
             Subcommand::Verify(cmd) => cmd.run(client).await,
             Subcommand::Erase(cmd) => cmd.run(client).await,
             Subcommand::Trace(cmd) => cmd.run(&lister),
@@ -465,7 +465,7 @@ async fn main() -> Result<()> {
             "The subcommand is not supported in remote mode."
         );
 
-        matches.run(client, config).await?;
+        matches.run(client, config, utc_offset).await?;
         // TODO: handle the report
         return Ok(());
     }
@@ -476,7 +476,7 @@ async fn main() -> Result<()> {
 
     // Run the command locally.
     let client = RpcClient::new_local_from_wire(tx, rx);
-    let result = matches.run(client, config).await;
+    let result = matches.run(client, config, utc_offset).await;
 
     // Wait for the server to shut down
     _ = handle.await.unwrap();
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 d7146a98..e74dbb55 100644
--- a/probe-rs-tools/src/bin/probe-rs/rpc/client.rs
+++ b/probe-rs-tools/src/bin/probe-rs/rpc/client.rs
@@ -42,7 +42,7 @@ use crate::{
             },
             reset::ResetCoreRequest,
             resume::ResumeAllCoresRequest,
-            rtt_client::{CreateRttClientRequest, LogOptions},
+            rtt_client::{CreateRttClientRequest, RttClientData, ScanRegion},
             stack_trace::{StackTraces, TakeStackTraceRequest},
             test::{ListTestsRequest, RunTestRequest, Test, TestResult, Tests},
             AttachEndpoint, BuildEndpoint, ChipInfoEndpoint, CreateRttClientEndpoint,
@@ -58,7 +58,7 @@ use crate::{
         transport::memory::{PostcardReceiver, PostcardSender, WireRx, WireTx},
         Key,
     },
-    util::rtt::client::RttClient,
+    util::rtt::{client::RttClient, RttChannelConfig},
     FormatOptions,
 };
 
@@ -552,19 +552,14 @@ impl SessionInterface {
 
     pub async fn create_rtt_client(
         &self,
-        mut path: Option<PathBuf>,
-        log_options: LogOptions,
-    ) -> anyhow::Result<Key<RttClient>> {
-        // TODO: RTT shouldn't need the firmware, we can extract location and process output locally
-        if let Some(ref mut path) = path {
-            *path = self.client.upload_file(&*path).await?;
-        }
-
+        scan_regions: ScanRegion,
+        config: Vec<RttChannelConfig>,
+    ) -> anyhow::Result<RttClientData> {
         self.client
             .send_resp::<CreateRttClientEndpoint, _>(&CreateRttClientRequest {
                 sessid: self.sessid,
-                path: path.map(|path| path.display().to_string()),
-                log_options,
+                scan_regions,
+                config,
             })
             .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 ca63c595..9eeb0977 100644
--- a/probe-rs-tools/src/bin/probe-rs/rpc/functions.rs
+++ b/probe-rs-tools/src/bin/probe-rs/rpc/functions.rs
@@ -145,34 +145,6 @@ pub struct RpcSpawnContext {
 }
 
 impl RpcSpawnContext {
-    pub async fn publish<T>(&self, seq_no: VarSeq, msg: &T::Message) -> anyhow::Result<()>
-    where
-        T: ?Sized,
-        T: Topic,
-        T::Message: Serialize + Schema,
-    {
-        anyhow::ensure!(!self.token.is_cancelled(), "RPC call cancelled");
-
-        self.sender
-            .publish::<T>(seq_no, msg)
-            .await
-            .map_err(|e| anyhow!("{:?}", e))
-    }
-
-    pub fn publish_blocking<T>(&self, seq_no: VarSeq, msg: T::Message) -> anyhow::Result<()>
-    where
-        T: Topic,
-        T::Message: Serialize + Schema + Send + Sync + Sized + 'static,
-    {
-        let handle = tokio::runtime::Handle::current();
-        let this = self.clone();
-        handle.block_on(async move {
-            tokio::spawn(async move { this.publish::<T>(seq_no, &msg).await })
-                .await
-                .unwrap()
-        })
-    }
-
     fn dry_run(&self, sessid: Key<Session>) -> bool {
         self.state.dry_run(sessid)
     }
@@ -191,6 +163,49 @@ impl RpcSpawnContext {
     pub fn cancellation_token(&self) -> CancellationToken {
         self.token.clone()
     }
+
+    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,
+        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 ctx = self.clone();
+        let blocking = tokio::task::spawn_blocking(move || task(ctx, request, channel_sender));
+
+        tokio::select! {
+            _ = sender => unreachable!(),
+            response = blocking => {
+                response.unwrap()
+            }
+        }
+    }
 }
 
 pub struct RpcContext {
@@ -264,6 +279,20 @@ impl RpcContext {
     pub fn lister(&self) -> Lister {
         Lister::new()
     }
+
+    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,
+        F: Send + 'static,
+        REQ: Send + 'static,
+        RESP: Send + 'static,
+    {
+        self.spawn_ctxt()
+            .run_blocking::<T, F, REQ, RESP>(request, task)
+            .await
+    }
 }
 
 async fn cancel_handler(
diff --git a/probe-rs-tools/src/bin/probe-rs/rpc/functions/flash.rs b/probe-rs-tools/src/bin/probe-rs/rpc/functions/flash.rs
index a7dab8fc..d144de57 100644
--- a/probe-rs-tools/src/bin/probe-rs/rpc/functions/flash.rs
+++ b/probe-rs-tools/src/bin/probe-rs/rpc/functions/flash.rs
@@ -1,7 +1,4 @@
-use postcard_rpc::{
-    header::{VarHeader, VarSeq},
-    server::SpawnContext,
-};
+use postcard_rpc::header::VarHeader;
 use postcard_schema::Schema;
 use probe_rs::{
     flashing::{self, FileDownloadError, FlashLoader, FlashProgress},
@@ -9,6 +6,7 @@ use probe_rs::{
     Session,
 };
 use serde::{Deserialize, Serialize};
+use tokio::sync::mpsc::Sender;
 
 use crate::{
     rpc::{
@@ -360,13 +358,15 @@ impl From<probe_rs::flashing::BootInfo> for BootInfo {
 }
 
 pub async fn flash(ctx: &mut RpcContext, _header: VarHeader, request: FlashRequest) -> NoResponse {
-    let ctx = ctx.spawn_ctxt();
-    tokio::task::spawn_blocking(move || flash_impl(ctx, request))
+    ctx.run_blocking::<ProgressEventTopic, _, _, _>(request, flash_impl)
         .await
-        .unwrap()
 }
 
-fn flash_impl(ctx: RpcSpawnContext, request: FlashRequest) -> NoResponse {
+fn flash_impl(
+    ctx: RpcSpawnContext,
+    request: FlashRequest,
+    sender: Sender<ProgressEvent>,
+) -> NoResponse {
     let dry_run = ctx.dry_run(request.sessid);
     let mut session = ctx.session_blocking(request.sessid);
 
@@ -402,10 +402,7 @@ fn flash_impl(ctx: RpcSpawnContext, request: FlashRequest) -> NoResponse {
     options.verify = request.options.verify;
     options.disable_double_buffering = request.options.disable_double_buffering;
     options.progress = Some(FlashProgress::new(move |event| {
-        ProgressEvent::from_library_event(event, |event| {
-            ctx.publish_blocking::<ProgressEventTopic>(VarSeq::Seq2(0), event)
-                .unwrap()
-        });
+        ProgressEvent::from_library_event(event, |event| sender.blocking_send(event).unwrap());
     }));
 
     // run flash download
@@ -438,13 +435,15 @@ pub enum EraseCommand {
 }
 
 pub async fn erase(ctx: &mut RpcContext, _header: VarHeader, request: EraseRequest) -> NoResponse {
-    let ctx = ctx.spawn_ctxt();
-    tokio::task::spawn_blocking(move || erase_impl(ctx, request))
+    ctx.run_blocking::<ProgressEventTopic, _, _, _>(request, erase_impl)
         .await
-        .unwrap()
 }
 
-fn erase_impl(ctx: RpcSpawnContext, request: EraseRequest) -> NoResponse {
+fn erase_impl(
+    ctx: RpcSpawnContext,
+    request: EraseRequest,
+    sender: Sender<ProgressEvent>,
+) -> NoResponse {
     let mut session = ctx.session_blocking(request.sessid);
 
     let progress = FlashProgress::new(move |event| {
@@ -465,8 +464,7 @@ fn erase_impl(ctx: RpcSpawnContext, request: EraseRequest) -> NoResponse {
             | ProgressEvent::Failed(Operation::Erase)
             | ProgressEvent::Finished(Operation::Erase) = event
             {
-                ctx.publish_blocking::<ProgressEventTopic>(VarSeq::Seq2(0), event)
-                    .unwrap()
+                sender.blocking_send(event).unwrap()
             }
         });
     });
@@ -497,13 +495,15 @@ pub async fn verify(
     _header: VarHeader,
     request: VerifyRequest,
 ) -> VerifyResponse {
-    let ctx = ctx.spawn_ctxt();
-    tokio::task::spawn_blocking(move || verify_impl(ctx, request))
+    ctx.run_blocking::<ProgressEventTopic, _, _, _>(request, verify_impl)
         .await
-        .unwrap()
 }
 
-fn verify_impl(ctx: RpcSpawnContext, request: VerifyRequest) -> VerifyResponse {
+fn verify_impl(
+    ctx: RpcSpawnContext,
+    request: VerifyRequest,
+    sender: Sender<ProgressEvent>,
+) -> VerifyResponse {
     let mut session = ctx.session_blocking(request.sessid);
     let loader = ctx.object_mut_blocking(request.loader);
 
@@ -525,8 +525,7 @@ fn verify_impl(ctx: RpcSpawnContext, request: VerifyRequest) -> VerifyResponse {
             | ProgressEvent::Failed(Operation::Verify)
             | ProgressEvent::Finished(Operation::Verify) = event
             {
-                ctx.publish_blocking::<ProgressEventTopic>(VarSeq::Seq2(0), event)
-                    .unwrap()
+                sender.blocking_send(event).unwrap()
             }
         });
     });
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 8768e701..0ce7834e 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
@@ -1,4 +1,4 @@
-use std::{fmt::Write, num::NonZeroU32, time::Duration};
+use std::{num::NonZeroU32, time::Duration};
 
 use crate::{
     rpc::{
@@ -8,13 +8,12 @@ use crate::{
     },
     util::rtt::client::RttClient,
 };
-use postcard_rpc::{
-    header::{VarHeader, VarSeq},
-    server::Sender,
-};
+use anyhow::Context;
+use postcard_rpc::{header::VarHeader, server::Sender};
 use postcard_schema::Schema;
 use probe_rs::{semihosting::SemihostingCommand, BreakpointCause, Core, HaltReason, Session};
 use serde::{Deserialize, Serialize};
+use tokio::sync::mpsc;
 
 #[derive(Serialize, Deserialize, Schema)]
 pub enum MonitorMode {
@@ -24,7 +23,7 @@ pub enum MonitorMode {
 
 #[derive(Serialize, Deserialize, Schema)]
 pub enum MonitorEvent {
-    RttOutput(String), // Monitor supports a single channel only
+    RttOutput { channel: u32, bytes: Vec<u8> },
     SemihostingOutput(SemihostingOutput),
 }
 
@@ -47,14 +46,15 @@ pub struct MonitorRequest {
 }
 
 pub async fn monitor(
-    ctx: RpcSpawnContext,
+    mut ctx: RpcSpawnContext,
     header: VarHeader,
     request: MonitorRequest,
     sender: Sender<WireTxImpl>,
 ) {
-    let resp = tokio::task::spawn_blocking(move || monitor_impl(ctx, request).map_err(Into::into))
+    let resp = ctx
+        .run_blocking::<MonitorTopic, _, _, _>(request, monitor_impl)
         .await
-        .unwrap();
+        .map_err(Into::into);
 
     sender
         .reply::<MonitorEndpoint>(header.seq_no, &resp)
@@ -62,17 +62,15 @@ pub async fn monitor(
         .unwrap();
 }
 
-fn monitor_impl(ctx: RpcSpawnContext, request: MonitorRequest) -> anyhow::Result<()> {
+fn monitor_impl(
+    ctx: RpcSpawnContext,
+    request: MonitorRequest,
+    sender: mpsc::Sender<MonitorEvent>,
+) -> anyhow::Result<()> {
     let mut session = ctx.session_blocking(request.sessid);
 
-    let mut semihosting_sink = MonitorEventHandler::new({
-        let ctx = ctx.clone();
-        move |event| {
-            ctx.publish_blocking::<MonitorTopic>(VarSeq::Seq2(0), event)
-                .unwrap();
-        }
-    });
-    let mut rtt_sink = RttStreamer::new(ctx.clone());
+    let mut semihosting_sink =
+        MonitorEventHandler::new(|event| sender.blocking_send(event).unwrap());
 
     let mut rtt_client = request
         .options
@@ -117,7 +115,11 @@ fn monitor_impl(ctx: RpcSpawnContext, request: MonitorRequest) -> anyhow::Result
         &mut core,
         request.options.catch_hardfault,
         request.options.catch_reset,
-        &mut rtt_sink,
+        |channel, bytes| {
+            sender
+                .blocking_send(MonitorEvent::RttOutput { channel, bytes })
+                .with_context(|| "Failed to send RTT output")
+        },
         None,
         |halt_reason, core| semihosting_sink.handle_halt(halt_reason, core),
     )?;
@@ -125,27 +127,6 @@ fn monitor_impl(ctx: RpcSpawnContext, request: MonitorRequest) -> anyhow::Result
     Ok(())
 }
 
-pub struct RttStreamer {
-    ctx: RpcSpawnContext,
-}
-
-impl RttStreamer {
-    pub fn new(ctx: RpcSpawnContext) -> Self {
-        Self { ctx }
-    }
-}
-
-impl Write for RttStreamer {
-    fn write_str(&mut self, s: &str) -> std::fmt::Result {
-        self.ctx
-            .publish_blocking::<MonitorTopic>(
-                VarSeq::Seq2(0),
-                MonitorEvent::RttOutput(s.to_string()),
-            )
-            .map_err(|_| std::fmt::Error)
-    }
-}
-
 struct MonitorEventHandler<F: FnMut(MonitorEvent)> {
     sender: F,
     semihosting_reader: SemihostingReader,
diff --git a/probe-rs-tools/src/bin/probe-rs/rpc/functions/rtt_client.rs b/probe-rs-tools/src/bin/probe-rs/rpc/functions/rtt_client.rs
index 046b1325..e9ed3d92 100644
--- a/probe-rs-tools/src/bin/probe-rs/rpc/functions/rtt_client.rs
+++ b/probe-rs-tools/src/bin/probe-rs/rpc/functions/rtt_client.rs
@@ -5,34 +5,49 @@ use crate::{
     },
     util::rtt::{client::RttClient, RttChannelConfig, RttConfig},
 };
-use anyhow::Context as _;
 use postcard_rpc::header::VarHeader;
 use postcard_schema::Schema;
-use probe_rs::{flashing::FormatKind, rtt::ScanRegion, Session};
+use probe_rs::{rtt, Session};
 use serde::{Deserialize, Serialize};
 
-#[derive(Default, Debug, Serialize, Deserialize, Schema)]
-pub struct LogOptions {
-    /// Suppress filename and line number information from the rtt log
-    pub no_location: bool,
+/// Used to specify which memory regions to scan for the RTT control block.
+#[derive(Clone, Debug, Default, Serialize, Deserialize, Schema)]
+pub enum ScanRegion {
+    /// Scans all RAM regions known to probe-rs. This is the default and should always work, however
+    /// if your device has a lot of RAM, scanning all of it is slow.
+    #[default]
+    Ram,
 
-    /// The format string to use when printing defmt encoded log messages from the target.
-    ///
-    /// See https://defmt.ferrous-systems.com/custom-log-output
-    pub log_format: Option<String>,
+    /// Limit scanning to the memory addresses covered by the default region of the target.
+    TargetDefault,
 
-    /// Scan the memory to find the RTT control block
-    pub rtt_scan_memory: bool,
+    /// Limit scanning to the memory addresses covered by all of the given ranges. It is up to the
+    /// user to ensure that reading from this range will not read from undefined memory.
+    Ranges(Vec<(u64, u64)>),
+
+    /// Tries to find the control block starting at this exact address. It is up to the user to
+    /// ensure that reading the necessary bytes after the pointer will no read from undefined
+    /// memory.
+    Exact(u64),
 }
 
 #[derive(Serialize, Deserialize, Schema)]
 pub struct CreateRttClientRequest {
     pub sessid: Key<Session>,
-    pub path: Option<String>,
-    pub log_options: LogOptions,
+
+    /// Scan the memory to find the RTT control block
+    pub scan_regions: ScanRegion,
+
+    /// Channel configuration.
+    pub config: Vec<RttChannelConfig>,
 }
 
-pub type CreateRttClientResponse = RpcResult<Key<RttClient>>;
+#[derive(Serialize, Deserialize, Schema)]
+pub struct RttClientData {
+    pub handle: Key<RttClient>,
+}
+
+pub type CreateRttClientResponse = RpcResult<RttClientData>;
 
 pub async fn create_rtt_client(
     ctx: &mut RpcContext,
@@ -41,39 +56,24 @@ pub async fn create_rtt_client(
 ) -> CreateRttClientResponse {
     let session = ctx.session(request.sessid).await;
 
-    let rtt_scan_regions = match request.log_options.rtt_scan_memory {
-        true => session.target().rtt_scan_regions.clone(),
-        false => ScanRegion::Ranges(vec![]),
-    };
-    let mut rtt_config = RttConfig::default();
-    rtt_config.channels.push(RttChannelConfig {
-        channel_number: Some(0),
-        show_location: !request.log_options.no_location,
-        log_format: request.log_options.log_format.clone(),
-        ..Default::default()
-    });
-    let elf = if let Some(path) = request.path.as_deref() {
-        let format = FormatKind::from_optional(session.target().default_format.as_deref())
-            .expect("Failed to parse a default binary format. This shouldn't happen.");
-        if matches!(format, FormatKind::Elf | FormatKind::Idf) {
-            Some(
-                tokio::fs::read(path)
-                    .await
-                    .context("Failed to open firmware binary")?,
-            )
-        } else {
-            None
+    let rtt_scan_regions = match request.scan_regions {
+        ScanRegion::Ram => rtt::ScanRegion::Ram,
+        ScanRegion::TargetDefault => session.target().rtt_scan_regions.clone(),
+        ScanRegion::Ranges(ranges) => {
+            rtt::ScanRegion::Ranges(ranges.into_iter().map(|(start, end)| start..end).collect())
         }
-    } else {
-        None
+        ScanRegion::Exact(addr) => rtt::ScanRegion::Exact(addr),
     };
 
     let client = RttClient::new(
-        elf.as_deref(),
-        session.target(),
-        rtt_config,
+        RttConfig {
+            enabled: true,
+            channels: request.config,
+        },
         rtt_scan_regions,
-    )?;
+    );
 
-    Ok(ctx.store_object(client).await)
+    Ok(RttClientData {
+        handle: ctx.store_object(client).await,
+    })
 }
diff --git a/probe-rs-tools/src/bin/probe-rs/rpc/utils/run_loop.rs b/probe-rs-tools/src/bin/probe-rs/rpc/utils/run_loop.rs
index be5eb229..f68c6360 100644
--- a/probe-rs-tools/src/bin/probe-rs/rpc/utils/run_loop.rs
+++ b/probe-rs-tools/src/bin/probe-rs/rpc/utils/run_loop.rs
@@ -1,14 +1,12 @@
 use tokio_util::sync::CancellationToken;
 
-use std::fmt::Write;
 use std::thread;
 use std::time::{Duration, Instant};
 
 use anyhow::{anyhow, Result};
-use probe_rs::{rtt::Error as RttError, Core, Error, HaltReason, VectorCatchCondition};
+use probe_rs::{Core, Error, HaltReason, VectorCatchCondition};
 
 use crate::util::rtt::client::RttClient;
-use crate::util::rtt::ChannelDataCallbacks;
 
 pub struct RunLoop<'a> {
     pub core_id: usize,
@@ -40,7 +38,7 @@ impl RunLoop<'_> {
         core: &mut Core,
         catch_hardfault: bool,
         catch_reset: bool,
-        output_stream: &mut dyn Write,
+        rtt_callback: impl FnMut(u32, Vec<u8>) -> Result<()>,
         timeout: Option<Duration>,
         mut predicate: F,
     ) -> Result<ReturnReason<R>>
@@ -71,7 +69,7 @@ impl RunLoop<'_> {
         }
         let start = Instant::now();
 
-        let result = self.do_run_until(core, output_stream, timeout, start, &mut predicate);
+        let result = self.do_run_until(core, rtt_callback, timeout, start, &mut predicate);
 
         // Always clean up after RTT but don't overwrite the original result.
         if let Some(ref mut rtt_client) = self.rtt_client {
@@ -89,7 +87,7 @@ impl RunLoop<'_> {
     fn do_run_until<F, R>(
         &mut self,
         core: &mut Core,
-        output_stream: &mut dyn Write,
+        mut rtt_callback: impl FnMut(u32, Vec<u8>) -> Result<()>,
         timeout: Option<Duration>,
         start: Instant,
         predicate: &mut F,
@@ -123,11 +121,23 @@ impl RunLoop<'_> {
                 }
             }
 
-            let had_rtt_data = if let Some(ref mut rtt_client) = self.rtt_client {
-                poll_rtt(rtt_client, core, output_stream)?
-            } else {
-                false
-            };
+            let mut had_rtt_data = false;
+            if let Some(ref mut rtt_client) = self.rtt_client {
+                _ = rtt_client.try_attach(core);
+                for channel in 0..rtt_client.up_channels().len() {
+                    let bytes = rtt_client.poll_channel(core, channel as u32)?;
+                    if !bytes.is_empty() {
+                        had_rtt_data = true;
+                        let res = rtt_callback(channel as u32, bytes.to_vec());
+
+                        if self.cancellation_token.is_cancelled() {
+                            return Ok(ReturnReason::Cancelled);
+                        }
+
+                        res?;
+                    }
+                }
+            }
 
             if let Some(reason) = return_reason {
                 return reason;
@@ -156,37 +166,3 @@ impl RunLoop<'_> {
         }
     }
 }
-
-/// Poll RTT and print the received buffer.
-fn poll_rtt<S: Write + ?Sized>(
-    rtt_client: &mut RttClient,
-    core: &mut Core<'_>,
-    out_stream: &mut S,
-) -> Result<bool, anyhow::Error> {
-    struct OutCollector<'a, O: Write + ?Sized> {
-        out_stream: &'a mut O,
-        had_data: bool,
-    }
-
-    impl<O: Write + ?Sized> ChannelDataCallbacks for OutCollector<'_, O> {
-        fn on_string_data(&mut self, _channel: usize, data: String) -> Result<(), RttError> {
-            if data.is_empty() {
-                return Ok(());
-            }
-            self.had_data = true;
-            self.out_stream
-                .write_str(&data)
-                .map_err(|err| anyhow!(err))?;
-            Ok(())
-        }
-    }
-
-    let mut out = OutCollector {
-        out_stream,
-        had_data: false,
-    };
-
-    rtt_client.poll(core, &mut out)?;
-
-    Ok(out.had_data)
-}
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 86120013..3100c7af 100644
--- a/probe-rs-tools/src/bin/probe-rs/util/cli.rs
+++ b/probe-rs-tools/src/bin/probe-rs/util/cli.rs
@@ -1,10 +1,12 @@
 //! CLI-specific building blocks.
 
-use std::{future::Future, path::Path, time::Instant};
+use std::{future::Future, ops::DerefMut, path::Path, time::Instant};
 
+use anyhow::Context;
 use colored::Colorize;
 use libtest_mimic::{Failed, Trial};
-use tokio::runtime::Handle;
+use time::UtcOffset;
+use tokio::{runtime::Handle, sync::mpsc::UnboundedSender};
 use tokio_util::sync::CancellationToken;
 
 use crate::{
@@ -16,6 +18,7 @@ use crate::{
             probe::{
                 AttachRequest, AttachResult, DebugProbeEntry, DebugProbeSelector, SelectProbeResult,
             },
+            rtt_client::ScanRegion,
             stack_trace::StackTrace,
             test::{Test, TestResult},
             CancelTopic,
@@ -26,7 +29,10 @@ use crate::{
         common_options::{BinaryDownloadOptions, ProbeOptions},
         flash::CliProgressBars,
         logging,
-        rtt::client::RttClient,
+        rtt::{
+            self, client::RttClient, DataFormat, DefmtProcessor, DefmtState, RttChannelConfig,
+            RttDataHandler, RttDecoder, RttSymbolError,
+        },
     },
     FormatOptions,
 };
@@ -111,13 +117,86 @@ pub async fn select_probe(
     }
 }
 
+pub async fn rtt_client(
+    session: &SessionInterface,
+    path: &Path,
+    mut scan_regions: ScanRegion,
+    log_format: Option<String>,
+    show_location: bool,
+    timestamp_offset: Option<UtcOffset>,
+) -> anyhow::Result<CliRttClient> {
+    let elf = tokio::fs::read(path)
+        .await
+        .with_context(|| format!("Failed to read firmware from {}", path.display()))?;
+
+    let mut load_defmt_data = false;
+    match rtt::get_rtt_symbol_from_bytes(&elf) {
+        // Do not scan the memory for the control block.
+        Ok(address) => {
+            scan_regions = ScanRegion::Exact(address);
+            load_defmt_data = true;
+        }
+        Err(RttSymbolError::RttSymbolNotFound) => {
+            load_defmt_data = true;
+        }
+        _ => {}
+    }
+
+    let defmt_data = if load_defmt_data {
+        DefmtState::try_from_bytes(&elf)?
+    } else {
+        None
+    };
+
+    // The CLI only supports a single RTT up channel, and no down channels.
+    let data_format;
+    let decoder = if let Some(defmt_data) = defmt_data.clone() {
+        data_format = DataFormat::Defmt;
+        RttDecoder::Defmt {
+            processor: DefmtProcessor::new(
+                defmt_data,
+                timestamp_offset.is_some(),
+                show_location,
+                log_format.as_deref(),
+            ),
+        }
+    } else {
+        data_format = DataFormat::String;
+        RttDecoder::String {
+            timestamp_offset,
+            last_line_done: false,
+        }
+    };
+
+    let rtt_client = session
+        .create_rtt_client(
+            scan_regions,
+            vec![RttChannelConfig {
+                channel_number: Some(0),
+                mode: if data_format == DataFormat::Defmt {
+                    Some(rtt::ChannelMode::BlockIfFull)
+                } else {
+                    None
+                },
+                ..Default::default()
+            }],
+        )
+        .await?;
+    let client_handle = rtt_client.handle;
+
+    Ok(CliRttClient {
+        handle: client_handle,
+        data_format: decoder,
+    })
+}
+
 pub async fn flash(
     session: &SessionInterface,
     path: &Path,
     chip_erase: bool,
     format: FormatOptions,
     download_options: BinaryDownloadOptions,
-    rtt_client: Option<Key<RttClient>>,
+    rtt_client: Option<&mut CliRttClient>,
 ) -> anyhow::Result<BootInfo> {
     // Start timer.
     let flash_timer = Instant::now();
@@ -169,7 +248,7 @@ pub async fn flash(
             .flash(
                 options,
                 loader.loader,
-                rtt_client,
+                rtt_client.as_ref().map(|c| c.handle),
                 &mut handle_progress_events,
             )
             .await?;
@@ -204,10 +283,13 @@ pub async fn monitor(
     session: &SessionInterface,
     mode: MonitorMode,
     path: &Path,
+    mut rtt_client: Option<CliRttClient>,
     options: MonitorOptions,
     print_stack_trace: bool,
 ) -> anyhow::Result<()> {
-    let monitor = session.monitor(mode, options, print_monitor_event);
+    let monitor = session.monitor(mode, options, |msg| {
+        print_monitor_event(&mut rtt_client.as_mut(), msg)
+    });
 
     let mut cancelled = false;
 
@@ -230,14 +312,17 @@ pub async fn test(
     libtest_args: libtest_mimic::Arguments,
     print_stack_trace: bool,
     path: &Path,
-    rtt_client: Option<Key<RttClient>>,
+    mut rtt_client: Option<CliRttClient>,
 ) -> anyhow::Result<()> {
     tracing::info!("libtest args {:?}", libtest_args);
     let token = CancellationToken::new();
 
+    let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel::<MonitorEvent>();
+
+    let rtt_handle = rtt_client.as_ref().map(|rtt| rtt.handle);
     let test = async {
         let tests = session
-            .list_tests(boot_info, rtt_client, print_monitor_event)
+            .list_tests(boot_info, rtt_handle, |msg| sender.send(msg).unwrap())
             .await?;
 
         if token.is_cancelled() {
@@ -247,7 +332,7 @@ pub async fn test(
         let tests = tests
             .tests
             .into_iter()
-            .map(|test| create_trial(session, path, rtt_client, &token, test))
+            .map(|test| create_trial(session, path, rtt_handle, sender.clone(), &token, test))
             .collect::<Vec<_>>();
 
         tokio::task::spawn_blocking(move || {
@@ -260,7 +345,21 @@ pub async fn test(
         .await?
     };
 
-    let result = with_ctrl_c(test, async {
+    let log = async {
+        while let Some(event) = receiver.recv().await {
+            print_monitor_event(&mut rtt_client.as_mut(), event);
+        }
+        futures_util::future::pending().await
+    };
+
+    let test_and_log = async {
+        tokio::select! {
+            result = test => result,
+            _ = log => anyhow::bail!("Log task resolved unexpectedly"),
+        }
+    };
+
+    let result = with_ctrl_c(test_and_log, async {
         token.cancel();
         session.client().publish::<CancelTopic>(&()).await.unwrap();
     })
@@ -277,6 +376,7 @@ fn create_trial(
     session: &SessionInterface,
     path: &Path,
     rtt_client: Option<Key<RttClient>>,
+    sender: UnboundedSender<MonitorEvent>,
     token: &CancellationToken,
     test: Test,
 ) -> Trial {
@@ -287,35 +387,37 @@ fn create_trial(
     let session = session.clone();
     let token = token.clone();
 
-    Trial::test(name, move || {
-        if token.is_cancelled() {
-            eprintln!("Cancelled");
-            std::process::exit(0);
-        }
+    Trial::test(name, {
+        move || {
+            if token.is_cancelled() {
+                eprintln!("Cancelled");
+                std::process::exit(0);
+            }
 
-        let handle = tokio::spawn(async move {
-            match session
-                .run_test(test, rtt_client, print_monitor_event)
-                .await
-            {
-                Ok(TestResult::Success) => Ok(()),
-                Ok(TestResult::Cancelled) => {
-                    eprintln!("Cancelled");
-                    std::process::exit(0);
+            let handle = tokio::spawn(async move {
+                match session
+                    .run_test(test, rtt_client, |msg| sender.send(msg).unwrap())
+                    .await
+                {
+                    Ok(TestResult::Success) => Ok(()),
+                    Ok(TestResult::Cancelled) => {
+                        eprintln!("Cancelled");
+                        std::process::exit(0);
+                    }
+                    Ok(TestResult::Failed(message)) => {
+                        display_stack_trace(&session, &path).await?;
+
+                        Err(Failed::from(message))
+                    }
+                    Err(e) => {
+                        eprintln!("Error: {:?}", e);
+                        std::process::exit(1);
+                    }
                 }
-                Ok(TestResult::Failed(message)) => {
-                    display_stack_trace(&session, &path).await?;
+            });
 
-                    Err(Failed::from(message))
-                }
-                Err(e) => {
-                    eprintln!("Error: {:?}", e);
-                    std::process::exit(1);
-                }
-            }
-        });
-
-        Handle::current().block_on(handle).unwrap()
+            Handle::current().block_on(handle).unwrap()
+        }
     })
     .with_ignored_flag(ignored)
 }
@@ -333,18 +435,6 @@ async fn display_stack_trace(session: &SessionInterface, path: &Path) -> anyhow:
     Ok(())
 }
 
-fn print_monitor_event(event: MonitorEvent) {
-    match event {
-        MonitorEvent::RttOutput(str) => print!("{}", str),
-        MonitorEvent::SemihostingOutput(SemihostingOutput::StdOut(str)) => {
-            print!("{}", str)
-        }
-        MonitorEvent::SemihostingOutput(SemihostingOutput::StdErr(str)) => {
-            eprint!("{}", str)
-        }
-    }
-}
-
 /// Runs a future until complation, running another future when Ctrl+C is received.
 ///
 /// This function enables cooperative asynchronous cancellation without dropping the future.
@@ -366,3 +456,43 @@ where
 
     r
 }
+
+pub struct CliRttClient {
+    handle: Key<RttClient>,
+    data_format: RttDecoder,
+}
+
+impl CliRttClient {
+    pub fn handle(&self) -> Key<RttClient> {
+        self.handle
+    }
+}
+
+fn print_monitor_event(
+    rtt_client: &mut Option<impl DerefMut<Target = CliRttClient>>,
+    event: MonitorEvent,
+) {
+    match event {
+        MonitorEvent::RttOutput { channel, bytes } => {
+            if let Some(client) = rtt_client {
+                _ = client.data_format.process(channel, &bytes, &mut Printer);
+            }
+        }
+        MonitorEvent::SemihostingOutput(SemihostingOutput::StdOut(str)) => {
+            print!("{}", str)
+        }
+        MonitorEvent::SemihostingOutput(SemihostingOutput::StdErr(str)) => {
+            eprint!("{}", str)
+        }
+    }
+}
+
+struct Printer;
+impl RttDataHandler for Printer {
+    fn on_string_data(&mut self, channel: u32, data: String) -> Result<(), probe_rs::rtt::Error> {
+        if channel == 0 {
+            print!("{}", data);
+        }
+        Ok(())
+    }
+}
diff --git a/probe-rs-tools/src/bin/probe-rs/util/rtt.rs b/probe-rs-tools/src/bin/probe-rs/util/rtt.rs
index d182e736..4d0a591a 100644
--- a/probe-rs-tools/src/bin/probe-rs/util/rtt.rs
+++ b/probe-rs-tools/src/bin/probe-rs/util/rtt.rs
@@ -1,23 +1,21 @@
-use anyhow::{anyhow, Context};
-use defmt_decoder::log::format::{Formatter, FormatterConfig, FormatterFormat};
-use defmt_decoder::DecodeError;
-pub use probe_rs::rtt::ChannelMode;
-use probe_rs::rtt::{DownChannel, Error, Rtt, UpChannel};
+use postcard_schema::Schema;
+use probe_rs::rtt::{self, DownChannel, Error, Rtt, UpChannel};
 use probe_rs::{Core, MemoryInterface};
 use serde::{Deserialize, Serialize};
-use std::path::PathBuf;
-use std::sync::Arc;
-use std::{fmt, fmt::Write, path::Path};
-use time::{macros::format_description, OffsetDateTime, UtcOffset};
 
 pub(crate) mod client;
+pub(crate) mod processing;
+
+pub use processing::*;
 
 /// Used by serde to provide defaults for `RttChannelConfig::show_timestamps`
 fn default_show_timestamps() -> bool {
     true
 }
 
-#[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize, Deserialize, Default, docsplay::Display)]
+#[derive(
+    Debug, Copy, Clone, PartialEq, Eq, Default, docsplay::Display, Serialize, Deserialize, Schema,
+)]
 pub enum DataFormat {
     #[default]
     /// string
@@ -28,8 +26,45 @@ pub enum DataFormat {
     Defmt,
 }
 
+/// Specifies what to do when a channel doesn't have enough buffer space for a complete write on the
+/// target side.
+#[derive(Clone, Copy, Eq, PartialEq, Debug, Serialize, Deserialize, Schema)]
+#[repr(u32)]
+pub enum ChannelMode {
+    /// Skip writing the data completely if it doesn't fit in its entirety.
+    NoBlockSkip = 0,
+
+    /// Write as much as possible of the data and ignore the rest.
+    NoBlockTrim = 1,
+
+    /// Block (spin) if the buffer is full. Note that if the application writes within a critical
+    /// section, using this mode can cause the application to freeze if the buffer becomes full and
+    /// is not read by the host.
+    BlockIfFull = 2,
+}
+
+impl From<ChannelMode> for rtt::ChannelMode {
+    fn from(mode: ChannelMode) -> Self {
+        match mode {
+            ChannelMode::NoBlockSkip => rtt::ChannelMode::NoBlockSkip,
+            ChannelMode::NoBlockTrim => rtt::ChannelMode::NoBlockTrim,
+            ChannelMode::BlockIfFull => rtt::ChannelMode::BlockIfFull,
+        }
+    }
+}
+
+impl From<rtt::ChannelMode> for ChannelMode {
+    fn from(mode: rtt::ChannelMode) -> Self {
+        match mode {
+            rtt::ChannelMode::NoBlockSkip => ChannelMode::NoBlockSkip,
+            rtt::ChannelMode::NoBlockTrim => ChannelMode::NoBlockTrim,
+            rtt::ChannelMode::BlockIfFull => ChannelMode::BlockIfFull,
+        }
+    }
+}
+
 /// The initial configuration for RTT (Real Time Transfer). This configuration is complimented with the additional information specified for each of the channels in `RttChannel`.
-#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+#[derive(Debug, Clone, Serialize, Deserialize, Default, Schema)]
 pub struct RttConfig {
     #[serde(default, rename = "rttEnabled")]
     pub enabled: bool,
@@ -41,7 +76,7 @@ pub struct RttConfig {
 
 impl RttConfig {
     /// Returns the configuration for the specified channel number, if it exists.
-    pub fn channel_config(&self, channel_number: usize) -> Option<&RttChannelConfig> {
+    pub fn channel_config(&self, channel_number: u32) -> Option<&RttChannelConfig> {
         self.channels
             .iter()
             .find(|ch| ch.channel_number == Some(channel_number))
@@ -49,10 +84,10 @@ impl RttConfig {
 }
 
 /// The User specified configuration for each active RTT Channel.
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize, Schema)]
 #[serde(rename_all = "camelCase")]
 pub struct RttChannelConfig {
-    pub channel_number: Option<usize>,
+    pub channel_number: Option<u32>,
     #[serde(default)]
     pub data_format: DataFormat,
 
@@ -86,315 +121,55 @@ impl Default for RttChannelConfig {
     }
 }
 
-pub enum ChannelDataFormat {
-    String {
-        /// UTC offset used for creating timestamps, if enabled.
-        ///
-        /// Getting the offset can fail in multi-threaded programs,
-        /// so it needs to be stored.
-        timestamp_offset: Option<UtcOffset>,
-        last_line_done: bool,
-    },
-    BinaryLE,
-    Defmt {
-        formatter: Formatter,
-        // CWD to strip from file paths in defmt output
-        cwd: PathBuf,
-        defmt_data: Option<Arc<DefmtState>>,
-    },
-}
-
-impl From<&ChannelDataFormat> for DataFormat {
-    fn from(config: &ChannelDataFormat) -> Self {
-        match config {
-            ChannelDataFormat::String { .. } => DataFormat::String,
-            ChannelDataFormat::BinaryLE => DataFormat::BinaryLE,
-            ChannelDataFormat::Defmt { .. } => DataFormat::Defmt,
-        }
-    }
-}
-
-impl fmt::Debug for ChannelDataFormat {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match self {
-            ChannelDataFormat::String {
-                timestamp_offset,
-                last_line_done,
-            } => f
-                .debug_struct("String")
-                .field("timestamp_offset", timestamp_offset)
-                .field("last_line_done", last_line_done)
-                .finish(),
-            ChannelDataFormat::BinaryLE => f.debug_struct("BinaryLE").finish(),
-            ChannelDataFormat::Defmt { .. } => f.debug_struct("Defmt").finish_non_exhaustive(),
-        }
-    }
-}
-
-impl ChannelDataFormat {
-    /// Returns whether the channel is expected to output binary data (`true`)
-    /// or human-readable strings (`false`).
-    pub fn is_binary(&self) -> bool {
-        matches!(self, ChannelDataFormat::BinaryLE)
-    }
-
-    fn process(
-        &mut self,
-        number: usize,
-        buffer: &[u8],
-        collector: &mut impl ChannelDataCallbacks,
-    ) -> Result<(), Error> {
-        // FIXME: clean this up by splitting the enum variants out into separate structs
-        match self {
-            ChannelDataFormat::BinaryLE => collector.on_binary_data(number, buffer),
-            ChannelDataFormat::String {
-                timestamp_offset,
-                ref mut last_line_done,
-            } => {
-                let string = Self::process_string(buffer, *timestamp_offset, last_line_done)?;
-                collector.on_string_data(number, string)
-            }
-            ChannelDataFormat::Defmt {
-                ref formatter,
-                ref cwd,
-                ref defmt_data,
-            } => {
-                let string = Self::process_defmt(buffer, defmt_data.as_deref(), formatter, cwd)?;
-                collector.on_string_data(number, string)
-            }
-        }
-    }
-
-    fn process_string(
-        buffer: &[u8],
-        offset: Option<UtcOffset>,
-        last_line_done: &mut bool,
-    ) -> Result<String, Error> {
-        let incoming = String::from_utf8_lossy(buffer);
-
-        let Some(offset) = offset else {
-            return Ok(incoming.to_string());
-        };
-
-        let timestamp = OffsetDateTime::now_utc()
-            .to_offset(offset)
-            .format(format_description!(
-                "[hour repr:24]:[minute]:[second].[subsecond digits:3]"
-            ))
-            .expect("Incorrect format string. This shouldn't happen.");
-
-        let mut formatted_data = String::new();
-        for line in incoming.split_inclusive('\n') {
-            if *last_line_done {
-                write!(formatted_data, "{timestamp}: ").expect("Writing to String cannot fail");
-            }
-            write!(formatted_data, "{line}").expect("Writing to String cannot fail");
-            *last_line_done = line.ends_with('\n');
-        }
-        Ok(formatted_data)
-    }
-
-    fn process_defmt(
-        buffer: &[u8],
-        defmt_state: Option<&DefmtState>,
-        formatter: &Formatter,
-        cwd: &Path,
-    ) -> Result<String, Error> {
-        let Some(DefmtState { table, locs }) = defmt_state else {
-            return Ok(String::from(
-                "Trying to process defmt data but table or locations could not be loaded.\n",
-            ));
-        };
-
-        let mut stream_decoder = table.new_stream_decoder();
-
-        // FIXME: this assumes we read frames atomically which is implementation-defined and we
-        // should be able to handle the case where a frame is split across two reads with a
-        // temporary buffer.
-        stream_decoder.received(buffer);
-
-        let mut formatted_data = String::new();
-        loop {
-            match stream_decoder.decode() {
-                Ok(frame) => {
-                    let loc = locs.as_ref().and_then(|locs| locs.get(&frame.index()));
-                    let (file, line, module) = if let Some(loc) = loc {
-                        let relpath = loc.file.strip_prefix(cwd).unwrap_or(&loc.file);
-                        (
-                            relpath.display().to_string(),
-                            Some(loc.line.try_into().unwrap()),
-                            Some(loc.module.as_str()),
-                        )
-                    } else {
-                        (
-                            format!(
-                                "└─ <invalid location: defmt frame-index: {}>",
-                                frame.index()
-                            ),
-                            None,
-                            None,
-                        )
-                    };
-                    let s = formatter.format_frame(frame, Some(&file), line, module);
-                    writeln!(formatted_data, "{s}").expect("Writing to String cannot fail");
-                }
-                Err(DecodeError::UnexpectedEof) => break,
-                Err(DecodeError::Malformed) if table.encoding().can_recover() => {
-                    // If recovery is possible, skip the current frame and continue with new data.
-                }
-                Err(DecodeError::Malformed) => {
-                    return Err(Error::Other(anyhow!(
-                        "Unrecoverable error while decoding Defmt \
-                        data. Some data may have been lost: {}",
-                        DecodeError::Malformed
-                    )));
-                }
-            }
-        }
-
-        Ok(formatted_data)
-    }
-}
-
-pub trait ChannelDataCallbacks {
-    fn on_binary_data(&mut self, channel: usize, data: &[u8]) -> Result<(), Error> {
-        let mut formatted_data = String::with_capacity(data.len() * 4);
-        for element in data {
-            // Width of 4 allows 0xFF to be printed.
-            write!(&mut formatted_data, "{element:#04x}").expect("Writing to String cannot fail");
-        }
-        self.on_string_data(channel, formatted_data)
-    }
-
-    fn on_string_data(&mut self, channel: usize, data: String) -> Result<(), Error>;
-}
-
 #[derive(Debug)]
 pub struct RttActiveUpChannel {
     pub up_channel: UpChannel,
-    pub data_format: ChannelDataFormat,
+
     rtt_buffer: Box<[u8]>,
+    bytes_buffered: usize,
 
-    /// If set, the original mode of the channel before we changed it. Upon exit we should do
+    /// If set, the original mode of the channel before we first changed it. Upon exit we should do
     /// our best to restore the original mode.
-    original_mode: Option<ChannelMode>,
+    original_mode: Option<rtt::ChannelMode>,
 }
 
 impl RttActiveUpChannel {
-    pub fn new(
-        core: &mut Core,
-        up_channel: UpChannel,
-        channel_config: &RttChannelConfig,
-        timestamp_offset: UtcOffset,
-        defmt_data: Option<Arc<DefmtState>>,
-    ) -> Result<Self, Error> {
-        let is_defmt_channel = up_channel.name() == Some("defmt");
-
-        let data_format = match channel_config.data_format {
-            DataFormat::String if !is_defmt_channel => ChannelDataFormat::String {
-                timestamp_offset: channel_config.show_timestamps.then_some(timestamp_offset),
-                last_line_done: true,
-            },
-
-            DataFormat::BinaryLE if !is_defmt_channel => ChannelDataFormat::BinaryLE,
-
-            // either DataFormat::Defmt is configured, or defmt_enabled is true
-            _ => {
-                let has_timestamp = if let Some(ref defmt) = defmt_data {
-                    defmt.table.has_timestamp()
-                } else {
-                    tracing::warn!("No `Table` definition in DWARF info; compile your program with `debug = 2` to enable location info.");
-                    false
-                };
-
-                // Format options:
-                // 1. Oneline format with optional location
-                // 2. Custom format for the channel
-                // 3. Default with optional location
-                let format = match channel_config.log_format.as_deref() {
-                    Some("oneline") => FormatterFormat::OneLine {
-                        with_location: channel_config.show_location,
-                    },
-                    Some(format) => FormatterFormat::Custom(format),
-                    None => FormatterFormat::Default {
-                        with_location: channel_config.show_location,
-                    },
-                };
-
-                ChannelDataFormat::Defmt {
-                    formatter: Formatter::new(FormatterConfig {
-                        format,
-                        is_timestamp_available: has_timestamp && channel_config.show_timestamps,
-                    }),
-                    cwd: std::env::current_dir().unwrap(),
-                    defmt_data,
-                }
-            }
-        };
-
-        let mut original_mode = None;
-        if let Some(mode) = channel_config.mode.or(
-            // Try not to corrupt the byte stream if using defmt
-            if matches!(data_format, ChannelDataFormat::Defmt { .. }) {
-                Some(ChannelMode::BlockIfFull)
-            } else {
-                None
-            },
-        ) {
-            original_mode = Some(up_channel.mode(core)?);
-            up_channel.set_mode(core, mode)?;
-        }
-
-        Ok(Self {
+    pub fn new(up_channel: UpChannel) -> Self {
+        Self {
             rtt_buffer: vec![0; up_channel.buffer_size().max(1)].into_boxed_slice(),
+            bytes_buffered: 0,
             up_channel,
-            data_format,
-            original_mode,
-        })
+            original_mode: None,
+        }
+    }
+
+    pub fn change_mode(&mut self, core: &mut Core, mode: ChannelMode) -> Result<(), Error> {
+        if self.original_mode.is_none() {
+            self.original_mode = Some(self.up_channel.mode(core)?);
+        }
+        self.up_channel.set_mode(core, mode.into())
     }
 
     pub fn channel_name(&self) -> String {
         self.up_channel
             .name()
             .map(ToString::to_string)
-            .unwrap_or_else(|| {
-                format!(
-                    "Unnamed {} RTT up channel - {}",
-                    DataFormat::from(&self.data_format),
-                    self.up_channel.number()
-                )
-            })
+            .unwrap_or_else(|| format!("Unnamed RTT up channel - {}", self.up_channel.number()))
     }
 
-    pub fn number(&self) -> usize {
-        self.up_channel.number()
+    pub fn number(&self) -> u32 {
+        self.up_channel.number() as u32
     }
 
-    /// Polls the RTT target for new data on the channel represented by `self`.
-    /// Processes all the new data into the channel internal buffer and
-    /// returns the number of bytes that was read.
-    pub fn poll_rtt(&mut self, core: &mut Core) -> Result<Option<usize>, Error> {
-        match self.up_channel.read(core, self.rtt_buffer.as_mut())? {
-            0 => Ok(None),
-            count => Ok(Some(count)),
-        }
+    /// Reads available channel data into the internal buffer.
+    pub fn poll(&mut self, core: &mut Core) -> Result<(), Error> {
+        self.bytes_buffered = self.up_channel.read(core, self.rtt_buffer.as_mut())?;
+        Ok(())
     }
 
-    /// Retrieves available data from the channel and if available, returns `Some(channel_number:String, formatted_data:String)`.
-    /// If no data is available, or we encounter a recoverable error, it returns `None` value for `formatted_data`.
-    /// Non-recoverable errors are propagated to the caller.
-    pub fn poll_process_rtt_data(
-        &mut self,
-        core: &mut Core,
-        collector: &mut impl ChannelDataCallbacks,
-    ) -> Result<(), Error> {
-        let Some(bytes_read) = self.poll_rtt(core)? else {
-            return Ok(());
-        };
-
-        let buffer = &self.rtt_buffer[..bytes_read];
-
-        self.data_format.process(self.number(), buffer, collector)
+    /// Returns the buffered data.
+    pub fn buffered_data(&self) -> &[u8] {
+        &self.rtt_buffer[..self.bytes_buffered]
     }
 
     /// Clean up temporary changes made to the channel.
@@ -423,60 +198,15 @@ impl RttActiveDownChannel {
             .unwrap_or_else(|| format!("Unnamed RTT down channel - {}", self.down_channel.number()))
     }
 
-    pub fn number(&self) -> usize {
-        self.down_channel.number()
+    pub fn number(&self) -> u32 {
+        self.down_channel.number() as u32
     }
 
-    pub fn push_rtt(&mut self, core: &mut Core<'_>, data: impl AsRef<[u8]>) -> Result<(), Error> {
+    pub fn write(&mut self, core: &mut Core<'_>, data: impl AsRef<[u8]>) -> Result<(), Error> {
         self.down_channel.write(core, data.as_ref()).map(|_| ())
     }
 }
 
-/// Once an active connection with the Target RTT control block has been established, we configure
-/// each of the active channels, and hold essential state information for successful communication.
-#[derive(Debug)]
-pub struct RttActiveTarget {
-    control_block_addr: u64,
-    pub active_up_channels: Vec<RttActiveUpChannel>,
-    pub active_down_channels: Vec<RttActiveDownChannel>,
-}
-
-/// defmt information common to all defmt channels.
-pub struct DefmtState {
-    pub table: defmt_decoder::Table,
-    pub locs: Option<defmt_decoder::Locations>,
-}
-impl DefmtState {
-    pub fn try_from_bytes(buffer: &[u8]) -> Result<Option<Self>, Error> {
-        let Some(table) =
-            defmt_decoder::Table::parse(buffer).with_context(|| "Failed to parse defmt data")?
-        else {
-            return Ok(None);
-        };
-
-        let locs = table
-            .get_locations(buffer)
-            .with_context(|| "Failed to parse defmt data")?;
-
-        let locs = if !table.is_empty() && locs.is_empty() {
-            tracing::warn!("Insufficient DWARF info; compile your program with `debug = 2` to enable location info.");
-            None
-        } else if table.indices().all(|idx| locs.contains_key(&(idx as u64))) {
-            Some(locs)
-        } else {
-            tracing::warn!("Location info is incomplete; it will be omitted from the output.");
-            None
-        };
-        Ok(Some(DefmtState { table, locs }))
-    }
-}
-
-impl fmt::Debug for DefmtState {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("DefmtState").finish_non_exhaustive()
-    }
-}
-
 /// Error type for RTT symbol lookup.
 #[derive(Debug, thiserror::Error, docsplay::Display)]
 pub enum RttSymbolError {
@@ -486,15 +216,18 @@ pub enum RttSymbolError {
     Goblin(#[source] goblin::error::Error),
 }
 
-impl RttActiveTarget {
+/// Once an active connection with the Target RTT control block has been established, we configure
+/// each of the active channels, and hold essential state information for successful communication.
+#[derive(Debug)]
+pub struct RttConnection {
+    control_block_addr: u64,
+    pub active_up_channels: Vec<RttActiveUpChannel>,
+    pub active_down_channels: Vec<RttActiveDownChannel>,
+}
+
+impl RttConnection {
     /// RttActiveTarget collects references to all the `RttActiveChannel`s, for latter polling/pushing of data.
-    pub fn new(
-        core: &mut Core,
-        rtt: Rtt,
-        defmt_state: Option<Arc<DefmtState>>,
-        rtt_config: &RttConfig,
-        timestamp_offset: UtcOffset,
-    ) -> Result<Self, Error> {
+    pub fn new(core: &mut Core, rtt: Rtt, rtt_config: &RttConfig) -> Result<Self, Error> {
         let control_block_addr = rtt.ptr();
         let mut active_up_channels = Vec::with_capacity(rtt.up_channels.len());
 
@@ -502,16 +235,16 @@ impl RttActiveTarget {
         // additional user configuration in a `RttChannelConfig`. If not, apply defaults.
         for channel in rtt.up_channels.into_iter() {
             let channel_config = rtt_config
-                .channel_config(channel.number())
+                .channel_config(channel.number() as u32)
                 .cloned()
                 .unwrap_or_default();
-            active_up_channels.push(RttActiveUpChannel::new(
-                core,
-                channel,
-                &channel_config,
-                timestamp_offset,
-                defmt_state.clone(),
-            )?);
+
+            let mut up_channel = RttActiveUpChannel::new(channel);
+            if let Some(mode) = channel_config.mode {
+                up_channel.change_mode(core, mode)?;
+            }
+
+            active_up_channels.push(up_channel);
         }
 
         let active_down_channels = rtt
@@ -527,46 +260,23 @@ impl RttActiveTarget {
         })
     }
 
-    pub fn get_rtt_symbol_from_bytes(buffer: &[u8]) -> Result<u64, RttSymbolError> {
-        match goblin::elf::Elf::parse(buffer) {
-            Ok(binary) => {
-                for sym in &binary.syms {
-                    if binary.strtab.get_at(sym.st_name) == Some("_SEGGER_RTT") {
-                        return Ok(sym.st_value);
-                    }
-                }
-                Err(RttSymbolError::RttSymbolNotFound)
-            }
-            Err(err) => Err(RttSymbolError::Goblin(err)),
-        }
-    }
-
     /// Polls the RTT target on all channels and returns available data.
     /// An error on any channel will return an error instead of incomplete data.
-    pub fn poll_rtt_fallible(
-        &mut self,
-        core: &mut Core,
-        collector: &mut impl ChannelDataCallbacks,
-    ) -> Result<(), Error> {
-        for channel in self.active_up_channels.iter_mut() {
-            channel.poll_process_rtt_data(core, collector)?;
+    pub fn poll_channel(&mut self, core: &mut Core, channel_idx: u32) -> Result<(), Error> {
+        let channel_idx = channel_idx as usize;
+        if let Some(channel) = self.active_up_channels.get_mut(channel_idx) {
+            channel.poll(core)
+        } else {
+            Err(Error::MissingChannel(channel_idx))
         }
-        Ok(())
     }
 
-    /// Polls the RTT target on all channels and returns available data.
-    /// An error on any channel will return an error instead of incomplete data.
-    pub fn poll_channel_fallible(
-        &mut self,
-        core: &mut Core,
-        channel: usize,
-        collector: &mut impl ChannelDataCallbacks,
-    ) -> Result<(), Error> {
-        if let Some(channel) = self.active_up_channels.get_mut(channel) {
-            channel.poll_process_rtt_data(core, collector)?;
-            Ok(())
+    pub fn channel_data(&self, channel_idx: u32) -> Result<&[u8], Error> {
+        let channel_idx = channel_idx as usize;
+        if let Some(channel) = self.active_up_channels.get(channel_idx) {
+            Ok(channel.buffered_data())
         } else {
-            Err(Error::MissingChannel(channel))
+            Err(Error::MissingChannel(channel_idx))
         }
     }
 
@@ -574,13 +284,14 @@ impl RttActiveTarget {
     pub fn write_down_channel(
         &mut self,
         core: &mut Core,
-        channel: usize,
+        channel_idx: u32,
         data: impl AsRef<[u8]>,
     ) -> Result<(), Error> {
-        if let Some(channel) = self.active_down_channels.get_mut(channel) {
-            channel.push_rtt(core, data)
+        let channel_idx = channel_idx as usize;
+        if let Some(channel) = self.active_down_channels.get_mut(channel_idx) {
+            channel.write(core, data)
         } else {
-            Err(Error::MissingChannel(channel))
+            Err(Error::MissingChannel(channel_idx))
         }
     }
 
@@ -601,3 +312,17 @@ impl RttActiveTarget {
         Ok(())
     }
 }
+
+pub fn get_rtt_symbol_from_bytes(buffer: &[u8]) -> Result<u64, RttSymbolError> {
+    match goblin::elf::Elf::parse(buffer) {
+        Ok(binary) => {
+            for sym in &binary.syms {
+                if binary.strtab.get_at(sym.st_name) == Some("_SEGGER_RTT") {
+                    return Ok(sym.st_value);
+                }
+            }
+            Err(RttSymbolError::RttSymbolNotFound)
+        }
+        Err(err) => Err(RttSymbolError::Goblin(err)),
+    }
+}
diff --git a/probe-rs-tools/src/bin/probe-rs/util/rtt/client.rs b/probe-rs-tools/src/bin/probe-rs/util/rtt/client.rs
index cf6d97ad..8813bef0 100644
--- a/probe-rs-tools/src/bin/probe-rs/util/rtt/client.rs
+++ b/probe-rs-tools/src/bin/probe-rs/util/rtt/client.rs
@@ -1,23 +1,16 @@
-use std::sync::Arc;
-
-use crate::util::rtt::{
-    ChannelDataCallbacks, DefmtState, RttActiveDownChannel, RttActiveTarget, RttActiveUpChannel,
-    RttConfig, RttSymbolError,
-};
+use crate::util::rtt::{RttActiveDownChannel, RttActiveUpChannel, RttConfig, RttConnection};
 use probe_rs::{
     rtt::{Error, Rtt, ScanRegion},
-    Core, Target,
+    Core, MemoryInterface,
 };
-use time::UtcOffset;
 
 pub struct RttClient {
-    pub defmt_data: Option<Arc<DefmtState>>,
     pub scan_region: ScanRegion,
-    pub timezone_offset: UtcOffset,
-    rtt_config: RttConfig,
+    pub config: RttConfig,
 
     /// The internal RTT handle, if we have successfully attached to the target.
-    target: Option<RttActiveTarget>,
+    target: Option<RttConnection>,
+    last_control_block_address: Option<u64>,
 
     /// If false, don't try to attach to the target.
     try_attaching: bool,
@@ -31,47 +24,16 @@ pub struct RttClient {
 }
 
 impl RttClient {
-    pub fn new(
-        elf: Option<&[u8]>,
-        target: &Target,
-        rtt_config: RttConfig,
-        scan_region: ScanRegion,
-    ) -> Result<Self, Error> {
-        let mut this = Self {
-            defmt_data: None,
+    pub fn new(config: RttConfig, scan_region: ScanRegion) -> Self {
+        Self {
             scan_region,
-            rtt_config,
+            config,
             target: None,
+            last_control_block_address: None,
             try_attaching: true,
-            timezone_offset: UtcOffset::UTC,
             polled_data: false,
             core_id: 0,
-        };
-
-        if let Some(elf) = elf {
-            let mut init_defmt = false;
-            match RttActiveTarget::get_rtt_symbol_from_bytes(elf) {
-                Ok(address) => {
-                    this.scan_region = ScanRegion::Exact(address);
-                    this.core_id = target.core_index_by_address(address).unwrap_or(0);
-
-                    init_defmt = true;
-                }
-                Err(RttSymbolError::Goblin(_)) => {
-                    // Not an ELF
-                }
-                Err(RttSymbolError::RttSymbolNotFound) => {
-                    // We can still try to use defmt, we might find the control block.
-                    init_defmt = true;
-                }
-            }
-
-            if init_defmt {
-                this.defmt_data = DefmtState::try_from_bytes(elf)?.map(Arc::new);
-            }
         }
-
-        Ok(this)
     }
 
     pub fn try_attach(&mut self, core: &mut Core) -> Result<bool, Error> {
@@ -83,85 +45,89 @@ impl RttClient {
             return Ok(false);
         }
 
-        match Rtt::attach_region(core, &self.scan_region).and_then(|rtt| {
-            RttActiveTarget::new(
-                core,
-                rtt,
-                self.defmt_data.clone(),
-                &self.rtt_config,
-                self.timezone_offset,
-            )
-            .map(Some)
-        }) {
-            Ok(rtt) => self.target = rtt,
-            Err(Error::ControlBlockNotFound) => {}
+        let location = if let Some(location) = self.last_control_block_address {
+            location
+        } else {
+            let location = match Rtt::find_contol_block(core, &self.scan_region) {
+                Ok(location) => location,
+                Err(Error::ControlBlockNotFound) => {
+                    tracing::debug!("Failed to attach - control block not found");
+                    return Ok(false);
+                }
+                Err(Error::NoControlBlockLocation) => {
+                    tracing::debug!("Failed to attach - control block location not specified");
+                    self.try_attaching = false;
+                    return Ok(false);
+                }
+                Err(error) => return Err(error),
+            };
+
+            self.last_control_block_address = Some(location);
+            location
+        };
+
+        let rtt = match Rtt::attach_at(core, location) {
+            Ok(rtt) => rtt,
+            Err(Error::ControlBlockNotFound) => {
+                self.last_control_block_address = None;
+                tracing::debug!("Failed to attach - control block not found");
+                return Ok(false);
+            }
             Err(Error::ControlBlockCorrupted(error)) => {
-                tracing::debug!("RTT control block corrupted ({error})");
+                tracing::debug!("Failed to attach - control block corrupted: {}", error);
+                return Ok(false);
             }
-            Err(Error::NoControlBlockLocation) => self.try_attaching = false,
             Err(error) => return Err(error),
         };
 
-        Ok(self.target.is_some())
-    }
-
-    pub fn poll(
-        &mut self,
-        core: &mut Core,
-        collector: &mut impl ChannelDataCallbacks,
-    ) -> Result<(), Error> {
-        self.try_attach(core)?;
-
-        let Some(target) = self.target.as_mut() else {
-            return Ok(());
+        match RttConnection::new(core, rtt, &self.config) {
+            Ok(rtt) => self.target = Some(rtt),
+            Err(Error::ControlBlockCorrupted(error)) => {
+                tracing::debug!("Failed to attach - control block corrupted: {}", error);
+            }
+            Err(error) => return Err(error),
         };
 
-        let result = target.poll_rtt_fallible(core, collector);
-        self.handle_poll_result(result)
+        Ok(self.target.is_some())
     }
 
-    pub fn poll_channel(
-        &mut self,
-        core: &mut Core,
-        channel: usize,
-        collector: &mut impl ChannelDataCallbacks,
-    ) -> Result<(), Error> {
+    pub fn poll_channel(&mut self, core: &mut Core, channel: u32) -> Result<&[u8], Error> {
         self.try_attach(core)?;
 
-        let Some(target) = self.target.as_mut() else {
-            return Ok(());
-        };
+        if let Some(ref mut target) = self.target {
+            match target.poll_channel(core, channel) {
+                Ok(()) => self.polled_data = true,
 
-        let result = target.poll_channel_fallible(core, channel, collector);
-        self.handle_poll_result(result)
-    }
-
-    fn handle_poll_result(&mut self, result: Result<(), Error>) -> Result<(), Error> {
-        match result {
-            Ok(()) => self.polled_data = true,
-            Err(Error::ControlBlockCorrupted(error)) => {
-                if self.polled_data {
-                    tracing::warn!("RTT control block corrupted ({error}), re-attaching");
+                Err(Error::ControlBlockCorrupted(error)) => {
+                    if self.polled_data {
+                        tracing::warn!("RTT control block corrupted ({error}), re-attaching");
+                    }
+                    self.target = None;
+                    self.polled_data = false;
                 }
-                self.target = None;
-                self.polled_data = false;
-            }
-            Err(Error::ReadPointerChanged) => {
-                if self.polled_data {
-                    tracing::warn!("RTT read pointer changed, re-attaching");
+                Err(Error::ReadPointerChanged) => {
+                    if self.polled_data {
+                        tracing::warn!("RTT read pointer changed, re-attaching");
+                    }
+                    self.target = None;
+                    self.polled_data = false;
                 }
-                self.target = None;
-                self.polled_data = false;
+
+                Err(other) => return Err(other),
             }
-            other => return other,
         }
-        Ok(())
+
+        if let Some(ref target) = self.target {
+            return target.channel_data(channel);
+        }
+
+        Ok(&[])
     }
 
     pub(crate) fn write_down_channel(
         &mut self,
         core: &mut Core,
-        channel: usize,
+        channel: u32,
         input: impl AsRef<[u8]>,
     ) -> Result<(), Error> {
         self.try_attach(core)?;
@@ -181,18 +147,47 @@ impl RttClient {
         Ok(())
     }
 
+    /// This function prevents probe-rs from attaching to an RTT control block that is not
+    /// supposed to be valid. This is useful when probe-rs has reset the MCU before attaching,
+    /// or during/after flashing, when the MCU has not yet been started.
     pub(crate) fn clear_control_block(&mut self, core: &mut Core) -> Result<(), Error> {
         self.try_attach(core)?;
 
-        let Some(target) = self.target.as_mut() else {
-            // If we can't attach, we don't have a valid
-            // control block and don't have to do anything.
-            return Ok(());
-        };
-
-        target.clear_control_block(core)?;
+        if let Some(mut target) = self.target.take() {
+            target.clear_control_block(core)?;
+        } else {
+            // While the entire block isn't valid in itself, some parts of it may be.
+            // Depending on the firmware, the control block may be initialized in such
+            // an order where probe-rs can attach to it before it is fully valid.
+            if let Some(location) = self.last_control_block_address.take() {
+                if let ScanRegion::Exact(scan_location) = self.scan_region {
+                    // If we know the exact location where a control block should be, we can clear
+                    // the whole block.
+                    if location == scan_location {
+                        const SIZE_32B: usize = 16 + 2 * 4;
+                        const SIZE_64B: usize = 16 + 2 * 8;
+
+                        let zeros = [0; SIZE_64B];
+
+                        if core.is_64_bit() {
+                            core.write_8(location, &zeros[..SIZE_64B])?;
+                        } else {
+                            core.write_8(location, &zeros[..SIZE_32B])?;
+                        }
+                    }
+                } else {
+                    // If we have to scan for the location or we somehow found the magic string
+                    // somewhere else, we can only clear the magic string.
+                    let mut magic = [0; Rtt::RTT_ID.len()];
+                    core.read_8(location, &mut magic)?;
+                    if magic == Rtt::RTT_ID {
+                        core.write_8(location, &[0; 16])?;
+                    }
+                }
+            }
 
-        self.target = None;
+            // There's nothing we can do if we don't know where the control block is.
+        }
 
         Ok(())
     }
diff --git a/probe-rs-tools/src/bin/probe-rs/util/rtt/processing.rs b/probe-rs-tools/src/bin/probe-rs/util/rtt/processing.rs
new file mode 100644
index 00000000..c513547d
--- /dev/null
+++ b/probe-rs-tools/src/bin/probe-rs/util/rtt/processing.rs
@@ -0,0 +1,283 @@
+use anyhow::{anyhow, Context};
+use defmt_decoder::{
+    log::format::{Formatter, FormatterConfig, FormatterFormat},
+    DecodeError, StreamDecoder,
+};
+use probe_rs::rtt::Error;
+use time::{macros::format_description, OffsetDateTime, UtcOffset};
+
+use std::{
+    fmt::{self, Write},
+    sync::Arc,
+};
+
+use crate::util::rtt::DataFormat;
+
+pub enum RttDecoder {
+    String {
+        /// UTC offset used for creating timestamps, if enabled.
+        ///
+        /// Getting the offset can fail in multi-threaded programs,
+        /// so it needs to be stored.
+        timestamp_offset: Option<UtcOffset>,
+        last_line_done: bool,
+    },
+    BinaryLE,
+    Defmt {
+        processor: DefmtProcessor,
+    },
+}
+
+impl From<&RttDecoder> for DataFormat {
+    fn from(config: &RttDecoder) -> Self {
+        match config {
+            RttDecoder::String { .. } => DataFormat::String,
+            RttDecoder::BinaryLE => DataFormat::BinaryLE,
+            RttDecoder::Defmt { .. } => DataFormat::Defmt,
+        }
+    }
+}
+
+impl fmt::Debug for RttDecoder {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            RttDecoder::String {
+                timestamp_offset,
+                last_line_done,
+            } => f
+                .debug_struct("String")
+                .field("timestamp_offset", timestamp_offset)
+                .field("last_line_done", last_line_done)
+                .finish(),
+            RttDecoder::BinaryLE => f.debug_struct("BinaryLE").finish(),
+            RttDecoder::Defmt { .. } => f.debug_struct("Defmt").finish_non_exhaustive(),
+        }
+    }
+}
+
+impl RttDecoder {
+    /// Returns whether the channel is expected to output binary data (`true`)
+    /// or human-readable strings (`false`).
+    pub fn is_binary(&self) -> bool {
+        matches!(self, RttDecoder::BinaryLE)
+    }
+
+    pub fn process(
+        &mut self,
+        number: u32,
+        buffer: &[u8],
+        collector: &mut impl RttDataHandler,
+    ) -> Result<(), Error> {
+        // FIXME: clean this up by splitting the enum variants out into separate structs
+        match self {
+            RttDecoder::BinaryLE => collector.on_binary_data(number, buffer),
+            RttDecoder::String {
+                timestamp_offset,
+                ref mut last_line_done,
+            } => {
+                let string = Self::process_string(buffer, *timestamp_offset, last_line_done)?;
+                collector.on_string_data(number, string)
+            }
+            RttDecoder::Defmt { processor } => {
+                let string = processor.process(buffer)?;
+                collector.on_string_data(number, string)
+            }
+        }
+    }
+
+    fn process_string(
+        buffer: &[u8],
+        offset: Option<UtcOffset>,
+        last_line_done: &mut bool,
+    ) -> Result<String, Error> {
+        let incoming = String::from_utf8_lossy(buffer);
+
+        let Some(offset) = offset else {
+            return Ok(incoming.to_string());
+        };
+
+        let timestamp = OffsetDateTime::now_utc()
+            .to_offset(offset)
+            .format(format_description!(
+                "[hour repr:24]:[minute]:[second].[subsecond digits:3]"
+            ))
+            .expect("Incorrect format string. This shouldn't happen.");
+
+        let mut formatted_data = String::new();
+        for line in incoming.split_inclusive('\n') {
+            if *last_line_done {
+                write!(formatted_data, "{timestamp}: ").expect("Writing to String cannot fail");
+            }
+            write!(formatted_data, "{line}").expect("Writing to String cannot fail");
+            *last_line_done = line.ends_with('\n');
+        }
+        Ok(formatted_data)
+    }
+}
+
+pub trait RttDataHandler {
+    fn on_binary_data(&mut self, channel: u32, data: &[u8]) -> Result<(), Error> {
+        let mut formatted_data = String::with_capacity(data.len() * 4);
+        for element in data {
+            // Width of 4 allows 0xFF to be printed.
+            write!(&mut formatted_data, "{element:#04x}").expect("Writing to String cannot fail");
+        }
+        self.on_string_data(channel, formatted_data)
+    }
+
+    fn on_string_data(&mut self, channel: u32, data: String) -> Result<(), Error>;
+}
+
+pub struct DefmtStateInner {
+    pub table: defmt_decoder::Table,
+    pub locs: Option<defmt_decoder::Locations>,
+}
+
+impl DefmtStateInner {
+    pub fn try_from_bytes(buffer: &[u8]) -> Result<Option<Self>, Error> {
+        let Some(table) =
+            defmt_decoder::Table::parse(buffer).with_context(|| "Failed to parse defmt data")?
+        else {
+            return Ok(None);
+        };
+
+        let locs = table
+            .get_locations(buffer)
+            .with_context(|| "Failed to parse defmt data")?;
+
+        let locs = if !table.is_empty() && locs.is_empty() {
+            tracing::warn!("Insufficient DWARF info; compile your program with `debug = 2` to enable location info.");
+            None
+        } else if table.indices().all(|idx| locs.contains_key(&(idx as u64))) {
+            Some(locs)
+        } else {
+            tracing::warn!("Location info is incomplete; it will be omitted from the output.");
+            None
+        };
+        Ok(Some(DefmtStateInner { table, locs }))
+    }
+}
+
+/// defmt information common to all defmt channels.
+#[derive(Clone)]
+pub struct DefmtState {
+    inner: Arc<DefmtStateInner>,
+}
+impl DefmtState {
+    pub fn try_from_bytes(buffer: &[u8]) -> Result<Option<Self>, Error> {
+        Ok(
+            DefmtStateInner::try_from_bytes(buffer)?.map(|inner| DefmtState {
+                inner: Arc::new(inner),
+            }),
+        )
+    }
+
+    fn as_ref(&self) -> &DefmtStateInner {
+        &self.inner
+    }
+
+    pub fn table(&self) -> &defmt_decoder::Table {
+        &self.inner.table
+    }
+}
+
+impl fmt::Debug for DefmtState {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("DefmtState").finish_non_exhaustive()
+    }
+}
+
+pub struct DefmtProcessor {
+    formatter: Formatter,
+    // Fields are dropped in declaration order. `decoder` is holding a reference to defmt_data's
+    // inner table, so it must be dropped first.
+    decoder: Box<dyn StreamDecoder>,
+    defmt_data: DefmtState,
+}
+
+impl DefmtProcessor {
+    pub fn new(
+        defmt_data: DefmtState,
+        show_timestamps: bool,
+        show_location: bool,
+        log_format: Option<&str>,
+    ) -> Self {
+        let has_timestamp = defmt_data.table().has_timestamp();
+
+        // Format options:
+        // 1. Oneline format with optional location
+        // 2. Custom format for the channel
+        // 3. Default with optional location
+        let format = match log_format {
+            None | Some("oneline") => FormatterFormat::OneLine {
+                with_location: show_location,
+            },
+            Some("full") => FormatterFormat::Default {
+                with_location: show_location,
+            },
+            Some(format) => FormatterFormat::Custom(format),
+        };
+
+        Self {
+            formatter: Formatter::new(FormatterConfig {
+                format,
+                is_timestamp_available: has_timestamp && show_timestamps,
+            }),
+            decoder: unsafe {
+                // Extend lifetime to 'static. We can do this because we hold a reference to the
+                // defmt_data's inner table for the lifetime of the processor.
+                std::mem::transmute::<Box<dyn StreamDecoder>, Box<dyn StreamDecoder + 'static>>(
+                    defmt_data.as_ref().table.new_stream_decoder(),
+                )
+            },
+            defmt_data: defmt_data.clone(),
+        }
+    }
+
+    fn process(&mut self, buffer: &[u8]) -> Result<String, Error> {
+        let DefmtStateInner { table, locs } = self.defmt_data.as_ref();
+        self.decoder.received(buffer);
+
+        let mut formatted_data = String::new();
+        loop {
+            match self.decoder.decode() {
+                Ok(frame) => {
+                    let loc = locs.as_ref().and_then(|locs| locs.get(&frame.index()));
+                    let (file, line, module) = if let Some(loc) = loc {
+                        (
+                            loc.file.display().to_string(),
+                            Some(loc.line.try_into().unwrap()),
+                            Some(loc.module.as_str()),
+                        )
+                    } else {
+                        (
+                            format!(
+                                "└─ <invalid location: defmt frame-index: {}>",
+                                frame.index()
+                            ),
+                            None,
+                            None,
+                        )
+                    };
+                    let s = self
+                        .formatter
+                        .format_frame(frame, Some(&file), line, module);
+                    writeln!(formatted_data, "{s}").expect("Writing to String cannot fail");
+                }
+                Err(DecodeError::UnexpectedEof) => break,
+                Err(DecodeError::Malformed) if table.encoding().can_recover() => {
+                    // If recovery is possible, skip the current frame and continue with new data.
+                }
+                Err(DecodeError::Malformed) => {
+                    return Err(Error::Other(anyhow!(
+                        "Unrecoverable error while decoding Defmt \
+                        data. Some data may have been lost: {}",
+                        DecodeError::Malformed
+                    )));
+                }
+            }
+        }
+
+        Ok(formatted_data)
+    }
+}
diff --git a/probe-rs/src/rtt.rs b/probe-rs/src/rtt.rs
index 680f8b32..a653c1bc 100644
--- a/probe-rs/src/rtt.rs
+++ b/probe-rs/src/rtt.rs
@@ -51,8 +51,8 @@ use crate::Session;
 use crate::{config::MemoryRegion, Core, MemoryInterface};
 use std::ops::Range;
 use std::thread;
+use std::time::Duration;
 use std::time::Instant;
-use std::{borrow::Cow, time::Duration};
 use zerocopy::FromBytes;
 
 /// The RTT interface.
@@ -147,7 +147,7 @@ impl RttControlBlockHeader {
         }
     }
 
-    pub fn minimal_header_size(is_64_bit: bool) -> usize {
+    pub const fn minimal_header_size(is_64_bit: bool) -> usize {
         if is_64_bit {
             std::mem::size_of::<RttControlBlockHeaderInner<u64>>()
         } else {
@@ -231,27 +231,19 @@ impl RttControlBlockHeader {
 //     RttChannel down_channels[max_down_channels]; // array of down (host to target) channels.
 // }
 impl Rtt {
-    const RTT_ID: [u8; 16] = *b"SEGGER RTT\0\0\0\0\0\0";
+    /// The magic string expected to be found at the beginning of the RTT control block.
+    pub const RTT_ID: [u8; 16] = *b"SEGGER RTT\0\0\0\0\0\0";
 
-    fn from(
+    /// Tries to attach to an RTT control block at the specified memory address.
+    pub fn attach_at(
         core: &mut Core,
         // Pointer from which to scan
         ptr: u64,
-        // Memory contents read in advance, starting from ptr
-        mem_in: Option<&[u8]>,
-    ) -> Result<Option<Rtt>, Error> {
+    ) -> Result<Rtt, Error> {
         let is_64_bit = core.is_64_bit();
 
-        let mut mem = match mem_in {
-            Some(mem) => Cow::Borrowed(mem),
-            None => {
-                // If memory wasn't passed in, read the minimum header size
-                let new_length = RttControlBlockHeader::minimal_header_size(is_64_bit);
-                let mut mem = vec![0; new_length];
-                core.read(ptr, &mut mem)?;
-                Cow::Owned(mem)
-            }
-        };
+        let mut mem = [0u8; RttControlBlockHeader::minimal_header_size(true)];
+        core.read(ptr, &mut mem)?;
 
         let rtt_header = RttControlBlockHeader::try_from_header(is_64_bit, &mem)
             .ok_or(Error::ControlBlockNotFound)?;
@@ -277,29 +269,17 @@ impl Rtt {
             )));
         }
 
-        let cb_len = rtt_header.total_rtt_buffer_size();
-
-        if let Cow::Owned(mem) = &mut mem {
-            // If memory wasn't passed in, read the rest of the control block
-            mem.resize(cb_len, 0);
-            core.read(
-                ptr + rtt_header.header_size() as u64,
-                &mut mem[rtt_header.header_size()..cb_len],
-            )?;
-        }
-
-        // Validate that the entire control block fits within the region
-        if mem.len() < cb_len {
-            tracing::debug!("Control block doesn't fit in scanned memory region.");
-            return Ok(None);
-        }
+        // Read the rest of the control block
+        let channel_buffer_len = rtt_header.total_rtt_buffer_size() - rtt_header.header_size();
+        let mut mem = vec![0; channel_buffer_len];
+        core.read(ptr + rtt_header.header_size() as u64, &mut mem)?;
 
         let mut up_channels = Vec::new();
         let mut down_channels = Vec::new();
 
         let channel_buffer_size = rtt_header.channel_buffer_size();
 
-        let up_channels_start = rtt_header.header_size();
+        let up_channels_start = 0;
         let up_channels_len = max_up_channels * channel_buffer_size;
         let up_channels_raw_buffer = &mem[up_channels_start..][..up_channels_len];
         let up_channels_buffer = rtt_header.parse_channel_buffers(up_channels_raw_buffer)?;
@@ -309,54 +289,58 @@ impl Rtt {
         let down_channels_raw_buffer = &mem[down_channels_start..][..down_channels_len];
         let down_channels_buffer = rtt_header.parse_channel_buffers(down_channels_raw_buffer)?;
 
-        let mut offset = up_channels_start as u64;
-        for (i, b) in up_channels_buffer.into_iter().enumerate() {
-            let buffer_size = b.size() as u64;
+        let mut offset = ptr + rtt_header.header_size() as u64 + up_channels_start as u64;
+        for (channel_index, buffer) in up_channels_buffer.into_iter().enumerate() {
+            let buffer_size = buffer.size() as u64;
 
-            if let Some(chan) = Channel::from(core, i, ptr + offset, b)? {
+            if let Some(chan) = Channel::from(core, channel_index, offset, buffer)? {
                 up_channels.push(UpChannel(chan));
             } else {
-                tracing::warn!("Buffer for up channel {i} not initialized");
+                tracing::warn!("Buffer for up channel {channel_index} not initialized");
             }
             offset += buffer_size;
         }
 
-        for (i, b) in down_channels_buffer.into_iter().enumerate() {
-            let buffer_size = b.size() as u64;
+        let mut offset = ptr + rtt_header.header_size() as u64 + down_channels_start as u64;
+        for (channel_index, buffer) in down_channels_buffer.into_iter().enumerate() {
+            let buffer_size = buffer.size() as u64;
 
-            if let Some(chan) = Channel::from(core, i, ptr + offset, b)? {
+            if let Some(chan) = Channel::from(core, channel_index, offset, buffer)? {
                 down_channels.push(DownChannel(chan));
             } else {
-                tracing::warn!("Buffer for down channel {i} not initialized");
+                tracing::warn!("Buffer for down channel {channel_index} not initialized");
             }
             offset += buffer_size;
         }
 
-        Ok(Some(Rtt {
+        Ok(Rtt {
             ptr,
             up_channels,
             down_channels,
-        }))
+        })
+    }
+
+    /// Attempts to detect an RTT control block in the specified RAM region(s) and returns an
+    /// instance if a valid control block was found.
+    pub fn attach_region(core: &mut Core, region: &ScanRegion) -> Result<Rtt, Error> {
+        let ptr = Self::find_contol_block(core, region)?;
+        Self::attach_at(core, ptr)
     }
 
     /// Attempts to detect an RTT control block anywhere in the target RAM and returns an instance
     /// if a valid control block was found.
-    ///
-    /// `core` can be e.g. an owned `Core` or a shared `Rc<Core>`.
     pub fn attach(core: &mut Core) -> Result<Rtt, Error> {
         Self::attach_region(core, &ScanRegion::default())
     }
 
     /// Attempts to detect an RTT control block in the specified RAM region(s) and returns an
-    /// instance if a valid control block was found.
-    ///
-    /// `core` can be e.g. an owned `Core` or a shared `Rc<Core>`.
-    pub fn attach_region(core: &mut Core, region: &ScanRegion) -> Result<Rtt, Error> {
+    /// address if a valid control block location was found.
+    pub fn find_contol_block(core: &mut Core, region: &ScanRegion) -> Result<u64, Error> {
         let ranges = match region.clone() {
             ScanRegion::Exact(addr) => {
                 tracing::debug!("Scanning at exact address: {:#010x}", addr);
 
-                return Rtt::from(core, addr, None)?.ok_or(Error::ControlBlockNotFound);
+                return Ok(addr);
             }
             ScanRegion::Ram => {
                 tracing::debug!("Scanning whole RAM");
@@ -399,9 +383,9 @@ impl Rtt {
 
                 let target_ptr = range.start + offset as u64;
 
-                Rtt::from(core, target_ptr, Some(&mem[offset..])).transpose()
+                Some(target_ptr)
             })
-            .collect::<Result<Vec<_>, _>>()?;
+            .collect::<Vec<_>>();
 
         match instances.len() {
             0 => Err(Error::ControlBlockNotFound),
@@ -484,7 +468,7 @@ pub enum Error {
     ControlBlockNotFound,
 
     /// Multiple control blocks found in target memory: {display_list(_0)}.
-    MultipleControlBlocksFound(Vec<Rtt>),
+    MultipleControlBlocksFound(Vec<u64>),
 
     /// The control block has been corrupted: {0}
     ControlBlockCorrupted(String),
@@ -508,9 +492,9 @@ pub enum Error {
     MissingChannel(usize),
 }
 
-fn display_list(list: &[Rtt]) -> String {
+fn display_list(list: &[u64]) -> String {
     list.iter()
-        .map(|rtt| format!("{:#010x}", rtt.ptr))
+        .map(|ptr| format!("{:#010x}", ptr))
         .collect::<Vec<_>>()
         .join(", ")
 }
@@ -568,15 +552,7 @@ mod test {
 
     #[test]
     fn test_how_control_block_list_looks() {
-        fn rtt(ptr: u32) -> Rtt {
-            Rtt {
-                ptr: ptr.into(),
-                up_channels: Vec::new(),
-                down_channels: Vec::new(),
-            }
-        }
-
-        let error = Error::MultipleControlBlocksFound(vec![rtt(0x2000), rtt(0x3000)]);
+        let error = Error::MultipleControlBlocksFound(vec![0x2000, 0x3000]);
         assert_eq!(
             error.to_string(),
             "Multiple control blocks found in target memory: 0x00002000, 0x00003000."
diff --git a/probe-rs/src/rtt/channel.rs b/probe-rs/src/rtt/channel.rs
index e7c3ef0b..d1dbca9d 100644
--- a/probe-rs/src/rtt/channel.rs
+++ b/probe-rs/src/rtt/channel.rs
@@ -3,6 +3,7 @@ use crate::{Core, MemoryInterface};
 use probe_rs_target::RegionMergeIterator;
 use std::cmp::min;
 use std::ffi::CStr;
+use std::num::NonZeroU64;
 use zerocopy::{FromBytes, Immutable, KnownLayout};
 
 /// Trait for channel information shared between up and down channels.
@@ -82,10 +83,10 @@ impl RttChannelBuffer {
         }
     }
 
-    pub fn standard_name_pointer(&self) -> u64 {
+    pub fn standard_name_pointer(&self) -> Option<NonZeroU64> {
         match self {
-            RttChannelBuffer::Buffer32(x) => u64::from(x.standard_name_pointer),
-            RttChannelBuffer::Buffer64(x) => x.standard_name_pointer,
+            RttChannelBuffer::Buffer32(x) => NonZeroU64::new(u64::from(x.standard_name_pointer)),
+            RttChannelBuffer::Buffer64(x) => NonZeroU64::new(x.standard_name_pointer),
         }
     }
 
@@ -226,7 +227,11 @@ impl Channel {
         // This should at least catch most cases where the control block is partially initialized.
         this.read_pointers(core, "")?;
         // Read channel name just after the pointer was validated to be within an expected range.
-        this.name = read_c_string(core, this.info.standard_name_pointer())?;
+        this.name = if let Some(ptr) = this.info.standard_name_pointer() {
+            read_c_string(core, ptr)?
+        } else {
+            None
+        };
         this.mode(core)?;
 
         Ok(Some(this))
@@ -283,7 +288,7 @@ impl Channel {
 
             if buffer_offset_larger_than_size_of_buffer {
                 return Err(Error::ControlBlockCorrupted(format!(
-                    "{which} pointer is {value} while buffer size is {} for {channel_kind}channel {} ({})",
+                    "{which} pointer is {value:#010x} while buffer size is {:#010x} for {channel_kind}channel {} ({})",
                     self.info.size_of_buffer(),
                     self.number,
                     self.name().unwrap_or("no name"),
@@ -535,21 +540,17 @@ impl RttChannel for DownChannel {
 }
 
 /// Reads a null-terminated string from target memory. Lossy UTF-8 decoding is used.
-fn read_c_string(core: &mut Core, ptr: u64) -> Result<Option<String>, Error> {
-    // Find out which memory range contains the pointer
-    if ptr == 0 {
-        // If the pointer is null, return None.
-        return Ok(None);
-    }
-
+fn read_c_string(core: &mut Core, ptr: NonZeroU64) -> Result<Option<String>, Error> {
+    let ptr = ptr.get();
     let Some(range) = core
         .memory_regions()
         .filter(|r| r.is_ram() || r.is_nvm())
         .find_map(|r| r.contains(ptr).then_some(r.address_range()))
     else {
-        // If the pointer is not within any valid range, return None.
-        tracing::warn!("RTT channel name points to unrecognized memory. Bad target description?");
-        return Ok(None);
+        return Err(Error::ControlBlockCorrupted(format!(
+            "The channel name pointer is not in a valid memory region: {:#X}",
+            ptr
+        )));
     };
 
     // Read up to 128 bytes not going past the end of the region


Test patch:
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 fde014b9..801cb1d3 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,18 +1,17 @@
 use std::time::Duration;
 
-use postcard_rpc::{
-    header::{VarHeader, VarSeq},
-    server::Sender,
-};
+use anyhow::Context;
+use postcard_rpc::{header::VarHeader, server::Sender};
 use postcard_schema::Schema;
 use probe_rs::{semihosting::SemihostingCommand, BreakpointCause, Core, HaltReason, Session};
 use serde::{Deserialize, Serialize};
+use tokio::sync::mpsc;
 
 use crate::{
     rpc::{
         functions::{
             flash::BootInfo,
-            monitor::{MonitorEvent, RttStreamer, SemihostingOutput, SemihostingReader},
+            monitor::{MonitorEvent, SemihostingOutput, SemihostingReader},
             ListTestsEndpoint, MonitorTopic, RpcResult, RpcSpawnContext, RunTestEndpoint,
             WireTxImpl,
         },
@@ -116,15 +115,15 @@ pub struct ListTestsRequest {
 pub type ListTestsResponse = RpcResult<Tests>;
 
 pub async fn list_tests(
-    ctx: RpcSpawnContext,
+    mut ctx: RpcSpawnContext,
     header: VarHeader,
     request: ListTestsRequest,
     sender: Sender<WireTxImpl>,
 ) {
-    let resp =
-        tokio::task::spawn_blocking(move || list_tests_impl(ctx, request).map_err(Into::into))
-            .await
-            .unwrap();
+    let resp = ctx
+        .run_blocking::<MonitorTopic, _, _, _>(request, list_tests_impl)
+        .await
+        .map_err(Into::into);
 
     sender
         .reply::<ListTestsEndpoint>(header.seq_no, &resp)
@@ -132,16 +131,13 @@ pub async fn list_tests(
         .unwrap();
 }
 
-fn list_tests_impl(ctx: RpcSpawnContext, request: ListTestsRequest) -> anyhow::Result<Tests> {
+fn list_tests_impl(
+    ctx: RpcSpawnContext,
+    request: ListTestsRequest,
+    sender: mpsc::Sender<MonitorEvent>,
+) -> anyhow::Result<Tests> {
     let mut session = ctx.session_blocking(request.sessid);
-    let mut list_handler = ListEventHandler::new({
-        let ctx = ctx.clone();
-        move |event| {
-            ctx.publish_blocking::<MonitorTopic>(VarSeq::Seq2(0), event)
-                .unwrap();
-        }
-    });
-    let mut rtt_sink = RttStreamer::new(ctx.clone());
+    let mut list_handler = ListEventHandler::new(|event| sender.blocking_send(event).unwrap());
 
     let mut rtt_client = request
         .rtt_client
@@ -149,12 +145,6 @@ fn list_tests_impl(ctx: RpcSpawnContext, request: ListTestsRequest) -> anyhow::R
 
     let core_id = rtt_client.as_ref().map(|rtt| rtt.core_id()).unwrap_or(0);
 
-    let mut run_loop = RunLoop {
-        core_id,
-        rtt_client: rtt_client.as_deref_mut(),
-        cancellation_token: ctx.cancellation_token(),
-    };
-
     match request.boot_info {
         BootInfo::FromRam {
             vector_table_addr, ..
@@ -167,15 +157,28 @@ fn list_tests_impl(ctx: RpcSpawnContext, request: ListTestsRequest) -> anyhow::R
             session
                 .core(core_id)?
                 .reset_and_halt(Duration::from_millis(100))?;
+            if let Some(rtt_client) = rtt_client.as_mut() {
+                rtt_client.clear_control_block(&mut session.core(core_id)?)?;
+            }
         }
     }
 
+    let mut run_loop = RunLoop {
+        core_id,
+        rtt_client: rtt_client.as_deref_mut(),
+        cancellation_token: ctx.cancellation_token(),
+    };
+
     let mut core = session.core(0)?;
     match run_loop.run_until(
         &mut core,
         true,
         true,
-        &mut rtt_sink,
+        |channel, bytes| {
+            sender
+                .blocking_send(MonitorEvent::RttOutput { channel, bytes })
+                .with_context(|| "Failed to send RTT output")
+        },
         Some(Duration::from_secs(5)),
         |halt_reason, core| list_handler.handle_halt(halt_reason, core),
     )? {
@@ -201,15 +204,15 @@ pub struct RunTestRequest {
 pub type RunTestResponse = RpcResult<TestResult>;
 
 pub async fn run_test(
-    ctx: RpcSpawnContext,
+    mut ctx: RpcSpawnContext,
     header: VarHeader,
     request: RunTestRequest,
     sender: Sender<WireTxImpl>,
 ) {
-    let resp: RunTestResponse =
-        tokio::task::spawn_blocking(move || run_test_impl(ctx, request).map_err(Into::into))
-            .await
-            .unwrap();
+    let resp = ctx
+        .run_blocking::<MonitorTopic, _, _, _>(request, run_test_impl)
+        .await
+        .map_err(Into::into);
 
     sender
         .reply::<RunTestEndpoint>(header.seq_no, &resp)
@@ -217,7 +220,11 @@ pub async fn run_test(
         .unwrap();
 }
 
-fn run_test_impl(ctx: RpcSpawnContext, request: RunTestRequest) -> anyhow::Result<TestResult> {
+fn run_test_impl(
+    ctx: RpcSpawnContext,
+    request: RunTestRequest,
+    sender: mpsc::Sender<MonitorEvent>,
+) -> anyhow::Result<TestResult> {
     tracing::info!("Running test {}", request.test.name);
 
     let timeout = request.test.timeout.map(|t| Duration::from_secs(t as u64));
@@ -233,15 +240,13 @@ fn run_test_impl(ctx: RpcSpawnContext, request: RunTestRequest) -> anyhow::Resul
     let mut core = session.core(core_id)?;
     core.reset_and_halt(Duration::from_millis(100))?;
 
+    if let Some(rtt_client) = rtt_client.as_mut() {
+        rtt_client.clear_control_block(&mut core)?;
+    }
+
     let expected_outcome = request.test.expected_outcome;
-    let mut run_handler = RunEventHandler::new(request.test, {
-        let ctx = ctx.clone();
-        move |event| {
-            ctx.publish_blocking::<MonitorTopic>(VarSeq::Seq2(0), event)
-                .unwrap();
-        }
-    });
-    let mut rtt_sink = RttStreamer::new(ctx.clone());
+    let mut run_handler =
+        RunEventHandler::new(request.test, |event| sender.blocking_send(event).unwrap());
 
     let mut run_loop = RunLoop {
         core_id,
@@ -253,7 +258,11 @@ fn run_test_impl(ctx: RpcSpawnContext, request: RunTestRequest) -> anyhow::Resul
         &mut core,
         true,
         true,
-        &mut rtt_sink,
+        |channel, bytes| {
+            sender
+                .blocking_send(MonitorEvent::RttOutput { channel, bytes })
+                .with_context(|| "Failed to send RTT output")
+        },
         Some(timeout),
         |halt_reason, core| run_handler.handle_halt(halt_reason, core),
     )? {