htgazurex1212/artiq-zynq
1
0
Fork 0
forked from M-Labs/artiq-zynq
Code Pull Requests Activity

Compare commits

base: htgazurex1212:767b725db7dc57238926e178e77af2af88c56a57
Branches Tags
htgazurex1212:master htgazurex1212:HTGAzureX1212/nac3-malloc M-Labs:master M-Labs:release-9 M-Labs:release-8 M-Labs:drtio_tidy_up M-Labs:release-7 M-Labs:release-6
...
compare: htgazurex1212:78080eae2bbfe204345695d91d534f9269577190
Branches Tags
htgazurex1212:master htgazurex1212:HTGAzureX1212/nac3-malloc M-Labs:master M-Labs:release-9 M-Labs:release-8 M-Labs:drtio_tidy_up M-Labs:release-7 M-Labs:release-6

7 Commits
767b725db7 ... 78080eae2b

Author SHA1 Message Date
Sebastien Bourdeauducq
78080eae2b flake: add new required fields for buildPythonPackage 2025-07-31 15:17:07 +08:00
Simon Renblad
f1e79310ec update cargo lock 2025-07-31 15:07:04 +08:00
Simon Renblad
3a65d6c2a5 libksupport: wrap RTIO_DEVICE_MAP in OnceLock 2025-07-31 15:07:04 +08:00
Simon Renblad
734fd11ad6 libboard_artiq: wrap LOGGER in OnceLock 2025-07-31 15:07:04 +08:00
Sebastien Bourdeauducq
5843138a8e flake: update dependencies 2025-07-31 15:06:03 +08:00
ivan
1ab755838a ksupport: expose libc/compiler-rt strlen and bcmp 2025-07-30 18:17:09 +08:00
Simon Renblad
e21873d227 runtime: make aux_mutex static 2025-07-25 16:27:01 +08:00
13 changed files with 234 additions and 336 deletions
Expand all files Collapse all files

81
flake.lock generated
View File

@@ -12,11 +12,11 @@
"src-pythonparser": "src-pythonparser"
},
"locked": {
"lastModified": 1752484307,
"narHash": "sha256-KJ56ujYOMpOl487npaFSeu1e7gpsS+2V3ZpdQLZ1OOE=",
"lastModified": 1753945412,
"narHash": "sha256-TWMNAkzFWSg0MyR0MylG5Va9K+KnQ6wPiCc5UXxjJx8=",
"ref": "refs/heads/master",
"rev": "cb5727c1ee1c6f4a81f25ee62557c3957b1a622b",
"revCount": 9290,
"rev": "b2ef8fb3987fb4eb142f42bcad4f06c48433f9dc",
"revCount": 9304,
"type": "git",
"url": "https://github.com/m-labs/artiq.git"
},
@@ -38,11 +38,11 @@
]
},
"locked": {
"lastModified": 1748847424,
"narHash": "sha256-BXpxu3fowHV0u6DVTYn6efQ8X3PV6qC4TgiuCsSy3/0=",
"lastModified": 1753274984,
"narHash": "sha256-VZhmNafIj++z2cYLwCxWvnhOctqnIOAvA1/r+WEy+LI=",
"owner": "m-labs",
"repo": "artiq-comtools",
"rev": "5552bf9b5e6e150dafbaf5cd20bd68c82949ac9a",
"rev": "25f1f2aae1b7e38f1315e40e448211430c3924df",
"type": "github"
},
"original": {
@@ -51,6 +51,29 @@
"type": "github"
}
},
"fenix": {
"inputs": {
"nixpkgs": [
"artiq",
"naersk",
"nixpkgs"
],
"rust-analyzer-src": "rust-analyzer-src"
},
"locked": {
"lastModified": 1752475459,
"narHash": "sha256-z6QEu4ZFuHiqdOPbYss4/Q8B0BFhacR8ts6jO/F/aOU=",
"owner": "nix-community",
"repo": "fenix",
"rev": "bf0d6f70f4c9a9cf8845f992105652173f4b617f",
"type": "github"
},
"original": {
"owner": "nix-community",
"repo": "fenix",
"type": "github"
}
},
"flake-utils": {
"inputs": {
"systems": "systems"
@@ -71,17 +94,18 @@
},
"naersk": {
"inputs": {
"fenix": "fenix",
"nixpkgs": [
"artiq",
"nixpkgs"
]
},
"locked": {
"lastModified": 1745925850,
"narHash": "sha256-cyAAMal0aPrlb1NgzMxZqeN1mAJ2pJseDhm2m6Um8T0=",
"lastModified": 1752689277,
"narHash": "sha256-uldUBFkZe/E7qbvxa3mH1ItrWZyT6w1dBKJQF/3ZSsc=",
"owner": "nix-community",
"repo": "naersk",
"rev": "38bc60bbc157ae266d4a0c96671c6c742ee17a5f",
"rev": "0e72363d0938b0208d6c646d10649164c43f4d64",
"type": "github"
},
"original": {
@@ -113,11 +137,11 @@
},
"nixpkgs": {
"locked": {
"lastModified": 1750741721,
"narHash": "sha256-Z0djmTa1YmnGMfE9jEe05oO4zggjDmxOGKwt844bUhE=",
"lastModified": 1752950548,
"narHash": "sha256-NS6BLD0lxOrnCiEOcvQCDVPXafX1/ek1dfJHX1nUIzc=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "4b1164c3215f018c4442463a27689d973cffd750",
"rev": "c87b95e25065c028d31a94f06a62927d18763fdf",
"type": "github"
},
"original": {
@@ -133,6 +157,23 @@
"zynq-rs": "zynq-rs"
}
},
"rust-analyzer-src": {
"flake": false,
"locked": {
"lastModified": 1752428706,
"narHash": "sha256-EJcdxw3aXfP8Ex1Nm3s0awyH9egQvB2Gu+QEnJn2Sfg=",
"owner": "rust-lang",
"repo": "rust-analyzer",
"rev": "591e3b7624be97e4443ea7b5542c191311aa141d",
"type": "github"
},
"original": {
"owner": "rust-lang",
"ref": "nightly",
"repo": "rust-analyzer",
"type": "github"
}
},
"rust-overlay": {
"inputs": {
"nixpkgs": [
@@ -184,11 +225,11 @@
]
},
"locked": {
"lastModified": 1752168638,
"narHash": "sha256-wH+RiSaALa9Y4DVW2kDOTXUAjCmREIjq3oN9GuA6L70=",
"lastModified": 1753275053,
"narHash": "sha256-ukZqfmWV0zyMTZL3UK4gBZlIsEW/HdE0Kc6DO+Opoes=",
"owner": "m-labs",
"repo": "sipyco",
"rev": "01f2e37ea13cb7f297321104319428b1ddf546fc",
"rev": "7a0d0d7a95695cf18f1b60b8320b10a26df53f36",
"type": "github"
},
"original": {
@@ -272,11 +313,11 @@
"rust-overlay": "rust-overlay_2"
},
"locked": {
"lastModified": 1752558724,
"narHash": "sha256-bOrhgijeSPQoUDs6jy/lw40pJK9dv/jjcVeIQPyobdk=",
"lastModified": 1753867338,
"narHash": "sha256-Xw50x/Tx+Y/dbqpAb4M8Fu7DCeYW6psmWnT2UPl2e4E=",
"ref": "refs/heads/master",
"rev": "e3f2c44dd9f9eaa522ca06a493064fccb1a7a6f3",
"revCount": 722,
"rev": "b69de9f37fe6b16c0bf8d3d6bfcec7fe05a3ee03",
"revCount": 726,
"type": "git",
"url": "https://git.m-labs.hk/m-labs/zynq-rs"
},

10
flake.nix
View File

@@ -23,6 +23,8 @@
inherit pname version;
sha256 = "sha256-4JLTP4uVwxcaL7NOV57+DFSwKQ3X+W/6onYkN2AdkKc=";
};
pyproject = true;
build-system = [pkgs.python3Packages.setuptools];
};
artiq-netboot = pkgs.python3Packages.buildPythonPackage rec {
@@ -34,6 +36,8 @@
rev = "04f69eb07df73abe4b89fde2c24084f7664f2104";
sha256 = "0ql4fr8m8gpb2yql8aqsdqsssxb8zqd6l65kl1f6s9845zy7shs9";
};
pyproject = true;
build-system = [pkgs.python3Packages.setuptools];
};
ramda = pkgs.python3Packages.buildPythonPackage {
@@ -46,6 +50,8 @@
rev = "d315a9717ebd639366bf3fe26bad9e3d08ec3c49";
sha256 = "sha256-bmSt/IHDnULsZjsC6edELnNH7LoJSVF4L4XhwBAXRkY=";
};
pyproject = true;
build-system = [pkgs.python3Packages.setuptools];
nativeBuildInputs = with pkgs.python3Packages; [ pbr ];
propagatedBuildInputs = with pkgs.python3Packages; [ fastnumbers ];
@@ -69,8 +75,8 @@
rev = "98649a92ed7d4e43f75231e6ef9753e1212fab41";
sha256 = "sha256-0kEHK+l6gZW750tq89fHRxIh3Gnj5EP2GZX/neWaWzU=";
};
format = "pyproject";
pyproject = true;
build-system = [pkgs.python3Packages.setuptools];
propagatedBuildInputs = with pkgs.python3Packages; [ setuptools click numpy toolz jinja2 ramda artiqpkgs.migen artiqpkgs.misoc ];

