From 4363cdf9fa13fe4575ce3ee5722af2d15d2335f1 Mon Sep 17 00:00:00 2001 From: mwojcik Date: Fri, 8 Dec 2023 15:48:31 +0800 Subject: [PATCH] master: make use of the async message ready flag --- .../firmware/libproto_artiq/drtioaux_proto.rs | 13 ++ artiq/firmware/runtime/kernel.rs | 8 +- artiq/firmware/runtime/rtio_mgt.rs | 116 +++++++----------- artiq/firmware/runtime/session.rs | 1 + artiq/firmware/satman/main.rs | 15 +-- artiq/firmware/satman/routing.rs | 7 +- 6 files changed, 70 insertions(+), 90 deletions(-) diff --git a/artiq/firmware/libproto_artiq/drtioaux_proto.rs b/artiq/firmware/libproto_artiq/drtioaux_proto.rs index 0013c8171..1157e864c 100644 --- a/artiq/firmware/libproto_artiq/drtioaux_proto.rs +++ b/artiq/firmware/libproto_artiq/drtioaux_proto.rs @@ -712,4 +712,17 @@ impl Packet { _ => None } } + + pub fn expects_response(&self) -> bool { + // returns true if the routable packet should elicit a response + // e.g. reply, ACK packets end a conversation, + // and firmware should not wait for response + match self { + Packet::DmaAddTraceReply { .. } | Packet::DmaRemoveTraceReply { .. } | + Packet::DmaPlaybackReply { .. } | Packet::SubkernelLoadRunReply { .. } | + Packet::SubkernelMessageAck { .. } | Packet::DmaPlaybackStatus { .. } | + Packet::SubkernelFinished { .. } => false, + _ => true + } + } } diff --git a/artiq/firmware/runtime/kernel.rs b/artiq/firmware/runtime/kernel.rs index 124b17f19..855648845 100644 --- a/artiq/firmware/runtime/kernel.rs +++ b/artiq/firmware/runtime/kernel.rs @@ -364,8 +364,11 @@ pub mod subkernel { { let _lock = subkernel_mutex.lock(io)?; match unsafe { SUBKERNELS.get(&id).unwrap().state } { - SubkernelState::Finished { .. } => return Err(Error::SubkernelFinished), + SubkernelState::Finished { status: FinishStatus::Ok } | SubkernelState::Running => (), + SubkernelState::Finished { + status: FinishStatus::CommLost, + } => return Err(Error::SubkernelFinished), _ => return Err(Error::IncorrectState) } } @@ -385,7 +388,8 @@ pub mod subkernel { } } match unsafe { SUBKERNELS.get(&id).unwrap().state } { - SubkernelState::Finished { .. } => return Ok(None), + SubkernelState::Finished { status: FinishStatus::CommLost } | + SubkernelState::Finished { status: FinishStatus::Exception(_) } => return Ok(None), _ => () } Err(()) diff --git a/artiq/firmware/runtime/rtio_mgt.rs b/artiq/firmware/runtime/rtio_mgt.rs index 75347da68..5a9263e55 100644 --- a/artiq/firmware/runtime/rtio_mgt.rs +++ b/artiq/firmware/runtime/rtio_mgt.rs @@ -78,6 +78,16 @@ pub mod drtio { } } + fn link_has_async_ready(linkno: u8) -> bool { + let linkno = linkno as usize; + let async_ready; + unsafe { + async_ready = (csr::DRTIO[linkno].async_messages_ready_read)() == 1; + (csr::DRTIO[linkno].async_messages_ready_write)(1); + } + async_ready + } + fn recv_aux_timeout(io: &Io, linkno: u8, timeout: u32) -> Result { let max_time = clock::get_ms() + timeout as u64; loop { @@ -117,6 +127,8 @@ pub mod drtio { drtioaux::send(linkno, &drtioaux::Packet::SubkernelMessageAck { destination: from } ).unwrap(); + // give the satellite some time to process the message + io.sleep(10).unwrap(); }, // routable packets drtioaux::Packet::DmaAddTraceRequest { destination, .. } | @@ -147,45 +159,9 @@ pub mod drtio { } } else { warn!("[LINK#{}] Error handling async packets ({})", linkno, reply.unwrap_err()); + return; } - else { - drtioaux::send(dest_link, &packet).unwrap(); - } - None - }} - } - match packet { - // packets to be consumed locally - drtioaux::Packet::DmaPlaybackStatus { id, destination: 0, error, channel, timestamp } => { - remote_dma::playback_done(io, ddma_mutex, id, 0, error, channel, timestamp); - None - }, - drtioaux::Packet::SubkernelFinished { id, destination: 0, with_exception, exception_src } => { - subkernel::subkernel_finished(io, subkernel_mutex, id, with_exception, exception_src); - None - }, - drtioaux::Packet::SubkernelMessage { id, source: from, destination: 0, status, length, data } => { - subkernel::message_handle_incoming(io, subkernel_mutex, id, status, length as usize, &data); - // acknowledge receiving part of the message - drtioaux::send(linkno, - &drtioaux::Packet::SubkernelMessageAck { destination: from } - ).unwrap(); - None - }, - // routable packets - drtioaux::Packet::DmaAddTraceRequest { destination, .. } => route_packet!(destination), - drtioaux::Packet::DmaAddTraceReply { destination, .. } => route_packet!(destination), - drtioaux::Packet::DmaRemoveTraceRequest { destination, .. } => route_packet!(destination), - drtioaux::Packet::DmaRemoveTraceReply { destination, .. } => route_packet!(destination), - drtioaux::Packet::DmaPlaybackRequest { destination, .. } => route_packet!(destination), - drtioaux::Packet::DmaPlaybackReply { destination, .. } => route_packet!(destination), - drtioaux::Packet::SubkernelLoadRunRequest { destination, .. } => route_packet!(destination), - drtioaux::Packet::SubkernelLoadRunReply { destination, .. } => route_packet!(destination), - drtioaux::Packet::SubkernelMessage { destination, .. } => route_packet!(destination), - drtioaux::Packet::SubkernelMessageAck { destination, .. } => route_packet!(destination), - drtioaux::Packet::DmaPlaybackStatus { destination, .. } => route_packet!(destination), - drtioaux::Packet::SubkernelFinished { destination, .. } => route_packet!(destination), - other => Some(other) + } } } @@ -358,44 +334,35 @@ pub mod drtio { let linkno = hop - 1; if destination_up(up_destinations, destination) { if up_links[linkno as usize] { - loop { - let reply = aux_transact(io, aux_mutex, linkno, - &drtioaux::Packet::DestinationStatusRequest { - destination: destination - }); - if let Ok(reply) = reply { - let reply = process_async_packets(io, ddma_mutex, subkernel_mutex, routing_table, linkno, reply); - match reply { - Some(drtioaux::Packet::DestinationDownReply) => { - destination_set_up(routing_table, up_destinations, destination, false); - remote_dma::destination_changed(io, aux_mutex, ddma_mutex, routing_table, destination, false); - subkernel::destination_changed(io, aux_mutex, subkernel_mutex, routing_table, destination, false); - } - Some(drtioaux::Packet::DestinationOkReply) => (), - Some(drtioaux::Packet::DestinationSequenceErrorReply { channel }) => { - error!("[DEST#{}] RTIO sequence error involving channel 0x{:04x}:{}", destination, channel, resolve_channel_name(channel as u32)); - unsafe { SEEN_ASYNC_ERRORS |= ASYNC_ERROR_SEQUENCE_ERROR }; - } - Some(drtioaux::Packet::DestinationCollisionReply { channel }) => { - error!("[DEST#{}] RTIO collision involving channel 0x{:04x}:{}", destination, channel, resolve_channel_name(channel as u32)); - unsafe { SEEN_ASYNC_ERRORS |= ASYNC_ERROR_COLLISION }; - } - Some(drtioaux::Packet::DestinationBusyReply { channel }) => { - error!("[DEST#{}] RTIO busy error involving channel 0x{:04x}:{}", destination, channel, resolve_channel_name(channel as u32)); - unsafe { SEEN_ASYNC_ERRORS |= ASYNC_ERROR_BUSY }; - } - Some(packet) => error!("[DEST#{}] received unexpected aux packet: {:?}", destination, packet), - None => { - // continue asking until we get Destination...Reply or error out - // wait a bit not to overwhelm the receiver causing gateway errors - io.sleep(10).unwrap(); - continue; - } + let reply = aux_transact(io, aux_mutex, linkno, + &drtioaux::Packet::DestinationStatusRequest { + destination: destination + }); + if let Ok(reply) = reply { + match reply { + drtioaux::Packet::DestinationDownReply => { + destination_set_up(routing_table, up_destinations, destination, false); + remote_dma::destination_changed(io, aux_mutex, ddma_mutex, routing_table, destination, false); + subkernel::destination_changed(io, aux_mutex, subkernel_mutex, routing_table, destination, false); } - } else { - error!("[DEST#{}] communication failed ({:?})", destination, reply.unwrap_err()); + drtioaux::Packet::DestinationOkReply => (), + drtioaux::Packet::DestinationSequenceErrorReply { channel } => { + error!("[DEST#{}] RTIO sequence error involving channel 0x{:04x}:{}", destination, channel, resolve_channel_name(channel as u32)); + unsafe { SEEN_ASYNC_ERRORS |= ASYNC_ERROR_SEQUENCE_ERROR }; + } + drtioaux::Packet::DestinationCollisionReply { channel } => { + error!("[DEST#{}] RTIO collision involving channel 0x{:04x}:{}", destination, channel, resolve_channel_name(channel as u32)); + unsafe { SEEN_ASYNC_ERRORS |= ASYNC_ERROR_COLLISION }; + } + drtioaux::Packet::DestinationBusyReply { channel } => { + error!("[DEST#{}] RTIO busy error involving channel 0x{:04x}:{}", destination, channel, resolve_channel_name(channel as u32)); + unsafe { SEEN_ASYNC_ERRORS |= ASYNC_ERROR_BUSY }; + } + packet => error!("[DEST#{}] received unexpected aux packet: {:?}", destination, packet), + } - break; + } else { + error!("[DEST#{}] communication failed ({:?})", destination, reply.unwrap_err()); } } else { destination_set_up(routing_table, up_destinations, destination, false); @@ -436,6 +403,7 @@ pub mod drtio { if up_links[linkno as usize] { /* link was previously up */ if link_rx_up(linkno) { + process_async_packets(&io, aux_mutex, ddma_mutex, subkernel_mutex, routing_table, linkno); process_unsolicited_aux(&io, aux_mutex, linkno); process_local_errors(linkno); } else { diff --git a/artiq/firmware/runtime/session.rs b/artiq/firmware/runtime/session.rs index 44b4bd0be..800c6d040 100644 --- a/artiq/firmware/runtime/session.rs +++ b/artiq/firmware/runtime/session.rs @@ -471,6 +471,7 @@ fn process_host_message(io: &Io, _aux_mutex: &Mutex, _ddma_mutex: &Mutex, _subke match subkernel::upload(io, _aux_mutex, _subkernel_mutex, _routing_table, _id) { Ok(_) => host_write(stream, host::Reply::LoadCompleted)?, Err(error) => { + subkernel::clear_subkernels(io, _subkernel_mutex)?; let mut description = String::new(); write!(&mut description, "{}", error).unwrap(); host_write(stream, host::Reply::LoadFailed(&description))? diff --git a/artiq/firmware/satman/main.rs b/artiq/firmware/satman/main.rs index 1a8e794a3..eec0bab8f 100644 --- a/artiq/firmware/satman/main.rs +++ b/artiq/firmware/satman/main.rs @@ -146,15 +146,6 @@ fn process_aux_packet(dmamgr: &mut DmaManager, analyzer: &mut Analyzer, kernelmg if hop == 0 { *self_destination = destination; - // async messages - if *rank == 1 { - // for now, master ignores the async_messages_ready packet - if let Some(packet) = router.get_upstream_packet() { - // pass any async or routed packets to master - // this does mean that DDMA/SK packets to master will "trickle down" to higher rank - return drtioaux::send(0, &packet) - } - } let errors; unsafe { errors = csr::drtiosat::rtio_error_read(); @@ -776,9 +767,9 @@ pub extern fn main() -> i32 { if let Some(status) = dma_manager.get_status() { info!("playback done, error: {}, channel: {}, timestamp: {}", status.error, status.channel, status.timestamp); router.route(drtioaux::Packet::DmaPlaybackStatus { - destination: status.source, id: status.id, error: status.error, - channel: status.channel, timestamp: status.timestamp - }, &routing_table, rank, destination) + source: destination, destination: status.source, id: status.id, + error: status.error, channel: status.channel, timestamp: status.timestamp + }, &routing_table, rank, destination); } kernelmgr.process_kern_requests(&mut router, &routing_table, rank, destination); diff --git a/artiq/firmware/satman/routing.rs b/artiq/firmware/satman/routing.rs index e129e4a54..118ad11af 100644 --- a/artiq/firmware/satman/routing.rs +++ b/artiq/firmware/satman/routing.rs @@ -101,8 +101,11 @@ impl Router { } pub fn get_upstream_packet(&mut self) -> Option { - self.upstream_notified = false; - self.upstream_queue.pop_front() + let packet = self.upstream_queue.pop_front(); + if packet.is_none() { + self.upstream_notified = false; + } + packet } pub fn get_downstream_packet(&mut self) -> Option<(usize, drtioaux::Packet)> {