drtio: change async messages to sync

src/runtime/src/rtio_mgt.rs

@@ -44,6 +44,23 @@ pub mod drtio {
         unsafe { (csr::DRTIO[linkno].rx_up_read)() == 1 }
     }
 
+    async fn process_async_packets(packet: Packet) -> Option<Packet> {
+        // returns None if an async packet has been consumed
+        match packet {
+            Packet::DmaPlaybackStatus {
+                id,
+                destination,
+                error,
+                channel,
+                timestamp,
+            } => {
+                remote_dma::playback_done(id, destination, error, channel, timestamp).await;
+                None
+            }
+            other => Some(other),
+        }
+    }
+
     async fn recv_aux_timeout(linkno: u8, timeout: u64, timer: GlobalTimer) -> Result<Packet, &'static str> {
         if !link_rx_up(linkno).await {
             return Err("link went down");
@@ -66,22 +83,7 @@ pub mod drtio {
         }
         let _lock = aux_mutex.async_lock().await;
         drtioaux_async::send(linkno, request).await.unwrap();
-        loop {
+        Ok(recv_aux_timeout(linkno, 200, timer).await?)
-            let reply = recv_aux_timeout(linkno, 200, timer).await;
-            match reply {
-                Ok(Packet::DmaPlaybackStatus {
-                    id,
-                    destination,
-                    error,
-                    channel,
-                    timestamp,
-                }) => {
-                    remote_dma::playback_done(id, destination, error, channel, timestamp).await;
-                }
-                Ok(packet) => return Ok(packet),
-                Err(e) => return Err(e),
-            }
-        }
     }
 
     async fn drain_buffer(linkno: u8, draining_time: Milliseconds, timer: GlobalTimer) {
@@ -190,14 +192,11 @@ pub mod drtio {
     async fn process_unsolicited_aux(aux_mutex: &Rc<Mutex<bool>>, linkno: u8) {
         let _lock = aux_mutex.async_lock().await;
         match drtioaux_async::recv(linkno).await {
-            Ok(Some(Packet::DmaPlaybackStatus {
+            Ok(Some(packet)) => {
-                id,
+                if let Some(packet) = process_async_packets(packet).await {
-                destination,
+                    warn!("[LINK#{}] unsolicited aux packet: {:?}", linkno, packet);
-                error,
+                }
-                channel,
+            }
-                timestamp,
-            })) => remote_dma::playback_done(id, destination, error, channel, timestamp).await,
-            Ok(Some(packet)) => warn!("[LINK#{}] unsolicited aux packet: {:?}", linkno, packet),
             Ok(None) => (),
             Err(_) => warn!("[LINK#{}] aux packet error", linkno),
         }
@@ -266,6 +265,7 @@ pub mod drtio {
                 let linkno = hop - 1;
                 if destination_up(up_destinations, destination).await {
                     if up_links[linkno as usize] {
+                        loop {
                             let reply = aux_transact(
                                 aux_mutex,
                                 linkno,
@@ -278,7 +278,13 @@ pub mod drtio {
                             match reply {
                                 Ok(Packet::DestinationDownReply) => {
                                     destination_set_up(routing_table, up_destinations, destination, false).await;
-                                remote_dma::destination_changed(aux_mutex, routing_table, timer, destination, false)
+                                    remote_dma::destination_changed(
+                                        aux_mutex,
+                                        routing_table,
+                                        timer,
+                                        destination,
+                                        false,
+                                    )
                                     .await;
                                 }
                                 Ok(Packet::DestinationOkReply) => (),
@@ -309,9 +315,16 @@ pub mod drtio {
                                     );
                                     unsafe { SEEN_ASYNC_ERRORS |= ASYNC_ERROR_BUSY };
                                 }
-                            Ok(packet) => error!("[DEST#{}] received unexpected aux packet: {:?}", destination, packet),
+                                Ok(packet) => match process_async_packets(packet).await {
+                                    Some(packet) => {
+                                        error!("[DEST#{}] received unexpected aux packet: {:?}", destination, packet)
+                                    }
+                                    None => continue,
+                                },
                                 Err(e) => error!("[DEST#{}] communication failed ({})", destination, e),
                             }
+                            break;
+                        }
                     } else {
                         destination_set_up(routing_table, up_destinations, destination, false).await;
                         remote_dma::destination_changed(aux_mutex, routing_table, timer, destination, false).await;

src/satman/src/main.rs

@@ -124,6 +124,23 @@ fn process_aux_packet(
             #[cfg(not(has_drtio_routing))]
             let hop = 0;
 
+            if let Some(status) = dma_manager.check_state() {
+                info!(
+                    "playback done, error: {}, channel: {}, timestamp: {}",
+                    status.error, status.channel, status.timestamp
+                );
+                return drtioaux::send(
+                    0,
+                    &drtioaux::Packet::DmaPlaybackStatus {
+                        destination: *_rank,
+                        id: status.id,
+                        error: status.error,
+                        channel: status.channel,
+                        timestamp: status.timestamp,
+                    },
+                );
+            }
+
             if hop == 0 {
                 let errors;
                 unsafe {
@@ -739,24 +756,6 @@ pub extern "C" fn main_core0() -> i32 {
                     error!("aux packet error: {:?}", e);
                 }
             }
-            if let Some(status) = dma_manager.check_state() {
-                info!(
-                    "playback done, error: {}, channel: {}, timestamp: {}",
-                    status.error, status.channel, status.timestamp
-                );
-                if let Err(e) = drtioaux::send(
-                    0,
-                    &drtioaux::Packet::DmaPlaybackStatus {
-                        destination: rank,
-                        id: status.id,
-                        error: status.error,
-                        channel: status.channel,
-                        timestamp: status.timestamp,
-                    },
-                ) {
-                    error!("error sending DMA playback status: {:?}", e);
-                }
-            }
         }
 
         drtiosat_reset_phy(true);