2
src/Cargo.lock generated
View File

@@ -255,6 +255,7 @@ name = "libasync"
version = "0.0.0"
dependencies = [
"embedded-hal",
"libcortex_a9",
"nb 1.0.0",
"smoltcp",
]
@@ -311,6 +312,7 @@ dependencies = [
"core_io",
"fatfs",
"libboard_zynq",
"libcortex_a9",
"log",
]

19
src/libboard_artiq/src/logger.rs
View File

@@ -1,7 +1,8 @@
use core::{cell::Cell, fmt::Write, mem::MaybeUninit};
use core::{cell::Cell, fmt::Write};
use libboard_zynq::{println, timer};
use libcortex_a9::mutex::{Mutex, MutexGuard};
use libcortex_a9::{mutex::{Mutex, MutexGuard},
once_lock::OnceLock};
use log::{LevelFilter, Log};
use log_buffer::LogBuffer;
@@ -42,7 +43,7 @@ pub struct BufferLogger {
buffer_filter: Cell<LevelFilter>,
}
static mut LOGGER: MaybeUninit<BufferLogger> = MaybeUninit::uninit();
static LOGGER: OnceLock<BufferLogger> = OnceLock::new();
impl BufferLogger {
pub fn new(buffer: &'static mut [u8]) -> BufferLogger {
@@ -53,17 +54,13 @@ impl BufferLogger {
}
}
#[allow(static_mut_refs)]
pub fn register(self) {
unsafe {
LOGGER.write(self);
log::set_logger(LOGGER.assume_init_ref()).expect("global logger can only be initialized once");
}
LOGGER.set(self).expect("LOGGER can only be initialized once");
log::set_logger(LOGGER.get().unwrap()).expect("global logger can only be initialized once");
}
#[allow(static_mut_refs)]
pub fn get_logger() -> &'static mut BufferLogger {
unsafe { LOGGER.assume_init_mut() }
pub fn get_logger() -> &'static BufferLogger {
LOGGER.get().expect("cannot get logger before it is initialized")
}
pub fn buffer<'a>(&'a self) -> Option<LogBufferRef<'a>> {

12
src/libksupport/src/kernel/api.rs
View File

@@ -254,6 +254,18 @@ pub fn resolve(required: &[u8]) -> Option<u32> {
fn memcmp(s1: *const u8, s2: *const u8, n: usize) -> i32;
}
),
api!(
bcmp,
extern "C" {
fn bcmp(s1: *const u8, s2: *const u8, n: usize) -> i32;
}
),
api!(
strlen,
extern "C" {
fn strlen(s: *const core::ffi::c_char) -> usize;
}
),
// exceptions
api!(_Unwind_Resume = unwind::_Unwind_Resume),

21
src/libksupport/src/lib.rs
View File

@@ -14,6 +14,7 @@ use byteorder::NativeEndian;
use io::{Cursor, ProtoRead};
use libasync::block_async;
use libconfig;
use libcortex_a9::once_lock::OnceLock;
use log::{error, warn};
#[cfg(has_drtiosat)]
pub use pl::csr::drtiosat as rtio_core;
@@ -107,7 +108,7 @@ pub async fn report_async_rtio_errors() {
}
}
static mut RTIO_DEVICE_MAP: BTreeMap<u32, String> = BTreeMap::new();
static RTIO_DEVICE_MAP: OnceLock<BTreeMap<u32, String>> = OnceLock::new();
fn read_device_map() -> BTreeMap<u32, String> {
let mut device_map: BTreeMap<u32, String> = BTreeMap::new();
@@ -138,16 +139,18 @@ fn read_device_map() -> BTreeMap<u32, String> {
}
pub fn resolve_channel_name(channel: u32) -> String {
unsafe {
match RTIO_DEVICE_MAP.get(&channel) {
Some(val) => val.clone(),
None => String::from("unknown"),
}
match RTIO_DEVICE_MAP
.get()
.expect("cannot get device map before it is set up")
.get(&channel)
{
Some(val) => val.clone(),
None => String::from("unknown"),
}
}
pub fn setup_device_map() {
unsafe {
RTIO_DEVICE_MAP = read_device_map();
}
RTIO_DEVICE_MAP
.set(read_device_map())
.expect("device map can only be initialized once");
}

12
src/runtime/src/analyzer.rs
View File

@@ -6,7 +6,7 @@ use core::cell::RefCell;
use libasync::{smoltcp::TcpStream, task};
use libboard_artiq::drtio_routing;
use libboard_zynq::smoltcp::Error;
use libcortex_a9::{cache, mutex::Mutex};
use libcortex_a9::cache;
use log::{debug, info, warn};
use crate::{pl, proto_async::*};
@@ -56,7 +56,6 @@ pub mod remote_analyzer {
}
pub async fn get_data(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &drtio_routing::RoutingTable,
up_destinations: &Rc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
) -> Result<RemoteBuffer, drtio::Error> {
@@ -66,7 +65,7 @@ pub mod remote_analyzer {
let mut remote_sent_bytes = 0;
let mut remote_total_bytes = 0;
let data_vec = match drtio::analyzer_query(aux_mutex, routing_table, up_destinations).await {
let data_vec = match drtio::analyzer_query(routing_table, up_destinations).await {
Ok(data_vec) => data_vec,
Err(e) => return Err(e),
};
@@ -107,7 +106,6 @@ async fn write_header(stream: &mut TcpStream, header: &Header) -> Result<(), Err
async fn handle_connection(
stream: &mut TcpStream,
_aux_mutex: &Rc<Mutex<bool>>,
_routing_table: &drtio_routing::RoutingTable,
_up_destinations: &Rc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
) -> Result<(), Error> {
@@ -133,7 +131,7 @@ async fn handle_connection(
}
#[cfg(has_drtio)]
let remote = remote_analyzer::get_data(_aux_mutex, _routing_table, _up_destinations).await;
let remote = remote_analyzer::get_data(_routing_table, _up_destinations).await;
#[cfg(has_drtio)]
let (header, remote_data) = match remote {
Ok(remote) => (
@@ -185,11 +183,9 @@ async fn handle_connection(
}
pub fn start(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &Rc<RefCell<drtio_routing::RoutingTable>>,
up_destinations: &Rc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
) {
let aux_mutex = aux_mutex.clone();
let routing_table = routing_table.clone();
let up_destinations = up_destinations.clone();
task::spawn(async move {
@@ -198,7 +194,7 @@ pub fn start(
let mut stream = TcpStream::accept(1382, 2048, 2048).await.unwrap();
disarm();
let routing_table = routing_table.borrow();
let _ = handle_connection(&mut stream, &aux_mutex, &routing_table, &up_destinations)
let _ = handle_connection(&mut stream, &routing_table, &up_destinations)
.await
.map_err(|e| warn!("connection terminated: {:?}", e));
let _ = stream.flush().await;

69
src/runtime/src/comms.rs
View File

@@ -194,7 +194,6 @@ async fn handle_run_kernel(
stream: Option<&TcpStream>,
control: &Rc<RefCell<kernel::Control>>,
_up_destinations: &Rc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &drtio_routing::RoutingTable,
) -> Result<()> {
control.borrow_mut().tx.async_send(kernel::Message::StartRequest).await;
@@ -358,13 +357,13 @@ async fn handle_run_kernel(
.await;
}
kernel::Message::DmaPutRequest(recorder) => {
let _id = rtio_dma::put_record(aux_mutex, routing_table, recorder).await;
let _id = rtio_dma::put_record(routing_table, recorder).await;
#[cfg(has_drtio)]
rtio_dma::remote_dma::upload_traces(aux_mutex, routing_table, _id).await;
rtio_dma::remote_dma::upload_traces(routing_table, _id).await;
}
kernel::Message::DmaEraseRequest(name) => {
// prevent possible OOM when we have large DMA record replacement.
rtio_dma::erase(name, aux_mutex, routing_table).await;
rtio_dma::erase(name, routing_table).await;
}
kernel::Message::DmaGetRequest(name) => {
let result = rtio_dma::retrieve(name).await;
@@ -376,7 +375,7 @@ async fn handle_run_kernel(
}
#[cfg(has_drtio)]
kernel::Message::DmaStartRemoteRequest { id, timestamp } => {
rtio_dma::remote_dma::playback(aux_mutex, routing_table, id as u32, timestamp as u64).await;
rtio_dma::remote_dma::playback(routing_table, id as u32, timestamp as u64).await;
}
#[cfg(has_drtio)]
kernel::Message::DmaAwaitRemoteRequest(id) => {
@@ -406,7 +405,7 @@ async fn handle_run_kernel(
| kernel::Message::I2cRestartRequest(busno)
| kernel::Message::I2cStopRequest(busno)
| kernel::Message::I2cSwitchSelectRequest { busno, .. } => {
let result = rtio_mgt::drtio::i2c_send_basic(aux_mutex, routing_table, &reply, busno).await;
let result = rtio_mgt::drtio::i2c_send_basic(routing_table, &reply, busno).await;
let reply = match result {
Ok(succeeded) => kernel::Message::I2cBasicReply(succeeded),
Err(_) => kernel::Message::I2cBasicReply(false),
@@ -415,7 +414,7 @@ async fn handle_run_kernel(
}
#[cfg(has_drtio)]
kernel::Message::I2cWriteRequest { busno, data } => {
let result = rtio_mgt::drtio::i2c_send_write(aux_mutex, routing_table, busno, data).await;
let result = rtio_mgt::drtio::i2c_send_write(routing_table, busno, data).await;
let reply = match result {
Ok((succeeded, ack)) => kernel::Message::I2cWriteReply { succeeded, ack },
Err(_) => kernel::Message::I2cWriteReply {
@@ -427,7 +426,7 @@ async fn handle_run_kernel(
}
#[cfg(has_drtio)]
kernel::Message::I2cReadRequest { busno, ack } => {
let result = rtio_mgt::drtio::i2c_send_read(aux_mutex, routing_table, busno, ack).await;
let result = rtio_mgt::drtio::i2c_send_read(routing_table, busno, ack).await;
let reply = match result {
Ok((succeeded, data)) => kernel::Message::I2cReadReply { succeeded, data },
Err(_) => kernel::Message::I2cReadReply {
@@ -444,7 +443,7 @@ async fn handle_run_kernel(
run,
timestamp,
} => {
let succeeded = match subkernel::load(aux_mutex, routing_table, id, run, timestamp).await {
let succeeded = match subkernel::load(routing_table, id, run, timestamp).await {
Ok(()) => true,
Err(e) => {
error!("Error loading subkernel: {:?}", e);
@@ -459,7 +458,7 @@ async fn handle_run_kernel(
}
#[cfg(has_drtio)]
kernel::Message::SubkernelAwaitFinishRequest { id, timeout } => {
let res = subkernel::await_finish(aux_mutex, routing_table, id, timeout).await;
let res = subkernel::await_finish(routing_table, id, timeout).await;
let response = match res {
Ok(res) => {
if res.status == subkernel::FinishStatus::CommLost {
@@ -480,7 +479,7 @@ async fn handle_run_kernel(
}
#[cfg(has_drtio)]
kernel::Message::SubkernelMsgSend { id, destination, data } => {
let res = subkernel::message_send(aux_mutex, routing_table, id, destination.unwrap(), data).await;
let res = subkernel::message_send(routing_table, id, destination.unwrap(), data).await;
match res {
Ok(_) => (),
Err(e) => {
@@ -505,7 +504,7 @@ async fn handle_run_kernel(
Err(SubkernelError::CommLost) => kernel::Message::SubkernelError(kernel::SubkernelStatus::CommLost),
Err(SubkernelError::SubkernelException) => {
// just retrieve the exception
let status = subkernel::await_finish(aux_mutex, routing_table, id as u32, timeout)
let status = subkernel::await_finish(routing_table, id as u32, timeout)
.await
.unwrap();
kernel::Message::SubkernelError(kernel::SubkernelStatus::Exception(status.exception.unwrap()))
@@ -563,7 +562,7 @@ async fn handle_run_kernel(
}
#[cfg(has_drtio)]
kernel::Message::RtioInitRequest => {
rtio_mgt::drtio::reset(aux_mutex, routing_table).await;
rtio_mgt::drtio::reset(routing_table).await;
control.borrow_mut().tx.async_send(kernel::Message::RtioInitReply).await;
}
#[cfg(has_drtio)]
@@ -575,7 +574,6 @@ async fn handle_run_kernel(
let linkno = routing_table.0[destination as usize][0] - 1;
let reply = loop {
let result = rtio_mgt::drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::CXPReadRequest {
@@ -617,7 +615,6 @@ async fn handle_run_kernel(
let linkno = routing_table.0[destination as usize][0] - 1;
let reply = loop {
let drtioaux_packet = rtio_mgt::drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::CXPWrite32Request {
@@ -658,7 +655,6 @@ async fn handle_run_kernel(
} => {
let linkno = routing_table.0[destination as usize][0] - 1;
let drtioaux_packet = rtio_mgt::drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::CXPROIViewerSetupRequest {
@@ -689,7 +685,6 @@ async fn handle_run_kernel(
let linkno = routing_table.0[destination as usize][0] - 1;
let reply = loop {
let drtioaux_packet = rtio_mgt::drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::CXPROIViewerDataRequest { destination },
@@ -736,7 +731,6 @@ async fn handle_flash_kernel(
buffer: &Vec<u8>,
control: &Rc<RefCell<kernel::Control>>,
_up_destinations: &Rc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
_aux_mutex: &Rc<Mutex<bool>>,
_routing_table: &drtio_routing::RoutingTable,
) -> Result<()> {
if buffer[0] == elf::ELFMAG0 && buffer[1] == elf::ELFMAG1 && buffer[2] == elf::ELFMAG2 && buffer[3] == elf::ELFMAG3
@@ -763,7 +757,7 @@ async fn handle_flash_kernel(
if up {
let subkernel_lib = entry.data().to_vec();
subkernel::add_subkernel(sid, dest, subkernel_lib).await;
match subkernel::upload(_aux_mutex, _routing_table, sid).await {
match subkernel::upload(_routing_table, sid).await {
Ok(_) => (),
Err(_) => return Err(Error::UnexpectedPattern),
}
@@ -824,7 +818,6 @@ async fn handle_connection(
stream: &mut TcpStream,
control: Rc<RefCell<kernel::Control>>,
up_destinations: &Rc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &drtio_routing::RoutingTable,
) -> Result<()> {
stream.set_ack_delay(None);
@@ -853,7 +846,7 @@ async fn handle_connection(
load_kernel(&buffer, &control, Some(stream)).await?;
}
Request::RunKernel => {
handle_run_kernel(Some(stream), &control, &up_destinations, aux_mutex, routing_table).await?;
handle_run_kernel(Some(stream), &control, &up_destinations, routing_table).await?;
}
Request::UploadSubkernel => {
#[cfg(has_drtio)]
@@ -862,7 +855,7 @@ async fn handle_connection(
let destination = read_i8(stream).await? as u8;
let buffer = read_bytes(stream, 1024 * 1024).await?;
subkernel::add_subkernel(id, destination, buffer).await;
match subkernel::upload(aux_mutex, routing_table, id).await {
match subkernel::upload(routing_table, id).await {
Ok(_) => write_header(stream, Reply::LoadCompleted).await?,
Err(_) => {
write_header(stream, Reply::LoadFailed).await?;
@@ -927,7 +920,6 @@ pub fn main() {
Sockets::init(32);
let aux_mutex: Rc<Mutex<bool>> = Rc::new(Mutex::new(false));
#[cfg(has_drtio)]
let drtio_routing_table = Rc::new(RefCell::new(drtio_routing::config_routing_table(pl::csr::DRTIO.len())));
#[cfg(not(has_drtio))]
@@ -936,38 +928,26 @@ pub fn main() {
#[cfg(has_drtio_routing)]
drtio_routing::interconnect_disable_all();
rtio_mgt::startup(&aux_mutex, &drtio_routing_table, &up_destinations);
rtio_mgt::startup(&drtio_routing_table, &up_destinations);
ksupport::setup_device_map();
analyzer::start(&aux_mutex, &drtio_routing_table, &up_destinations);
moninj::start(&aux_mutex, &drtio_routing_table);
analyzer::start(&drtio_routing_table, &up_destinations);
moninj::start(&drtio_routing_table);
let control: Rc<RefCell<kernel::Control>> = Rc::new(RefCell::new(kernel::Control::start()));
if let Ok(buffer) = libconfig::read("startup_kernel") {
info!("Loading startup kernel...");
let routing_table = drtio_routing_table.borrow();
if let Ok(()) = task::block_on(handle_flash_kernel(
&buffer,
&control,
&up_destinations,
&aux_mutex,
&routing_table,
)) {
if let Ok(()) = task::block_on(handle_flash_kernel(&buffer, &control, &up_destinations, &routing_table)) {
info!("Starting startup kernel...");
let _ = task::block_on(handle_run_kernel(
None,
&control,
&up_destinations,
&aux_mutex,
&routing_table,
));
let _ = task::block_on(handle_run_kernel(None, &control, &up_destinations, &routing_table));
info!("Startup kernel finished!");
} else {
error!("Error loading startup kernel!");
}
}
mgmt::start(Some(mgmt::DrtioContext(aux_mutex.clone(), drtio_routing_table.clone())));
mgmt::start(Some(drtio_routing_table.clone()));
task::spawn(async move {
let connection = Rc::new(Semaphore::new(1, 1));
@@ -1001,7 +981,6 @@ pub fn main() {
let terminate = terminate.clone();
let can_restart_idle = can_restart_idle.clone();
let up_destinations = up_destinations.clone();
let aux_mutex = aux_mutex.clone();
let routing_table = drtio_routing_table.clone();
// we make sure the value of terminate is 0 before we start
@@ -1012,7 +991,7 @@ pub fn main() {
select_biased! {
_ = (async {
if let Some(stream) = &mut maybe_stream {
let _ = handle_connection(stream, control.clone(), &up_destinations, &aux_mutex, &routing_table)
let _ = handle_connection(stream, control.clone(), &up_destinations, &routing_table)
.await
.map_err(|e| warn!("connection terminated: {}", e));
}
@@ -1021,10 +1000,10 @@ pub fn main() {
Some(buffer) => {
loop {
info!("loading idle kernel");
match handle_flash_kernel(&buffer, &control, &up_destinations, &aux_mutex, &routing_table).await {
match handle_flash_kernel(&buffer, &control, &up_destinations, &routing_table).await {
Ok(_) => {
info!("running idle kernel");
match handle_run_kernel(None, &control, &up_destinations, &aux_mutex, &routing_table).await {
match handle_run_kernel(None, &control, &up_destinations, &routing_table).await {
Ok(_) => info!("idle kernel finished"),
Err(_) => warn!("idle kernel running error")
}

78
src/runtime/src/mgmt.rs
View File

@@ -9,7 +9,6 @@ use libboard_artiq::{drtio_routing::RoutingTable,
logger::{BufferLogger, LogBufferRef}};
use libboard_zynq::smoltcp;
use libconfig;
use libcortex_a9::mutex::Mutex;
use log::{self, debug, error, info, warn};
use num_derive::FromPrimitive;
use num_traits::FromPrimitive;
@@ -144,7 +143,6 @@ mod remote_coremgmt {
pub async fn get_log(
stream: &mut TcpStream,
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
linkno: u8,
destination: u8,
@@ -152,7 +150,6 @@ mod remote_coremgmt {
let mut buffer = Vec::new();
loop {
let reply = drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::CoreMgmtGetLogRequest {
@@ -187,18 +184,11 @@ mod remote_coremgmt {
pub async fn clear_log(
stream: &mut TcpStream,
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
linkno: u8,
destination: u8,
) -> Result<()> {
let reply = drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::CoreMgmtClearLogRequest { destination },
)
.await;
let reply = drtio::aux_transact(linkno, routing_table, &Packet::CoreMgmtClearLogRequest { destination }).await;
match reply {
Ok(Packet::CoreMgmtReply { succeeded: true }) => {
@@ -220,7 +210,6 @@ mod remote_coremgmt {
pub async fn pull_log(
stream: &mut TcpStream,
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
linkno: u8,
destination: u8,
@@ -241,7 +230,6 @@ mod remote_coremgmt {
}
let reply = drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::CoreMgmtGetLogRequest {
@@ -276,14 +264,12 @@ mod remote_coremgmt {
pub async fn set_log_filter(
stream: &mut TcpStream,
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
linkno: u8,
destination: u8,
level: log::LevelFilter,
) -> Result<()> {
let reply = drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::CoreMgmtSetLogLevelRequest {
@@ -313,14 +299,12 @@ mod remote_coremgmt {
pub async fn set_uart_log_filter(
stream: &mut TcpStream,
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
linkno: u8,
destination: u8,
level: log::LevelFilter,
) -> Result<()> {
let reply = drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::CoreMgmtSetUartLogLevelRequest {
@@ -350,7 +334,6 @@ mod remote_coremgmt {
pub async fn config_read(
stream: &mut TcpStream,
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
linkno: u8,
destination: u8,
@@ -361,7 +344,6 @@ mod remote_coremgmt {
config_key[..len].clone_from_slice(key.as_bytes());
let mut reply = drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::CoreMgmtConfigReadRequest {
@@ -385,7 +367,6 @@ mod remote_coremgmt {
}
reply = drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::CoreMgmtConfigReadContinue {
@@ -410,7 +391,6 @@ mod remote_coremgmt {
pub async fn config_write(
stream: &mut TcpStream,
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
linkno: u8,
destination: u8,
@@ -423,7 +403,6 @@ mod remote_coremgmt {
match drtio::partition_data(
linkno,
aux_mutex,
routing_table,
&message,
|slice, status, len: usize| Packet::CoreMgmtConfigWriteRequest {
@@ -456,7 +435,6 @@ mod remote_coremgmt {
pub async fn config_remove(
stream: &mut TcpStream,
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
linkno: u8,
destination: u8,
@@ -467,7 +445,6 @@ mod remote_coremgmt {
config_key[..len].clone_from_slice(key.as_bytes());
let reply = drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::CoreMgmtConfigRemoveRequest {
@@ -498,13 +475,11 @@ mod remote_coremgmt {
pub async fn config_erase(
stream: &mut TcpStream,
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
linkno: u8,
destination: u8,
) -> Result<()> {
let reply = drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::CoreMgmtConfigEraseRequest {
@@ -533,13 +508,11 @@ mod remote_coremgmt {
pub async fn reboot(
stream: &mut TcpStream,
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
linkno: u8,
destination: u8,
) -> Result<()> {
let reply = drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::CoreMgmtRebootRequest {
@@ -568,13 +541,11 @@ mod remote_coremgmt {
pub async fn debug_allocator(
stream: &mut TcpStream,
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
linkno: u8,
destination: u8,
) -> Result<()> {
let reply = drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::CoreMgmtAllocatorDebugRequest {
@@ -601,7 +572,6 @@ mod remote_coremgmt {
pub async fn image_write(
stream: &mut TcpStream,
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
linkno: u8,
destination: u8,
@@ -610,7 +580,6 @@ mod remote_coremgmt {
let mut image = &image[..];
let alloc_reply = drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::CoreMgmtFlashRequest {
@@ -640,7 +609,6 @@ mod remote_coremgmt {
let last = image.is_empty();
let reply = drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::CoreMgmtFlashAddDataRequest {
@@ -836,13 +804,13 @@ mod local_coremgmt {
#[cfg(has_drtio)]
macro_rules! process {
($stream: ident, $drtio_context:ident, $destination:expr, $func:ident $(, $param:expr)*) => {{
($stream: ident, $routing_table:ident, $destination:expr, $func:ident $(, $param:expr)*) => {{
if $destination == 0 {
local_coremgmt::$func($stream, $($param, )*).await
} else if let Some(DrtioContext(ref aux_mutex, ref routing_table)) = $drtio_context {
} else if let Some(ref routing_table) = $routing_table {
let routing_table = routing_table.borrow();
let linkno = routing_table.0[$destination as usize][0] - 1 as u8;
remote_coremgmt::$func($stream, &aux_mutex, &routing_table, linkno, $destination, $($param, )*).await
remote_coremgmt::$func($stream, &routing_table, linkno, $destination, $($param, )*).await
} else {
error!("coremgmt-over-drtio not supported for panicked device, please reboot");
write_i8($stream, Reply::Error as i8).await?;
@@ -853,19 +821,15 @@ macro_rules! process {
#[cfg(not(has_drtio))]
macro_rules! process {
($stream: ident, $drtio_context:ident, $destination:expr, $func:ident $(, $param:expr)*) => {{
($stream: ident, $routing_table:ident, $destination:expr, $func:ident $(, $param:expr)*) => {{
local_coremgmt::$func($stream, $($param, )*).await
}}
}
#[allow(dead_code)]
#[derive(Clone)]
pub struct DrtioContext(pub Rc<Mutex<bool>>, pub Rc<RefCell<RoutingTable>>);
async fn handle_connection(
stream: &mut TcpStream,
pull_id: Rc<RefCell<u32>>,
_drtio_context: Option<DrtioContext>,
_routing_table: Option<Rc<RefCell<RoutingTable>>>,
) -> Result<()> {
if !expect(&stream, b"ARTIQ management\n").await? {
return Err(Error::UnexpectedPattern);
@@ -881,20 +845,20 @@ async fn handle_connection(
}
let msg: Request = FromPrimitive::from_i8(msg?).ok_or(Error::UnrecognizedPacket)?;
match msg {
Request::GetLog => process!(stream, _drtio_context, _destination, get_log),
Request::ClearLog => process!(stream, _drtio_context, _destination, clear_log),
Request::PullLog => process!(stream, _drtio_context, _destination, pull_log, &pull_id),
Request::GetLog => process!(stream, _routing_table, _destination, get_log),
Request::ClearLog => process!(stream, _routing_table, _destination, clear_log),
Request::PullLog => process!(stream, _routing_table, _destination, pull_log, &pull_id),
Request::SetLogFilter => {
let lvl = read_log_level_filter(stream).await?;
process!(stream, _drtio_context, _destination, set_log_filter, lvl)
process!(stream, _routing_table, _destination, set_log_filter, lvl)
}
Request::SetUartLogFilter => {
let lvl = read_log_level_filter(stream).await?;
process!(stream, _drtio_context, _destination, set_uart_log_filter, lvl)
process!(stream, _routing_table, _destination, set_uart_log_filter, lvl)
}
Request::ConfigRead => {
let key = read_key(stream).await?;
process!(stream, _drtio_context, _destination, config_read, &key)
process!(stream, _routing_table, _destination, config_read, &key)
}
Request::ConfigWrite => {
let key = read_key(stream).await?;
@@ -905,20 +869,20 @@ async fn handle_connection(
buffer.set_len(len);
}
read_chunk(stream, &mut buffer).await?;
process!(stream, _drtio_context, _destination, config_write, &key, buffer)
process!(stream, _routing_table, _destination, config_write, &key, buffer)
}
Request::ConfigRemove => {
let key = read_key(stream).await?;
process!(stream, _drtio_context, _destination, config_remove, &key)
process!(stream, _routing_table, _destination, config_remove, &key)
}
Request::Reboot => {
process!(stream, _drtio_context, _destination, reboot)
process!(stream, _routing_table, _destination, reboot)
}
Request::ConfigErase => {
process!(stream, _drtio_context, _destination, config_erase)
process!(stream, _routing_table, _destination, config_erase)
}
Request::DebugAllocator => {
process!(stream, _drtio_context, _destination, debug_allocator)
process!(stream, _routing_table, _destination, debug_allocator)
}
Request::Flash => {
let len = read_i32(stream).await?;
@@ -931,22 +895,22 @@ async fn handle_connection(
buffer.set_len(len as usize);
}
read_chunk(stream, &mut buffer).await?;
process!(stream, _drtio_context, _destination, image_write, buffer)
process!(stream, _routing_table, _destination, image_write, buffer)
}
}?;
}
}
pub fn start(drtio_context: Option<DrtioContext>) {
pub fn start(routing_table: Option<Rc<RefCell<RoutingTable>>>) {
task::spawn(async move {
let pull_id = Rc::new(RefCell::new(0u32));
loop {
let mut stream = TcpStream::accept(1380, 2048, 2048).await.unwrap();
let pull_id = pull_id.clone();
let drtio_context = drtio_context.clone();
let routing_table = routing_table.clone();
task::spawn(async move {
info!("received connection");
let _ = handle_connection(&mut stream, pull_id, drtio_context)
let _ = handle_connection(&mut stream, pull_id, routing_table)
.await
.map_err(|e| warn!("connection terminated: {:?}", e));
let _ = stream.flush().await;

37
src/runtime/src/moninj.rs
View File

@@ -5,7 +5,6 @@ use futures::{FutureExt, pin_mut, select_biased};
use libasync::{smoltcp::TcpStream, task};
use libboard_artiq::drtio_routing;
use libboard_zynq::{smoltcp, timer};
use libcortex_a9::mutex::Mutex;
use log::{debug, info, warn};
use num_derive::{FromPrimitive, ToPrimitive};
use num_traits::{FromPrimitive, ToPrimitive};
@@ -57,10 +56,10 @@ mod remote_moninj {
use log::error;
use super::*;
use crate::rtio_mgt::{drtio, drtio::Error as DrtioError};
use crate::rtio_mgt::{drtio,
drtio::{AUX_MUTEX, Error as DrtioError}};
pub async fn read_probe(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &drtio_routing::RoutingTable,
linkno: u8,
destination: u8,
@@ -68,7 +67,6 @@ mod remote_moninj {
probe: i8,
) -> i64 {
let reply = drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&drtioaux_async::Packet::MonitorRequest {
@@ -90,7 +88,6 @@ mod remote_moninj {
}
pub async fn inject(
aux_mutex: &Rc<Mutex<bool>>,
_routing_table: &drtio_routing::RoutingTable,
linkno: u8,
destination: u8,
@@ -98,7 +95,7 @@ mod remote_moninj {
overrd: i8,
value: i8,
) {
let _lock = aux_mutex.async_lock().await;
let _lock = AUX_MUTEX.async_lock().await;
drtioaux_async::send(
linkno,
&drtioaux_async::Packet::InjectionRequest {
@@ -113,7 +110,6 @@ mod remote_moninj {
}
pub async fn read_injection_status(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &drtio_routing::RoutingTable,
linkno: u8,
destination: u8,
@@ -121,7 +117,6 @@ mod remote_moninj {
overrd: i8,
) -> i8 {
let reply = drtio::aux_transact(
aux_mutex,
linkno,
routing_table,
&drtioaux_async::Packet::InjectionStatusRequest {
@@ -174,7 +169,7 @@ mod local_moninj {
#[cfg(has_drtio)]
macro_rules! dispatch {
($aux_mutex:ident, $routing_table:ident, $channel:expr, $func:ident $(, $param:expr)*) => {{
($routing_table:ident, $channel:expr, $func:ident $(, $param:expr)*) => {{
let destination = ($channel >> 16) as u8;
let channel = $channel;
let hop = $routing_table.0[destination as usize][0];
@@ -182,24 +177,20 @@ macro_rules! dispatch {
local_moninj::$func(channel.into(), $($param, )*)
} else {
let linkno = hop - 1 as u8;
remote_moninj::$func($aux_mutex, $routing_table, linkno, destination, channel, $($param, )*).await
remote_moninj::$func($routing_table, linkno, destination, channel, $($param, )*).await
}
}}
}
#[cfg(not(has_drtio))]
macro_rules! dispatch {
($aux_mutex:ident, $routing_table:ident, $channel:expr, $func:ident $(, $param:expr)*) => {{
($routing_table:ident, $channel:expr, $func:ident $(, $param:expr)*) => {{
let channel = $channel as u16;
local_moninj::$func(channel.into(), $($param, )*)
}}
}
async fn handle_connection(
stream: &TcpStream,
_aux_mutex: &Rc<Mutex<bool>>,
_routing_table: &drtio_routing::RoutingTable,
) -> Result<()> {
async fn handle_connection(stream: &TcpStream, _routing_table: &drtio_routing::RoutingTable) -> Result<()> {
if !expect(&stream, b"ARTIQ moninj\n").await? {
return Err(Error::UnexpectedPattern);
}
@@ -247,13 +238,13 @@ async fn handle_connection(
let channel = read_i32(&stream).await?;
let overrd = read_i8(&stream).await?;
let value = read_i8(&stream).await?;
dispatch!(_aux_mutex, _routing_table, channel, inject, overrd, value);
dispatch!(_routing_table, channel, inject, overrd, value);
debug!("INJECT channel {}, overrd {}, value {}", channel, overrd, value);
},
HostMessage::GetInjectionStatus => {
let channel = read_i32(&stream).await?;
let overrd = read_i8(&stream).await?;
let value = dispatch!(_aux_mutex, _routing_table, channel, read_injection_status, overrd);
let value = dispatch!(_routing_table, channel, read_injection_status, overrd);
write_i8(&stream, DeviceMessage::InjectionStatus.to_i8().unwrap()).await?;
write_i32(&stream, channel).await?;
write_i8(&stream, overrd).await?;
@@ -263,7 +254,7 @@ async fn handle_connection(
},
_ = timeout_f => {
for (&(channel, probe), previous) in probe_watch_list.iter_mut() {
let current = dispatch!(_aux_mutex, _routing_table, channel, read_probe, probe);
let current = dispatch!(_routing_table, channel, read_probe, probe);
if previous.is_none() || previous.unwrap() != current {
write_i8(&stream, DeviceMessage::MonitorStatus.to_i8().unwrap()).await?;
write_i32(&stream, channel).await?;
@@ -273,7 +264,7 @@ async fn handle_connection(
}
}
for (&(channel, overrd), previous) in inject_watch_list.iter_mut() {
let current = dispatch!(_aux_mutex, _routing_table, channel, read_injection_status, overrd);
let current = dispatch!(_routing_table, channel, read_injection_status, overrd);
if previous.is_none() || previous.unwrap() != current {
write_i8(&stream, DeviceMessage::InjectionStatus.to_i8().unwrap()).await?;
write_i32(&stream, channel).await?;
@@ -288,18 +279,16 @@ async fn handle_connection(
}
}
pub fn start(aux_mutex: &Rc<Mutex<bool>>, routing_table: &Rc<RefCell<drtio_routing::RoutingTable>>) {
let aux_mutex = aux_mutex.clone();
pub fn start(routing_table: &Rc<RefCell<drtio_routing::RoutingTable>>) {
let routing_table = routing_table.clone();
task::spawn(async move {
loop {
let aux_mutex = aux_mutex.clone();
let routing_table = routing_table.clone();
let stream = TcpStream::accept(1383, 2048, 2048).await.unwrap();
task::spawn(async move {
info!("received connection");
let routing_table = routing_table.borrow();
let result = handle_connection(&stream, &aux_mutex, &routing_table).await;
let result = handle_connection(&stream, &routing_table).await;
match result {
Err(Error::NetworkError(smoltcp::Error::Finished)) => info!("peer closed connection"),
Err(error) => warn!("connection terminated: {}", error),

58
src/runtime/src/rtio_dma.rs
View File

@@ -1,4 +1,4 @@
use alloc::{collections::BTreeMap, rc::Rc, string::String, vec::Vec};
use alloc::{collections::BTreeMap, string::String, vec::Vec};
#[cfg(has_drtio)]
use core::mem;
@@ -105,12 +105,11 @@ pub mod remote_dma {
Ok(playback_state)
}
pub async fn upload_traces(&mut self, aux_mutex: &Rc<Mutex<bool>>, routing_table: &RoutingTable) {
pub async fn upload_traces(&mut self, routing_table: &RoutingTable) {
let mut lock = self.traces.async_lock().await;
let trace_iter = lock.iter_mut();
for (destination, trace) in trace_iter {
match drtio::ddma_upload_trace(aux_mutex, routing_table, self.id, *destination, trace.get_trace()).await
{
match drtio::ddma_upload_trace(routing_table, self.id, *destination, trace.get_trace()).await {
Ok(_) => trace.state = RemoteState::Loaded,
Err(e) => error!("Error adding DMA trace on destination {}: {}", destination, e),
}
@@ -118,11 +117,11 @@ pub mod remote_dma {
*(self.done_count.async_lock().await) = 0;
}
pub async fn erase(&mut self, aux_mutex: &Rc<Mutex<bool>>, routing_table: &RoutingTable) {
pub async fn erase(&mut self, routing_table: &RoutingTable) {
let lock = self.traces.async_lock().await;
let trace_iter = lock.keys();
for destination in trace_iter {
match drtio::ddma_send_erase(aux_mutex, routing_table, self.id, *destination).await {
match drtio::ddma_send_erase(routing_table, self.id, *destination).await {
Ok(_) => (),
Err(e) => error!("Error adding DMA trace on destination {}: {}", destination, e),
}
@@ -140,7 +139,7 @@ pub mod remote_dma {
*(self.done_count.async_lock().await) += 1;
}
pub async fn playback(&self, aux_mutex: &Rc<Mutex<bool>>, routing_table: &RoutingTable, timestamp: u64) {
pub async fn playback(&self, routing_table: &RoutingTable, timestamp: u64) {
let mut dest_list: Vec<u8> = Vec::new();
{
let lock = self.traces.async_lock().await;
@@ -156,26 +155,18 @@ pub mod remote_dma {
// mutex lock must be dropped before sending a playback request to avoid a deadlock,
// if PlaybackStatus is sent from another satellite and the state must be updated.
for destination in dest_list {
match drtio::ddma_send_playback(aux_mutex, routing_table, self.id, destination, timestamp).await {
match drtio::ddma_send_playback(routing_table, self.id, destination, timestamp).await {
Ok(_) => (),
Err(e) => error!("Error during remote DMA playback: {}", e),
}
}
}
pub async fn destination_changed(
&mut self,
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
destination: u8,
up: bool,
) {
pub async fn destination_changed(&mut self, routing_table: &RoutingTable, destination: u8, up: bool) {
// update state of the destination, resend traces if it's up
if let Some(trace) = self.traces.async_lock().await.get_mut(&destination) {
if up {
match drtio::ddma_upload_trace(aux_mutex, routing_table, self.id, destination, trace.get_trace())
.await
{
match drtio::ddma_upload_trace(routing_table, self.id, destination, trace.get_trace()).await {
Ok(_) => trace.state = RemoteState::Loaded,
Err(e) => error!("Error adding DMA trace on destination {}: {}", destination, e),
}
@@ -201,22 +192,22 @@ pub mod remote_dma {
trace_set.await_done(timeout).await
}
pub async fn erase(aux_mutex: &Rc<Mutex<bool>>, routing_table: &RoutingTable, id: u32) {
pub async fn erase(routing_table: &RoutingTable, id: u32) {
let trace_set = unsafe { TRACES.get_mut(&id).unwrap() };
trace_set.erase(aux_mutex, routing_table).await;
trace_set.erase(routing_table).await;
unsafe {
TRACES.remove(&id);
}
}
pub async fn upload_traces(aux_mutex: &Rc<Mutex<bool>>, routing_table: &RoutingTable, id: u32) {
pub async fn upload_traces(routing_table: &RoutingTable, id: u32) {
let trace_set = unsafe { TRACES.get_mut(&id).unwrap() };
trace_set.upload_traces(aux_mutex, routing_table).await;
trace_set.upload_traces(routing_table).await;
}
pub async fn playback(aux_mutex: &Rc<Mutex<bool>>, routing_table: &RoutingTable, id: u32, timestamp: u64) {
pub async fn playback(routing_table: &RoutingTable, id: u32, timestamp: u64) {
let trace_set = unsafe { TRACES.get_mut(&id).unwrap() };
trace_set.playback(aux_mutex, routing_table, timestamp).await;
trace_set.playback(routing_table, timestamp).await;
}
pub async fn playback_done(id: u32, destination: u8, error: u8, channel: u32, timestamp: u64) {
@@ -224,17 +215,10 @@ pub mod remote_dma {
trace_set.playback_done(destination, error, channel, timestamp).await;
}
pub async fn destination_changed(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
destination: u8,
up: bool,
) {
pub async fn destination_changed(routing_table: &RoutingTable, destination: u8, up: bool) {
let trace_iter = unsafe { TRACES.values_mut() };
for trace_set in trace_iter {
trace_set
.destination_changed(aux_mutex, routing_table, destination, up)
.await;
trace_set.destination_changed(routing_table, destination, up).await;
}
}
@@ -244,7 +228,7 @@ pub mod remote_dma {
}
}
pub async fn put_record(_aux_mutex: &Rc<Mutex<bool>>, _routing_table: &RoutingTable, mut recorder: DmaRecorder) -> u32 {
pub async fn put_record(_routing_table: &RoutingTable, mut recorder: DmaRecorder) -> u32 {
#[cfg(has_drtio)]
let mut remote_traces: BTreeMap<u8, Vec<u8>> = BTreeMap::new();
@@ -293,7 +277,7 @@ pub async fn put_record(_aux_mutex: &Rc<Mutex<bool>>, _routing_table: &RoutingTa
#[cfg(has_drtio)]
{
if let Some((old_id, _v, _d)) = _old_record {
remote_dma::erase(_aux_mutex, _routing_table, old_id).await;
remote_dma::erase(_routing_table, old_id).await;
}
remote_dma::add_traces(ptr, remote_traces);
}
@@ -301,11 +285,11 @@ pub async fn put_record(_aux_mutex: &Rc<Mutex<bool>>, _routing_table: &RoutingTa
ptr
}
pub async fn erase(name: String, _aux_mutex: &Rc<Mutex<bool>>, _routing_table: &RoutingTable) {
pub async fn erase(name: String, _routing_table: &RoutingTable) {
let _entry = DMA_RECORD_STORE.lock().remove(&name);
#[cfg(has_drtio)]
if let Some((id, _v, _d)) = _entry {
remote_dma::erase(_aux_mutex, _routing_table, id).await;
remote_dma::erase(_routing_table, id).await;
}
}

139
src/runtime/src/rtio_mgt.rs
View File

@@ -30,6 +30,8 @@ pub mod drtio {
const DRTIO_EEM_LINKNOS: core::ops::Range<usize> =
(csr::DRTIO.len() - csr::CONFIG_EEM_DRTIO_COUNT as usize)..csr::DRTIO.len();
pub static AUX_MUTEX: Mutex<bool> = Mutex::new(false);
#[derive(Debug, PartialEq, Eq, Clone, Copy)]
pub enum Error {
Timeout,
@@ -66,16 +68,14 @@ pub mod drtio {
}
pub fn startup(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &Rc<RefCell<RoutingTable>>,
up_destinations: &Rc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
) {
let aux_mutex = aux_mutex.clone();
let routing_table = routing_table.clone();
let up_destinations = up_destinations.clone();
task::spawn(async move {
let routing_table = routing_table.borrow();
link_task(&aux_mutex, &routing_table, &up_destinations).await;
link_task(&routing_table, &up_destinations).await;
});
}
@@ -200,16 +200,11 @@ pub mod drtio {
}
}
pub async fn aux_transact(
aux_mutex: &Mutex<bool>,
linkno: u8,
routing_table: &RoutingTable,
request: &Packet,
) -> Result<Packet, Error> {
pub async fn aux_transact(linkno: u8, routing_table: &RoutingTable, request: &Packet) -> Result<Packet, Error> {
if !link_rx_up(linkno).await {
return Err(Error::LinkDown);
}
let _lock = aux_mutex.async_lock().await;
let _lock = AUX_MUTEX.async_lock().await;
drtioaux_async::send(linkno, request).await.unwrap();
loop {
let packet = recv_aux_timeout(linkno, 200).await?;
@@ -226,7 +221,7 @@ pub mod drtio {
}
}
async fn ping_remote(aux_mutex: &Rc<Mutex<bool>>, linkno: u8, routing_table: &RoutingTable) -> u32 {
async fn ping_remote(linkno: u8, routing_table: &RoutingTable) -> u32 {
let mut count = 0;
loop {
if !link_rx_up(linkno).await {
@@ -236,7 +231,7 @@ pub mod drtio {
if count > 100 {
return 0;
}
let reply = aux_transact(aux_mutex, linkno, routing_table, &Packet::EchoRequest).await;
let reply = aux_transact(linkno, routing_table, &Packet::EchoRequest).await;
match reply {
Ok(Packet::EchoReply) => {
// make sure receive buffer is drained
@@ -248,8 +243,8 @@ pub mod drtio {
}
}
async fn sync_tsc(aux_mutex: &Rc<Mutex<bool>>, linkno: u8) -> Result<(), Error> {
let _lock = aux_mutex.async_lock().await;
async fn sync_tsc(linkno: u8) -> Result<(), Error> {
let _lock = AUX_MUTEX.async_lock().await;
unsafe {
(csr::DRTIO[linkno as usize].set_time_write)(1);
@@ -265,14 +260,9 @@ pub mod drtio {
}
}
async fn load_routing_table(
aux_mutex: &Rc<Mutex<bool>>,
linkno: u8,
routing_table: &RoutingTable,
) -> Result<(), Error> {
async fn load_routing_table(linkno: u8, routing_table: &RoutingTable) -> Result<(), Error> {
for i in 0..drtio_routing::DEST_COUNT {
let reply = aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::RoutingSetPath {
@@ -288,13 +278,8 @@ pub mod drtio {
Ok(())
}
async fn set_rank(
aux_mutex: &Rc<Mutex<bool>>,
linkno: u8,
rank: u8,
routing_table: &RoutingTable,
) -> Result<(), Error> {
let reply = aux_transact(aux_mutex, linkno, routing_table, &Packet::RoutingSetRank { rank: rank }).await?;
async fn set_rank(linkno: u8, rank: u8, routing_table: &RoutingTable) -> Result<(), Error> {
let reply = aux_transact(linkno, routing_table, &Packet::RoutingSetRank { rank: rank }).await?;
match reply {
Packet::RoutingAck => Ok(()),
_ => Err(Error::UnexpectedReply),
@@ -317,8 +302,8 @@ pub mod drtio {
}
}
async fn process_unsolicited_aux(aux_mutex: &Mutex<bool>, linkno: u8, routing_table: &RoutingTable) {
let _lock = aux_mutex.async_lock().await;
async fn process_unsolicited_aux(linkno: u8, routing_table: &RoutingTable) {
let _lock = AUX_MUTEX.async_lock().await;
match drtioaux_async::recv(linkno).await {
Ok(Some(packet)) => {
if let Some(packet) = process_async_packets(linkno, routing_table, packet).await {
@@ -374,7 +359,6 @@ pub mod drtio {
}
async fn destination_survey(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
up_links: &[bool],
up_destinations: &Rc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
@@ -388,7 +372,6 @@ pub mod drtio {
if destination_up(up_destinations, destination).await {
if up_links[linkno as usize] {
let reply = aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::DestinationStatusRequest {
@@ -399,8 +382,8 @@ 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, destination, false).await;
subkernel::destination_changed(aux_mutex, routing_table, destination, false).await;
remote_dma::destination_changed(routing_table, destination, false).await;
subkernel::destination_changed(routing_table, destination, false).await;
}
Ok(Packet::DestinationOkReply) => (),
Ok(Packet::DestinationSequenceErrorReply { channel }) => {
@@ -438,13 +421,12 @@ pub mod drtio {
}
} else {
destination_set_up(routing_table, up_destinations, destination, false).await;
remote_dma::destination_changed(aux_mutex, routing_table, destination, false).await;
subkernel::destination_changed(aux_mutex, routing_table, destination, false).await;
remote_dma::destination_changed(routing_table, destination, false).await;
subkernel::destination_changed(routing_table, destination, false).await;
}
} else {
if up_links[linkno as usize] {
let reply = aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::DestinationStatusRequest {
@@ -457,8 +439,8 @@ pub mod drtio {
Ok(Packet::DestinationOkReply) => {
destination_set_up(routing_table, up_destinations, destination, true).await;
init_buffer_space(destination as u8, linkno).await;
remote_dma::destination_changed(aux_mutex, routing_table, destination, true).await;
subkernel::destination_changed(aux_mutex, routing_table, destination, true).await;
remote_dma::destination_changed(routing_table, destination, true).await;
subkernel::destination_changed(routing_table, destination, true).await;
}
Ok(packet) => error!("[DEST#{}] received unexpected aux packet: {:?}", destination, packet),
Err(e) => error!("[DEST#{}] communication failed ({})", destination, e),
@@ -470,7 +452,6 @@ pub mod drtio {
}
pub async fn link_task(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
up_destinations: &Rc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
) {
@@ -485,7 +466,7 @@ pub mod drtio {
if up_links[linkno as usize] {
/* link was previously up */
if link_rx_up(linkno).await {
process_unsolicited_aux(aux_mutex, linkno, routing_table).await;
process_unsolicited_aux(linkno, routing_table).await;
process_local_errors(linkno).await;
} else {
info!("[LINK#{}] link is down", linkno);
@@ -514,17 +495,17 @@ pub mod drtio {
if link_rx_up(linkno).await {
info!("[LINK#{}] link RX became up, pinging", linkno);
let ping_count = ping_remote(aux_mutex, linkno, routing_table).await;
let ping_count = ping_remote(linkno, routing_table).await;
if ping_count > 0 {
info!("[LINK#{}] remote replied after {} packets", linkno, ping_count);
up_links[linkno as usize] = true;
if let Err(e) = sync_tsc(aux_mutex, linkno).await {
if let Err(e) = sync_tsc(linkno).await {
error!("[LINK#{}] failed to sync TSC ({})", linkno, e);
}
if let Err(e) = load_routing_table(aux_mutex, linkno, routing_table).await {
if let Err(e) = load_routing_table(linkno, routing_table).await {
error!("[LINK#{}] failed to load routing table ({})", linkno, e);
}
if let Err(e) = set_rank(aux_mutex, linkno, 1 as u8, routing_table).await {
if let Err(e) = set_rank(linkno, 1 as u8, routing_table).await {
error!("[LINK#{}] failed to set rank ({})", linkno, e);
}
info!("[LINK#{}] link initialization completed", linkno);
@@ -534,12 +515,12 @@ pub mod drtio {
}
}
}
destination_survey(aux_mutex, routing_table, &up_links, up_destinations).await;
destination_survey(routing_table, &up_links, up_destinations).await;
timer::async_delay_ms(200).await;
}
}
pub async fn reset(aux_mutex: &Rc<Mutex<bool>>, routing_table: &RoutingTable) {
pub async fn reset(routing_table: &RoutingTable) {
for linkno in 0..csr::DRTIO.len() {
unsafe {
(csr::DRTIO[linkno].reset_write)(1);
@@ -555,7 +536,7 @@ pub mod drtio {
for linkno in 0..csr::DRTIO.len() {
let linkno = linkno as u8;
if link_rx_up(linkno).await {
let reply = aux_transact(&aux_mutex, linkno, routing_table, &Packet::ResetRequest).await;
let reply = aux_transact(linkno, routing_table, &Packet::ResetRequest).await;
match reply {
Ok(Packet::ResetAck) => (),
Ok(_) => error!("[LINK#{}] reset failed, received unexpected aux packet", linkno),
@@ -567,7 +548,6 @@ pub mod drtio {
pub async fn partition_data<PacketF, HandlerF>(
linkno: u8,
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
data: &[u8],
packet_f: PacketF,
@@ -591,14 +571,13 @@ pub mod drtio {
i += len;
let status = PayloadStatus::from_status(first, last);
let packet = packet_f(&slice, status, len);
let reply = aux_transact(aux_mutex, linkno, routing_table, &packet).await?;
let reply = aux_transact(linkno, routing_table, &packet).await?;
reply_handler_f(&reply)?;
}
Ok(())
}
pub async fn ddma_upload_trace(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
id: u32,
destination: u8,
@@ -608,7 +587,6 @@ pub mod drtio {
let master_destination = get_master_destination(routing_table);
partition_data(
linkno,
aux_mutex,
routing_table,
trace,
|slice, status, len| Packet::DmaAddTraceRequest {
@@ -648,16 +626,10 @@ pub mod drtio {
.await
}
pub async fn ddma_send_erase(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
id: u32,
destination: u8,
) -> Result<(), Error> {
pub async fn ddma_send_erase(routing_table: &RoutingTable, id: u32, destination: u8) -> Result<(), Error> {
let linkno = routing_table.0[destination as usize][0] - 1;
let master_destination = get_master_destination(routing_table);
let reply = aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::DmaRemoveTraceRequest {
@@ -693,7 +665,6 @@ pub mod drtio {
}
pub async fn ddma_send_playback(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
id: u32,
destination: u8,
@@ -702,7 +673,6 @@ pub mod drtio {
let linkno = routing_table.0[destination as usize][0] - 1;
let master_destination = get_master_destination(routing_table);
let reply = aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::DmaPlaybackRequest {
@@ -738,14 +708,9 @@ pub mod drtio {
}
}
async fn analyzer_get_data(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
destination: u8,
) -> Result<RemoteBuffer, Error> {
async fn analyzer_get_data(routing_table: &RoutingTable, destination: u8) -> Result<RemoteBuffer, Error> {
let linkno = routing_table.0[destination as usize][0] - 1;
let reply = aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::AnalyzerHeaderRequest {
@@ -767,7 +732,6 @@ pub mod drtio {
let mut last_packet = false;
while !last_packet {
let reply = aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::AnalyzerDataRequest {
@@ -794,21 +758,19 @@ pub mod drtio {
}
pub async fn analyzer_query(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
up_destinations: &Rc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
) -> Result<Vec<RemoteBuffer>, Error> {
let mut remote_buffers: Vec<RemoteBuffer> = Vec::new();
for i in 1..drtio_routing::DEST_COUNT {
if destination_up(up_destinations, i as u8).await {
remote_buffers.push(analyzer_get_data(aux_mutex, routing_table, i as u8).await?);
remote_buffers.push(analyzer_get_data(routing_table, i as u8).await?);
}
}
Ok(remote_buffers)
}
pub async fn subkernel_upload(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
id: u32,
destination: u8,
@@ -817,7 +779,6 @@ pub mod drtio {
let linkno = routing_table.0[destination as usize][0] - 1;
partition_data(
linkno,
aux_mutex,
routing_table,
data,
|slice, status, len| Packet::SubkernelAddDataRequest {
@@ -837,7 +798,6 @@ pub mod drtio {
}
pub async fn subkernel_load(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
id: u32,
destination: u8,
@@ -847,7 +807,6 @@ pub mod drtio {
let linkno = routing_table.0[destination as usize][0] - 1;
let master_destination = get_master_destination(routing_table);
let reply = aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::SubkernelLoadRunRequest {
@@ -884,17 +843,12 @@ pub mod drtio {
}
}
pub async fn subkernel_retrieve_exception(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
destination: u8,
) -> Result<Vec<u8>, Error> {
pub async fn subkernel_retrieve_exception(routing_table: &RoutingTable, destination: u8) -> Result<Vec<u8>, Error> {
let linkno = routing_table.0[destination as usize][0] - 1;
let mut remote_data: Vec<u8> = Vec::new();
let master_destination = get_master_destination(routing_table);
loop {
let reply = aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::SubkernelExceptionRequest {
@@ -925,7 +879,6 @@ pub mod drtio {
}
pub async fn subkernel_send_message(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
id: u32,
destination: u8,
@@ -935,7 +888,6 @@ pub mod drtio {
let master_destination = get_master_destination(routing_table);
partition_data(
linkno,
aux_mutex,
routing_table,
message,
|slice, status, len| Packet::SubkernelMessage {
@@ -955,7 +907,6 @@ pub mod drtio {
}
pub async fn i2c_send_basic(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
request: &KernelMessage,
busno: u32,
@@ -975,24 +926,18 @@ pub mod drtio {
_ => unreachable!(),
};
let linkno = routing_table.0[destination as usize][0] - 1;
let reply = aux_transact(aux_mutex, linkno, routing_table, &packet).await?;
let reply = aux_transact(linkno, routing_table, &packet).await?;
match reply {
Packet::I2cBasicReply { succeeded } => Ok(succeeded),
_ => Err(Error::UnexpectedReply),
}
}
pub async fn i2c_send_write(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
busno: u32,
data: u8,
) -> Result<(bool, bool), Error> {
pub async fn i2c_send_write(routing_table: &RoutingTable, busno: u32, data: u8) -> Result<(bool, bool), Error> {
let destination = (busno >> 16) as u8;
let busno = busno as u8;
let linkno = routing_table.0[destination as usize][0] - 1;
let reply = aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::I2cWriteRequest {
@@ -1008,17 +953,11 @@ pub mod drtio {
}
}
pub async fn i2c_send_read(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
busno: u32,
ack: bool,
) -> Result<(bool, u8), Error> {
pub async fn i2c_send_read(routing_table: &RoutingTable, busno: u32, ack: bool) -> Result<(bool, u8), Error> {
let destination = (busno >> 16) as u8;
let busno = busno as u8;
let linkno = routing_table.0[destination as usize][0] - 1;
let reply = aux_transact(
aux_mutex,
linkno,
routing_table,
&Packet::I2cReadRequest {
@@ -1040,14 +979,13 @@ pub mod drtio {
use super::*;
pub fn startup(
_aux_mutex: &Rc<Mutex<bool>>,
_routing_table: &Rc<RefCell<RoutingTable>>,
_up_destinations: &Rc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
) {
}
#[allow(dead_code)]
pub fn reset(_aux_mutex: Rc<Mutex<bool>>, _routing_table: &RoutingTable) {}
pub fn reset(_routing_table: &RoutingTable) {}
}
fn toggle_sed_spread(val: u8) {
@@ -1073,12 +1011,11 @@ fn setup_sed_spread() {
}
pub fn startup(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &Rc<RefCell<RoutingTable>>,
up_destinations: &Rc<RefCell<[bool; drtio_routing::DEST_COUNT]>>,
) {
setup_sed_spread();
drtio::startup(aux_mutex, routing_table, up_destinations);
drtio::startup(routing_table, up_destinations);
unsafe {
csr::rtio_core::reset_phy_write(1);
}

32
src/runtime/src/subkernel.rs
View File

@@ -1,4 +1,4 @@
use alloc::{collections::BTreeMap, rc::Rc, vec::Vec};
use alloc::{collections::BTreeMap, vec::Vec};
use libasync::task;
use libboard_artiq::{drtio_routing::RoutingTable,
@@ -72,9 +72,9 @@ pub async fn add_subkernel(id: u32, destination: u8, kernel: Vec<u8>) {
.insert(id, Subkernel::new(destination, kernel));
}
pub async fn upload(aux_mutex: &Rc<Mutex<bool>>, routing_table: &RoutingTable, id: u32) -> Result<(), Error> {
pub async fn upload(routing_table: &RoutingTable, id: u32) -> Result<(), Error> {
if let Some(subkernel) = SUBKERNELS.async_lock().await.get_mut(&id) {
drtio::subkernel_upload(aux_mutex, routing_table, id, subkernel.destination, &subkernel.data).await?;
drtio::subkernel_upload(routing_table, id, subkernel.destination, &subkernel.data).await?;
subkernel.state = SubkernelState::Uploaded;
Ok(())
} else {
@@ -82,18 +82,12 @@ pub async fn upload(aux_mutex: &Rc<Mutex<bool>>, routing_table: &RoutingTable, i
}
}
pub async fn load(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
id: u32,
run: bool,
timestamp: u64,
) -> Result<(), Error> {
pub async fn load(routing_table: &RoutingTable, id: u32, run: bool, timestamp: u64) -> Result<(), Error> {
if let Some(subkernel) = SUBKERNELS.async_lock().await.get_mut(&id) {
if subkernel.state != SubkernelState::Uploaded {
return Err(Error::IncorrectState);
}
drtio::subkernel_load(aux_mutex, routing_table, id, subkernel.destination, run, timestamp).await?;
drtio::subkernel_load(routing_table, id, subkernel.destination, run, timestamp).await?;
if run {
subkernel.state = SubkernelState::Running;
}
@@ -124,12 +118,12 @@ pub async fn subkernel_finished(id: u32, with_exception: bool, exception_src: u8
}
}
pub async fn destination_changed(aux_mutex: &Rc<Mutex<bool>>, routing_table: &RoutingTable, destination: u8, up: bool) {
pub async fn destination_changed(routing_table: &RoutingTable, destination: u8, up: bool) {
let mut locked_subkernels = SUBKERNELS.async_lock().await;
for (id, subkernel) in locked_subkernels.iter_mut() {
if subkernel.destination == destination {
if up {
match drtio::subkernel_upload(aux_mutex, routing_table, *id, destination, &subkernel.data).await {
match drtio::subkernel_upload(routing_table, *id, destination, &subkernel.data).await {
Ok(_) => subkernel.state = SubkernelState::Uploaded,
Err(e) => error!("Error adding subkernel on destination {}: {}", destination, e),
}
@@ -145,12 +139,7 @@ pub async fn destination_changed(aux_mutex: &Rc<Mutex<bool>>, routing_table: &Ro
}
}
pub async fn await_finish(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
id: u32,
timeout: i64,
) -> Result<SubkernelFinished, Error> {
pub async fn await_finish(routing_table: &RoutingTable, id: u32, timeout: i64) -> Result<SubkernelFinished, Error> {
match SUBKERNELS.async_lock().await.get(&id).unwrap().state {
SubkernelState::Running | SubkernelState::Finished { .. } => (),
_ => return Err(Error::IncorrectState),
@@ -186,7 +175,7 @@ pub async fn await_finish(
id: id,
status: status,
exception: if let FinishStatus::Exception(dest) = status {
Some(drtio::subkernel_retrieve_exception(aux_mutex, routing_table, dest).await?)
Some(drtio::subkernel_retrieve_exception(routing_table, dest).await?)
} else {
None
},
@@ -294,11 +283,10 @@ pub async fn message_await(id: u32, timeout: i64) -> Result<Message, Error> {
}
pub async fn message_send<'a>(
aux_mutex: &Rc<Mutex<bool>>,
routing_table: &RoutingTable,
id: u32,
destination: u8,
message: Vec<u8>,
) -> Result<(), Error> {
Ok(drtio::subkernel_send_message(aux_mutex, routing_table, id, destination, &message).await?)
Ok(drtio::subkernel_send_message(routing_table, id, destination, &message).await?)
}