M-Labs
/
artiq-zynq
10
1
Fork 19
Code Issues 14 Pull Requests 2 Releases Wiki Activity

Compare commits

base: M-Labs:master
Branches Tags
M-Labs:master
M-Labs:release-8
M-Labs:release-7
M-Labs:release-6
mwojcik:subkernel_exception_pass
mwojcik:repeater_async_forward
mwojcik:sed_spread_config
mwojcik:repeater_clear_buffer
mwojcik:master
mwojcik:release-6
..
compare: mwojcik:sed_spread_config
Branches Tags
mwojcik:subkernel_exception_pass
mwojcik:repeater_async_forward
mwojcik:sed_spread_config
mwojcik:repeater_clear_buffer
mwojcik:master
mwojcik:release-6
M-Labs:master
M-Labs:release-8
M-Labs:release-7
M-Labs:release-6

1 Commits
master ... sed_spread

Author SHA1 Message Date
mwojcik 85e1d4f6e4 allow toggling SED spread with flash config key 2024-07-09 17:13:04 +08:00
18 changed files with 263 additions and 1326 deletions
Show Stats Expand all files Collapse all files

80
flake.lock
View File

@ -3,19 +3,19 @@
"artiq": {
"inputs": {
"artiq-comtools": "artiq-comtools",
"mozilla-overlay": "mozilla-overlay",
"nixpkgs": "nixpkgs",
"rust-overlay": "rust-overlay",
"sipyco": "sipyco",
"src-migen": "src-migen",
"src-misoc": "src-misoc",
"src-pythonparser": "src-pythonparser"
},
"locked": {
"lastModified": 1724411572,
"narHash": "sha256-33vj/pJ9iaVvaFP8uuBKMSQPN20mRlCbeBkTCNc9WB4=",
"lastModified": 1717638354,
"narHash": "sha256-eyI8OsOrn/j8ChbCpyFpS5VXBW8xSNGGIboFGQj/d4I=",
"ref": "refs/heads/master",
"rev": "352cf907ee67f7db5478fe23217cd5fcb7334617",
"revCount": 8996,
"rev": "ebc1e3fb767d6c0eac6eac20c3afeaba2ab70d1a",
"revCount": 8833,
"type": "git",
"url": "https://github.com/m-labs/artiq.git"
},
@ -37,11 +37,11 @@
]
},
"locked": {
"lastModified": 1720768567,
"narHash": "sha256-3VoK7o5MtHtbHLrc6Pv+eQWFtaz5Gd/YWyV5TD3c5Ss=",
"lastModified": 1717637438,
"narHash": "sha256-BXFidNm3Em8iChPGu1L0s2bY+f2yQ0VVid4MuOoTehw=",
"owner": "m-labs",
"repo": "artiq-comtools",
"rev": "f93570d8f2ed5a3cfb3e1c16ab00f2540551e994",
"rev": "78d27026efe76a13f7b4698a554f55811369ec4d",
"type": "github"
},
"original": {
@ -100,18 +100,34 @@
"type": "github"
}
},
"mozilla-overlay_3": {
"flake": false,
"locked": {
"lastModified": 1704373101,
"narHash": "sha256-+gi59LRWRQmwROrmE1E2b3mtocwueCQqZ60CwLG+gbg=",
"owner": "mozilla",
"repo": "nixpkgs-mozilla",
"rev": "9b11a87c0cc54e308fa83aac5b4ee1816d5418a2",
"type": "github"
},
"original": {
"owner": "mozilla",
"repo": "nixpkgs-mozilla",
"type": "github"
}
},
"nixpkgs": {
"locked": {
"lastModified": 1723362943,
"narHash": "sha256-dFZRVSgmJkyM0bkPpaYRtG/kRMRTorUIDj8BxoOt1T4=",
"lastModified": 1717281328,
"narHash": "sha256-evZPzpf59oNcDUXxh2GHcxHkTEG4fjae2ytWP85jXRo=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "a58bc8ad779655e790115244571758e8de055e3d",
"rev": "b3b2b28c1daa04fe2ae47c21bb76fd226eac4ca1",
"type": "github"
},
"original": {
"owner": "NixOS",
"ref": "nixos-unstable",
"ref": "nixos-24.05",
"repo": "nixpkgs",
"type": "github"
}
@ -119,32 +135,10 @@
"root": {
"inputs": {
"artiq": "artiq",
"mozilla-overlay": "mozilla-overlay",
"mozilla-overlay": "mozilla-overlay_2",
"zynq-rs": "zynq-rs"
}
},
"rust-overlay": {
"inputs": {
"nixpkgs": [
"artiq",
"nixpkgs"
]
},
"locked": {
"lastModified": 1719454714,
"narHash": "sha256-MojqG0lyUINkEk0b3kM2drsU5vyaF8DFZe/FAlZVOGs=",
"owner": "oxalica",
"repo": "rust-overlay",
"rev": "d1c527659cf076ecc4b96a91c702d080b213801e",
"type": "github"
},
"original": {
"owner": "oxalica",
"ref": "snapshot/2024-08-01",
"repo": "rust-overlay",
"type": "github"
}
},
"sipyco": {
"inputs": {
"nixpkgs": [
@ -169,11 +163,11 @@
"src-migen": {
"flake": false,
"locked": {
"lastModified": 1721561053,
"narHash": "sha256-z3LRhNmKZrjr6rFD0yxtccSa/SWvFIYmb+G/D5d2Jd8=",
"lastModified": 1715484909,
"narHash": "sha256-4DCHBUBfc/VA+7NW2Hr0+JP4NnKPru2uVJyZjCCk0Ws=",
"owner": "m-labs",
"repo": "migen",
"rev": "9279e8623f8433bc4f23ac51e5e2331bfe544417",
"rev": "4790bb577681a8c3a8d226bc196a4e5deb39e4df",
"type": "github"
},
"original": {
@ -233,18 +227,18 @@
},
"zynq-rs": {
"inputs": {
"mozilla-overlay": "mozilla-overlay_2",
"mozilla-overlay": "mozilla-overlay_3",
"nixpkgs": [
"artiq",
"nixpkgs"
]
},
"locked": {
"lastModified": 1720537402,
"narHash": "sha256-ybvaQ48SVBqYVqgYmGUdefGZkni7PJ90qYQPHnFOwDs=",
"lastModified": 1716889070,
"narHash": "sha256-w+M5NfukmcMTt73+u6Cmltjn4qpQH8bcoTGskB2IzBE=",
"ref": "refs/heads/master",
"rev": "b2b3e5c933cbc4b7cb14adde480d7561a3ae71ee",
"revCount": 648,
"rev": "51b8111e799ebe90ca231ae4d02f84ede32632d5",
"revCount": 646,
"type": "git",
"url": "https://git.m-labs.hk/m-labs/zynq-rs"
},

5
flake.nix
View File

@ -18,11 +18,11 @@
fastnumbers = pkgs.python3Packages.buildPythonPackage rec {
pname = "fastnumbers";
version = "5.1.0";
version = "2.2.1";
src = pkgs.python3Packages.fetchPypi {
inherit pname version;
sha256 = "sha256-4JLTP4uVwxcaL7NOV57+DFSwKQ3X+W/6onYkN2AdkKc=";
sha256 = "0j15i54p7nri6hkzn1wal9pxri4pgql01wgjccig6ar0v5jjbvsy";
};
};
@ -126,7 +126,6 @@
lockFile = src/Cargo.lock;
outputHashes = {
"tar-no-std-0.1.8" = "sha256-xm17108v4smXOqxdLvHl9CxTCJslmeogjm4Y87IXFuM=";
"nalgebra-0.32.6" = "sha256-L/YudkVOtfGYoNQKBD7LMk/sMYgRDzPDdpGL5rO7G2I=";
};
};

77
src/Cargo.lock generated
View File

@ -2,15 +2,6 @@
# It is not intended for manual editing.
version = 3
[[package]]
name = "approx"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cab112f0a86d568ea0e627cc1d6be74a1e9cd55214684db5561995f6dad897c6"
dependencies = [
"num-traits",
]
[[package]]
name = "arrayvec"
version = "0.7.4"
@ -255,7 +246,6 @@ dependencies = [
"libsupport_zynq",
"log",
"log_buffer",
"nalgebra",
"nb 0.1.3",
"unwind",
"vcell",
@ -392,19 +382,6 @@ version = "0.7.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c75de51135344a4f8ed3cfe2720dc27736f7711989703a0b43aadf3753c55577"
[[package]]
name = "nalgebra"
version = "0.32.6"
source = "git+https://git.m-labs.hk/M-Labs/nalgebra.git?rev=dd00f9b#dd00f9b46046e0b931d1b470166db02fd29591be"
dependencies = [
"approx",
"num-complex",
"num-rational",
"num-traits",
"simba",
"typenum",
]
[[package]]
name = "nb"
version = "0.1.3"
@ -420,15 +397,6 @@ version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "546c37ac5d9e56f55e73b677106873d9d9f5190605e41a856503623648488cae"
[[package]]
name = "num-complex"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "26873667bbbb7c5182d4a37c1add32cdf09f841af72da53318fdb81543c15085"
dependencies = [
"num-traits",
]
[[package]]
name = "num-derive"
version = "0.3.3"
@ -440,26 +408,6 @@ dependencies = [
"syn",
]
[[package]]
name = "num-integer"
version = "0.1.46"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7969661fd2958a5cb096e56c8e1ad0444ac2bbcd0061bd28660485a44879858f"
dependencies = [
"num-traits",
]
[[package]]
name = "num-rational"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d41702bd167c2df5520b384281bc111a4b5efcf7fbc4c9c222c815b07e0a6a6a"
dependencies = [
"autocfg",
"num-integer",
"num-traits",
]
[[package]]
name = "num-traits"
version = "0.2.15"
@ -467,15 +415,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "578ede34cf02f8924ab9447f50c28075b4d3e5b269972345e7e0372b38c6cdcd"
dependencies = [
"autocfg",
"libm",
]
[[package]]
name = "paste"
version = "1.0.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a"
[[package]]
name = "pin-project-lite"
version = "0.2.9"
@ -582,18 +523,6 @@ version = "0.1.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d4f410fedcf71af0345d7607d246e7ad15faaadd49d240ee3b24e5dc21a820ac"
[[package]]
name = "simba"
version = "0.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "50582927ed6f77e4ac020c057f37a268fc6aebc29225050365aacbb9deeeddc4"
dependencies = [
"approx",
"num-complex",
"num-traits",
"paste",
]
[[package]]
name = "smoltcp"
version = "0.7.5"
@ -626,12 +555,6 @@ dependencies = [
"log",
]
[[package]]
name = "typenum"
version = "1.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "42ff0bf0c66b8238c6f3b578df37d0b7848e55df8577b3f74f92a69acceeb825"
[[package]]
name = "unicode-ident"
version = "1.0.5"

22
src/libboard_artiq/src/drtioaux_proto.rs
View File

@ -267,14 +267,12 @@ pub enum Packet {
exception_src: u8,
},
SubkernelExceptionRequest {
source: u8,
destination: u8,
},
SubkernelException {
destination: u8,
last: bool,
length: u16,
data: [u8; MASTER_PAYLOAD_MAX_SIZE],
data: [u8; SAT_PAYLOAD_MAX_SIZE],
},
SubkernelMessage {
source: u8,
@ -526,17 +524,14 @@ impl Packet {
exception_src: reader.read_u8()?,
},
0xc9 => Packet::SubkernelExceptionRequest {
source: reader.read_u8()?,
destination: reader.read_u8()?,
},
0xca => {
let destination = reader.read_u8()?;
let last = reader.read_bool()?;
let length = reader.read_u16()?;
let mut data: [u8; MASTER_PAYLOAD_MAX_SIZE] = [0; MASTER_PAYLOAD_MAX_SIZE];
let mut data: [u8; SAT_PAYLOAD_MAX_SIZE] = [0; SAT_PAYLOAD_MAX_SIZE];
reader.read_exact(&mut data[0..length as usize])?;
Packet::SubkernelException {
destination: destination,
last: last,
length: length,
data: data,
@ -901,19 +896,12 @@ impl Packet {
writer.write_bool(with_exception)?;
writer.write_u8(exception_src)?;
}
Packet::SubkernelExceptionRequest { source, destination } => {
Packet::SubkernelExceptionRequest { destination } => {
writer.write_u8(0xc9)?;
writer.write_u8(source)?;
writer.write_u8(destination)?;
}
Packet::SubkernelException {
destination,
last,
length,
data,
} => {
Packet::SubkernelException { last, length, data } => {
writer.write_u8(0xca)?;
writer.write_u8(destination)?;
writer.write_bool(last)?;
writer.write_u16(length)?;
writer.write_all(&data[0..length as usize])?;
@ -955,8 +943,6 @@ impl Packet {
Packet::SubkernelLoadRunReply { destination, .. } => Some(*destination),
Packet::SubkernelMessage { destination, .. } => Some(*destination),
Packet::SubkernelMessageAck { destination } => Some(*destination),
Packet::SubkernelExceptionRequest { destination, .. } => Some(*destination),
Packet::SubkernelException { destination, .. } => Some(*destination),
Packet::DmaPlaybackStatus { destination, .. } => Some(*destination),
Packet::SubkernelFinished { destination, .. } => Some(*destination),
_ => None,

6
src/libksupport/Cargo.toml.tpl
View File

@ -32,9 +32,3 @@ unwind = { path = "../libunwind" }
libc = { path = "../libc" }
io = { path = "../libio" }
libboard_artiq = { path = "../libboard_artiq" }
[dependencies.nalgebra]
git = "https://git.m-labs.hk/M-Labs/nalgebra.git"
rev = "dd00f9b"
default-features = false
features = ["libm", "alloc"]

124
src/libksupport/src/eh_artiq.rs
View File

@ -14,10 +14,8 @@
use core::mem;
use core_io::Error as ReadError;
use cslice::{AsCSlice, CSlice};
use cslice::CSlice;
use dwarf::eh::{self, EHAction, EHContext};
use io::{Cursor, ProtoRead};
use libc::{c_int, c_void, uintptr_t};
use log::{error, trace};
use unwind as uw;
@ -222,6 +220,8 @@ pub unsafe fn artiq_personality(
}
pub unsafe extern "C" fn raise(exception: *const Exception) -> ! {
use cslice::AsCSlice;
let count = EXCEPTION_BUFFER.exception_count;
let stack = &mut EXCEPTION_BUFFER.exception_stack;
let diff = exception as isize - EXCEPTION_BUFFER.exceptions.as_ptr() as isize;
@ -295,60 +295,6 @@ pub unsafe extern "C" fn raise(exception: *const Exception) -> ! {
unreachable!();
}
fn read_exception_string<'a>(reader: &mut Cursor<&[u8]>) -> Result<CSlice<'a, u8>, ReadError> {
let len = reader.read_u32()? as usize;
if len == usize::MAX {
let data = reader.read_u32()?;
Ok(unsafe { CSlice::new(data as *const u8, len) })
} else {
let pos = reader.position();
let slice = unsafe {
let ptr = reader.get_ref().as_ptr().offset(pos as isize);
CSlice::new(ptr, len)
};
reader.set_position(pos + len);
Ok(slice)
}
}
fn read_exception(raw_exception: &[u8]) -> Result<Exception, ReadError> {
let mut reader = Cursor::new(raw_exception);
let mut byte = reader.read_u8()?;
// to sync
while byte != 0x5a {
byte = reader.read_u8()?;
}
// skip sync bytes, 0x09 indicates exception
while byte != 0x09 {
byte = reader.read_u8()?;
}
let _len = reader.read_u32()?;
// ignore the remaining exceptions, stack traces etc. - unwinding from another device would be unwise anyway
Ok(Exception {
id: reader.read_u32()?,
message: read_exception_string(&mut reader)?,
param: [
reader.read_u64()? as i64,
reader.read_u64()? as i64,
reader.read_u64()? as i64,
],
file: read_exception_string(&mut reader)?,
line: reader.read_u32()?,
column: reader.read_u32()?,
function: read_exception_string(&mut reader)?,
})
}
pub fn raise_raw(raw_exception: &[u8]) -> ! {
use crate::artiq_raise;
if let Ok(exception) = read_exception(raw_exception) {
unsafe { raise(&exception) };
} else {
artiq_raise!("SubkernelError", "Error passing exception");
}
}
pub unsafe extern "C" fn resume() -> ! {
trace!("resume");
assert!(EXCEPTION_BUFFER.exception_count != 0);
@ -475,30 +421,20 @@ extern "C" fn stop_fn(
}
}
// Must be kept in sync with preallocate_runtime_exception_names() in `artiq.compiler.embedding`
static EXCEPTION_ID_LOOKUP: [(&str, u32); 22] = [
("RTIOUnderflow", 0),
("RTIOOverflow", 1),
("RTIODestinationUnreachable", 2),
("DMAError", 3),
("I2CError", 4),
("CacheError", 5),
("SPIError", 6),
("SubkernelError", 7),
("AssertionError", 8),
("AttributeError", 9),
("IndexError", 10),
("IOError", 11),
("KeyError", 12),
("NotImplementedError", 13),
("OverflowError", 14),
("RuntimeError", 15),
("TimeoutError", 16),
("TypeError", 17),
("ValueError", 18),
("ZeroDivisionError", 19),
("LinAlgError", 20),
("UnwrapNoneError", 21),
// Must be kept in sync with preallocate_runtime_exception_names() in artiq/language/embedding_map.py
static EXCEPTION_ID_LOOKUP: [(&str, u32); 12] = [
("RuntimeError", 0),
("RTIOUnderflow", 1),
("RTIOOverflow", 2),
("RTIODestinationUnreachable", 3),
("DMAError", 4),
("I2CError", 5),
("CacheError", 6),
("SPIError", 7),
("ZeroDivisionError", 8),
("IndexError", 9),
("UnwrapNoneError", 10),
("SubkernelError", 11),
];
pub fn get_exception_id(name: &str) -> u32 {
@ -533,29 +469,3 @@ macro_rules! artiq_raise {
}};
($name:expr, $message:expr) => {{ artiq_raise!($name, $message, 0, 0, 0) }};
}
/// Takes as input exception id from host
/// Generates a new exception with:
/// * `id` set to `exn_id`
/// * `message` set to corresponding exception name from `EXCEPTION_ID_LOOKUP`
///
/// The message is matched on host to ensure correct exception is being referred
/// This test checks the synchronization of exception ids for runtime errors
#[no_mangle]
pub extern "C" fn test_exception_id_sync(exn_id: u32) {
let message = EXCEPTION_ID_LOOKUP
.iter()
.find_map(|&(name, id)| if id == exn_id { Some(name) } else { None })
.unwrap_or("unallocated internal exception id");
let exn = Exception {
id: exn_id,
file: file!().as_c_slice(),
line: 0,
column: 0,
function: "test_exception_id_sync".as_c_slice(),
message: message.as_c_slice(),
param: [0, 0, 0],
};
unsafe { raise(&exn) };
}

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

@ -9,7 +9,7 @@ use log::{info, warn};
use super::subkernel;
use super::{cache,
core1::rtio_get_destination_status,
dma, linalg,
dma,
rpc::{rpc_recv, rpc_send, rpc_send_async}};
use crate::{eh_artiq, i2c, rtio};
@ -319,26 +319,6 @@ pub fn resolve(required: &[u8]) -> Option<u32> {
}
api!(yn = yn)
},
// linalg
api!(np_linalg_cholesky = linalg::np_linalg_cholesky),
api!(np_linalg_qr = linalg::np_linalg_qr),
api!(np_linalg_svd = linalg::np_linalg_svd),
api!(np_linalg_inv = linalg::np_linalg_inv),
api!(np_linalg_pinv = linalg::np_linalg_pinv),
api!(np_linalg_matrix_power = linalg::np_linalg_matrix_power),
api!(np_linalg_det = linalg::np_linalg_det),
api!(sp_linalg_lu = linalg::sp_linalg_lu),
api!(sp_linalg_schur = linalg::sp_linalg_schur),
api!(sp_linalg_hessenberg = linalg::sp_linalg_hessenberg),
/*
* syscall for unit tests
* Used in `artiq.tests.coredevice.test_exceptions.ExceptionTest.test_raise_exceptions_kernel`
* This syscall checks that the exception IDs used in the Python `EmbeddingMap` (in `artiq.language.embedding`)
* match the `EXCEPTION_ID_LOOKUP` defined in the firmware (`libksupport::src::eh_artiq`)
*/
api!(test_exception_id_sync = eh_artiq::test_exception_id_sync)
];
api.iter()
.find(|&&(exported, _)| exported.as_bytes() == required)

440
src/libksupport/src/kernel/linalg.rs
View File

@ -1,440 +0,0 @@
// Uses `nalgebra` crate to invoke `np_linalg` and `sp_linalg` functions
// When converting between `nalgebra::Matrix` and `NDArray` following considerations are necessary
//
// * Both `nalgebra::Matrix` and `NDArray` require their content to be stored in row-major order
// * `NDArray` data pointer can be directly read and converted to `nalgebra::Matrix` (row and column number must be known)
// * `nalgebra::Matrix::as_slice` returns the content of matrix in column-major order and initial data needs to be transposed before storing it in `NDArray` data pointer
use alloc::vec::Vec;
use core::slice;
use nalgebra::DMatrix;
use crate::artiq_raise;
pub struct InputMatrix {
pub ndims: usize,
pub dims: *const usize,
pub data: *mut f64,
}
impl InputMatrix {
fn get_dims(&mut self) -> Vec<usize> {
let dims = unsafe { slice::from_raw_parts(self.dims, self.ndims) };
dims.to_vec()
}
}
/// # Safety
///
/// `mat1` should point to a valid 2DArray of `f64` floats in row-major order
#[no_mangle]
pub unsafe extern "C" fn np_linalg_cholesky(mat1: *mut InputMatrix, out: *mut InputMatrix) {
let mat1 = mat1.as_mut().unwrap();
let out = out.as_mut().unwrap();
if mat1.ndims != 2 {
artiq_raise!(
"ValueError",
"expected 2D Vector Input, but received {1}D input)",
0,
mat1.ndims as i64,
0
);
}
let dim1 = (*mat1).get_dims();
if dim1[0] != dim1[1] {
artiq_raise!(
"ValueError",
"last 2 dimensions of the array must be square: {1} != {2}",
0,
dim1[0] as i64,
dim1[1] as i64
);
}
let outdim = out.get_dims();
let out_slice = slice::from_raw_parts_mut(out.data, outdim[0] * outdim[1]);
let data_slice1 = slice::from_raw_parts_mut(mat1.data, dim1[0] * dim1[1]);
let matrix1 = DMatrix::from_row_slice(dim1[0], dim1[1], data_slice1);
let result = matrix1.cholesky();
match result {
Some(res) => {
out_slice.copy_from_slice(res.unpack().transpose().as_slice());
}
None => {
artiq_raise!("LinAlgError", "Matrix is not positive definite");
}
};
}
/// # Safety
///
/// `mat1` should point to a valid 2DArray of `f64` floats in row-major order
#[no_mangle]
pub unsafe extern "C" fn np_linalg_qr(mat1: *mut InputMatrix, out_q: *mut InputMatrix, out_r: *mut InputMatrix) {
let mat1 = mat1.as_mut().unwrap();
let out_q = out_q.as_mut().unwrap();
let out_r = out_r.as_mut().unwrap();
if mat1.ndims != 2 {
artiq_raise!(
"ValueError",
"expected 2D Vector Input, but received {1}D input)",
0,
mat1.ndims as i64,
0
);
}
let dim1 = (*mat1).get_dims();
let outq_dim = (*out_q).get_dims();
let outr_dim = (*out_r).get_dims();
let data_slice1 = slice::from_raw_parts_mut(mat1.data, dim1[0] * dim1[1]);
let out_q_slice = slice::from_raw_parts_mut(out_q.data, outq_dim[0] * outq_dim[1]);
let out_r_slice = slice::from_raw_parts_mut(out_r.data, outr_dim[0] * outr_dim[1]);
// Refer to https://github.com/dimforge/nalgebra/issues/735
let matrix1 = DMatrix::from_row_slice(dim1[0], dim1[1], data_slice1);
let res = matrix1.qr();
let (q, r) = res.unpack();
// Uses different algo need to match numpy
out_q_slice.copy_from_slice(q.transpose().as_slice());
out_r_slice.copy_from_slice(r.transpose().as_slice());
}
/// # Safety
///
/// `mat1` should point to a valid 2DArray of `f64` floats in row-major order
#[no_mangle]
pub unsafe extern "C" fn np_linalg_svd(
mat1: *mut InputMatrix,
outu: *mut InputMatrix,
outs: *mut InputMatrix,
outvh: *mut InputMatrix,
) {
let mat1 = mat1.as_mut().unwrap();
let outu = outu.as_mut().unwrap();
let outs = outs.as_mut().unwrap();
let outvh = outvh.as_mut().unwrap();
if mat1.ndims != 2 {
artiq_raise!(
"ValueError",
"expected 2D Vector Input, but received {1}D input)",
0,
mat1.ndims as i64,
0
);
}
let dim1 = (*mat1).get_dims();
let outu_dim = (*outu).get_dims();
let outs_dim = (*outs).get_dims();
let outvh_dim = (*outvh).get_dims();
let data_slice1 = slice::from_raw_parts_mut(mat1.data, dim1[0] * dim1[1]);
let out_u_slice = slice::from_raw_parts_mut(outu.data, outu_dim[0] * outu_dim[1]);
let out_s_slice = slice::from_raw_parts_mut(outs.data, outs_dim[0]);
let out_vh_slice = slice::from_raw_parts_mut(outvh.data, outvh_dim[0] * outvh_dim[1]);
let matrix = DMatrix::from_row_slice(dim1[0], dim1[1], data_slice1);
let result = matrix.svd(true, true);
out_u_slice.copy_from_slice(result.u.unwrap().transpose().as_slice());
out_s_slice.copy_from_slice(result.singular_values.as_slice());
out_vh_slice.copy_from_slice(result.v_t.unwrap().transpose().as_slice());
}
/// # Safety
///
/// `mat1` should point to a valid 2DArray of `f64` floats in row-major order
#[no_mangle]
pub unsafe extern "C" fn np_linalg_inv(mat1: *mut InputMatrix, out: *mut InputMatrix) {
let mat1 = mat1.as_mut().unwrap();
let out = out.as_mut().unwrap();
if mat1.ndims != 2 {
artiq_raise!(
"ValueError",
"expected 2D Vector Input, but received {1}D input)",
0,
mat1.ndims as i64,
0
);
}
let dim1 = (*mat1).get_dims();
if dim1[0] != dim1[1] {
artiq_raise!(
"ValueError",
"last 2 dimensions of the array must be square: {1} != {2}",
0,
dim1[0] as i64,
dim1[1] as i64
);
}
let outdim = out.get_dims();
let out_slice = slice::from_raw_parts_mut(out.data, outdim[0] * outdim[1]);
let data_slice1 = slice::from_raw_parts_mut(mat1.data, dim1[0] * dim1[1]);
let matrix = DMatrix::from_row_slice(dim1[0], dim1[1], data_slice1);
if !matrix.is_invertible() {
artiq_raise!("LinAlgError", "no inverse for Singular Matrix");
}
let inv = matrix.try_inverse().unwrap();
out_slice.copy_from_slice(inv.transpose().as_slice());
}
/// # Safety
///
/// `mat1` should point to a valid 2DArray of `f64` floats in row-major order
#[no_mangle]
pub unsafe extern "C" fn np_linalg_pinv(mat1: *mut InputMatrix, out: *mut InputMatrix) {
let mat1 = mat1.as_mut().unwrap();
let out = out.as_mut().unwrap();
if mat1.ndims != 2 {
artiq_raise!(
"ValueError",
"expected 2D Vector Input, but received {1}D input)",
0,
mat1.ndims as i64,
0
);
}
let dim1 = (*mat1).get_dims();
let outdim = out.get_dims();
let out_slice = slice::from_raw_parts_mut(out.data, outdim[0] * outdim[1]);
let data_slice1 = slice::from_raw_parts_mut(mat1.data, dim1[0] * dim1[1]);
let matrix = DMatrix::from_row_slice(dim1[0], dim1[1], data_slice1);
let svd = matrix.svd(true, true);
let inv = svd.pseudo_inverse(1e-15);
match inv {
Ok(m) => {
out_slice.copy_from_slice(m.transpose().as_slice());
}
Err(_) => {
artiq_raise!("LinAlgError", "SVD computation does not converge");
}
}
}
/// # Safety
///
/// `mat1` should point to a valid 2DArray of `f64` floats in row-major order
#[no_mangle]
pub unsafe extern "C" fn np_linalg_matrix_power(mat1: *mut InputMatrix, mat2: *mut InputMatrix, out: *mut InputMatrix) {
let mat1 = mat1.as_mut().unwrap();
let mat2 = mat2.as_mut().unwrap();
let out = out.as_mut().unwrap();
if mat1.ndims != 2 {
artiq_raise!(
"ValueError",
"expected 2D Vector Input, but received {1}D input)",
0,
mat1.ndims as i64,
0
);
}
let dim1 = (*mat1).get_dims();
let power = slice::from_raw_parts_mut(mat2.data, 1);
let power = power[0];
let outdim = out.get_dims();
let out_slice = slice::from_raw_parts_mut(out.data, outdim[0] * outdim[1]);
let data_slice1 = slice::from_raw_parts_mut(mat1.data, dim1[0] * dim1[1]);
let mut abs_power = power;
if abs_power < 0.0 {
abs_power = abs_power * -1.0;
}
let matrix1 = DMatrix::from_row_slice(dim1[0], dim1[1], data_slice1);
if !matrix1.is_square() {
artiq_raise!(
"ValueError",
"last 2 dimensions of the array must be square: {1} != {2}",
0,
dim1[0] as i64,
dim1[1] as i64
);
}
let mut result = matrix1.pow(abs_power as u32);
if power < 0.0 {
if !matrix1.is_invertible() {
artiq_raise!("LinAlgError", "no inverse for Singular Matrix");
}
result = result.try_inverse().unwrap();
}
out_slice.copy_from_slice(result.transpose().as_slice());
}
/// # Safety
///
/// `mat1` should point to a valid 2DArray of `f64` floats in row-major order
#[no_mangle]
pub unsafe extern "C" fn np_linalg_det(mat1: *mut InputMatrix, out: *mut InputMatrix) {
let mat1 = mat1.as_mut().unwrap();
let out = out.as_mut().unwrap();
if mat1.ndims != 2 {
artiq_raise!(
"ValueError",
"expected 2D Vector Input, but received {1}D input)",
0,
mat1.ndims as i64,
0
);
}
let dim1 = (*mat1).get_dims();
let out_slice = slice::from_raw_parts_mut(out.data, 1);
let data_slice1 = slice::from_raw_parts_mut(mat1.data, dim1[0] * dim1[1]);
let matrix = DMatrix::from_row_slice(dim1[0], dim1[1], data_slice1);
if !matrix.is_square() {
artiq_raise!(
"ValueError",
"last 2 dimensions of the array must be square: {1} != {2}",
0,
dim1[0] as i64,
dim1[1] as i64
);
}
out_slice[0] = matrix.determinant();
}
/// # Safety
///
/// `mat1` should point to a valid 2DArray of `f64` floats in row-major order
#[no_mangle]
pub unsafe extern "C" fn sp_linalg_lu(mat1: *mut InputMatrix, out_l: *mut InputMatrix, out_u: *mut InputMatrix) {
let mat1 = mat1.as_mut().unwrap();
let out_l = out_l.as_mut().unwrap();
let out_u = out_u.as_mut().unwrap();
if mat1.ndims != 2 {
artiq_raise!(
"ValueError",
"expected 2D Vector Input, but received {1}D input)",
0,
mat1.ndims as i64,
0
);
}
let dim1 = (*mat1).get_dims();
let outl_dim = (*out_l).get_dims();
let outu_dim = (*out_u).get_dims();
let data_slice1 = slice::from_raw_parts_mut(mat1.data, dim1[0] * dim1[1]);
let out_l_slice = slice::from_raw_parts_mut(out_l.data, outl_dim[0] * outl_dim[1]);
let out_u_slice = slice::from_raw_parts_mut(out_u.data, outu_dim[0] * outu_dim[1]);
let matrix = DMatrix::from_row_slice(dim1[0], dim1[1], data_slice1);
let (_, l, u) = matrix.lu().unpack();
out_l_slice.copy_from_slice(l.transpose().as_slice());
out_u_slice.copy_from_slice(u.transpose().as_slice());
}
/// # Safety
///
/// `mat1` should point to a valid 2DArray of `f64` floats in row-major order
#[no_mangle]
pub unsafe extern "C" fn sp_linalg_schur(mat1: *mut InputMatrix, out_t: *mut InputMatrix, out_z: *mut InputMatrix) {
let mat1 = mat1.as_mut().unwrap();
let out_t = out_t.as_mut().unwrap();
let out_z = out_z.as_mut().unwrap();
if mat1.ndims != 2 {
artiq_raise!(
"ValueError",
"expected 2D Vector Input, but received {1}D input)",
0,
mat1.ndims as i64,
0
);
}
let dim1 = (*mat1).get_dims();
if dim1[0] != dim1[1] {
artiq_raise!(
"ValueError",
"last 2 dimensions of the array must be square: {1} != {2}",
0,
dim1[0] as i64,
dim1[1] as i64
);
}
let out_t_dim = (*out_t).get_dims();
let out_z_dim = (*out_z).get_dims();
let data_slice1 = slice::from_raw_parts_mut(mat1.data, dim1[0] * dim1[1]);
let out_t_slice = slice::from_raw_parts_mut(out_t.data, out_t_dim[0] * out_t_dim[1]);
let out_z_slice = slice::from_raw_parts_mut(out_z.data, out_z_dim[0] * out_z_dim[1]);
let matrix = DMatrix::from_row_slice(dim1[0], dim1[1], data_slice1);
let (z, t) = matrix.schur().unpack();
out_t_slice.copy_from_slice(t.transpose().as_slice());
out_z_slice.copy_from_slice(z.transpose().as_slice());
}
/// # Safety
///
/// `mat1` should point to a valid 2DArray of `f64` floats in row-major order
#[no_mangle]
pub unsafe extern "C" fn sp_linalg_hessenberg(
mat1: *mut InputMatrix,
out_h: *mut InputMatrix,
out_q: *mut InputMatrix,
) {
let mat1 = mat1.as_mut().unwrap();
let out_h = out_h.as_mut().unwrap();
let out_q = out_q.as_mut().unwrap();
if mat1.ndims != 2 {
artiq_raise!(
"ValueError",
"expected 2D Vector Input, but received {1}D input)",
0,
mat1.ndims as i64,
0
);
}
let dim1 = (*mat1).get_dims();
if dim1[0] != dim1[1] {
artiq_raise!(
"ValueError",
"last 2 dimensions of the array must be square: {1} != {2}",
0,
dim1[0] as i64,
dim1[1] as i64
);
}
let out_h_dim = (*out_h).get_dims();
let out_q_dim = (*out_q).get_dims();
let data_slice1 = slice::from_raw_parts_mut(mat1.data, dim1[0] * dim1[1]);
let out_h_slice = slice::from_raw_parts_mut(out_h.data, out_h_dim[0] * out_h_dim[1]);
let out_q_slice = slice::from_raw_parts_mut(out_q.data, out_q_dim[0] * out_q_dim[1]);
let matrix = DMatrix::from_row_slice(dim1[0], dim1[1], data_slice1);
let (q, h) = matrix.hessenberg().unpack();
out_h_slice.copy_from_slice(h.transpose().as_slice());
out_q_slice.copy_from_slice(q.transpose().as_slice());
}

9
src/libksupport/src/kernel/mod.rs
View File

@ -13,7 +13,6 @@ mod dma;
mod rpc;
pub use dma::DmaRecorder;
mod cache;
mod linalg;
#[cfg(has_drtio)]
mod subkernel;
@ -24,7 +23,6 @@ pub enum SubkernelStatus {
Timeout,
IncorrectState,
CommLost,
Exception(Vec<u8>),
OtherError,
}
@ -92,7 +90,9 @@ pub enum Message {
timeout: i64,
},
#[cfg(has_drtio)]
SubkernelAwaitFinishReply,
SubkernelAwaitFinishReply {
status: SubkernelStatus,
},
#[cfg(has_drtio)]
SubkernelMsgSend {
id: u32,
@ -109,10 +109,9 @@ pub enum Message {
},
#[cfg(has_drtio)]
SubkernelMsgRecvReply {
status: SubkernelStatus,
count: u8,
},
#[cfg(has_drtio)]
SubkernelError(SubkernelStatus),
}
static CHANNEL_0TO1: Mutex<Option<sync_channel::Sender<'static, Message>>> = Mutex::new(None);

61
src/libksupport/src/kernel/subkernel.rs
View File

@ -3,7 +3,7 @@ use alloc::vec::Vec;
use cslice::CSlice;
use super::{Message, SubkernelStatus, KERNEL_CHANNEL_0TO1, KERNEL_CHANNEL_1TO0};
use crate::{artiq_raise, eh_artiq, rpc::send_args};
use crate::{artiq_raise, rpc::send_args};
pub extern "C" fn load_run(id: u32, destination: u8, run: bool) {
unsafe {
@ -36,18 +36,21 @@ pub extern "C" fn await_finish(id: u32, timeout: i64) {
});
}
match unsafe { KERNEL_CHANNEL_0TO1.as_mut().unwrap() }.recv() {
Message::SubkernelAwaitFinishReply => (),
Message::SubkernelError(SubkernelStatus::IncorrectState) => {
artiq_raise!("SubkernelError", "Subkernel not running")
}
Message::SubkernelError(SubkernelStatus::Timeout) => artiq_raise!("SubkernelError", "Subkernel timed out"),
Message::SubkernelError(SubkernelStatus::CommLost) => {
artiq_raise!("SubkernelError", "Lost communication with satellite")
}
Message::SubkernelError(SubkernelStatus::OtherError) => {
artiq_raise!("SubkernelError", "An error occurred during subkernel operation")
}
Message::SubkernelError(SubkernelStatus::Exception(raw_exception)) => eh_artiq::raise_raw(&raw_exception),
Message::SubkernelAwaitFinishReply {
status: SubkernelStatus::NoError,
} => (),
Message::SubkernelAwaitFinishReply {
status: SubkernelStatus::IncorrectState,
} => artiq_raise!("SubkernelError", "Subkernel not running"),
Message::SubkernelAwaitFinishReply {
status: SubkernelStatus::Timeout,
} => artiq_raise!("SubkernelError", "Subkernel timed out"),
Message::SubkernelAwaitFinishReply {
status: SubkernelStatus::CommLost,
} => artiq_raise!("SubkernelError", "Lost communication with satellite"),
Message::SubkernelAwaitFinishReply {
status: SubkernelStatus::OtherError,
} => artiq_raise!("SubkernelError", "An error occurred during subkernel operation"),
_ => panic!("expected SubkernelAwaitFinishReply after SubkernelAwaitFinishRequest"),
}
}
@ -89,22 +92,30 @@ pub extern "C" fn await_message(id: i32, timeout: i64, tags: &CSlice<u8>, min: u
});
}
match unsafe { KERNEL_CHANNEL_0TO1.as_mut().unwrap() }.recv() {
Message::SubkernelMsgRecvReply { count } => {
Message::SubkernelMsgRecvReply {
status: SubkernelStatus::NoError,
count,
} => {
if min > count || count > max {
artiq_raise!("SubkernelError", "Received more or less arguments than required")
}
}
Message::SubkernelError(SubkernelStatus::IncorrectState) => {
artiq_raise!("SubkernelError", "Subkernel not running")
}
Message::SubkernelError(SubkernelStatus::Timeout) => artiq_raise!("SubkernelError", "Subkernel timed out"),
Message::SubkernelError(SubkernelStatus::CommLost) => {
artiq_raise!("SubkernelError", "Lost communication with satellite")
}
Message::SubkernelError(SubkernelStatus::OtherError) => {
artiq_raise!("SubkernelError", "An error occurred during subkernel operation")
}
Message::SubkernelError(SubkernelStatus::Exception(raw_exception)) => eh_artiq::raise_raw(&raw_exception),
Message::SubkernelMsgRecvReply {
status: SubkernelStatus::IncorrectState,
..
} => artiq_raise!("SubkernelError", "Subkernel not running"),
Message::SubkernelMsgRecvReply {
status: SubkernelStatus::Timeout,
..
} => artiq_raise!("SubkernelError", "Subkernel timed out"),
Message::SubkernelMsgRecvReply {
status: SubkernelStatus::CommLost,
..
} => artiq_raise!("SubkernelError", "Lost communication with satellite"),
Message::SubkernelMsgRecvReply {
status: SubkernelStatus::OtherError,
..
} => artiq_raise!("SubkernelError", "An error occurred during subkernel operation"),
_ => panic!("expected SubkernelMsgRecvReply after SubkernelMsgRecvRequest"),
}
// RpcRecvRequest should be called after this to receive message data

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

@ -422,23 +422,33 @@ async fn handle_run_kernel(
#[cfg(has_drtio)]
kernel::Message::SubkernelAwaitFinishRequest { id, timeout } => {
let res = subkernel::await_finish(aux_mutex, routing_table, timer, id, timeout).await;
let response = match res {
Ok(res) => {
let status = match res {
Ok(ref res) => {
if res.status == subkernel::FinishStatus::CommLost {
kernel::Message::SubkernelError(kernel::SubkernelStatus::CommLost)
} else if let Some(exception) = res.exception {
kernel::Message::SubkernelError(kernel::SubkernelStatus::Exception(exception))
kernel::SubkernelStatus::CommLost
} else if let Some(exception) = &res.exception {
error!("Exception in subkernel");
match stream {
None => (),
Some(stream) => {
write_chunk(stream, exception).await?;
}
}
// will not be called after exception is served
kernel::SubkernelStatus::OtherError
} else {
kernel::Message::SubkernelAwaitFinishReply
kernel::SubkernelStatus::NoError
}
}
Err(SubkernelError::Timeout) => kernel::Message::SubkernelError(kernel::SubkernelStatus::Timeout),
Err(SubkernelError::IncorrectState) => {
kernel::Message::SubkernelError(kernel::SubkernelStatus::IncorrectState)
}
Err(_) => kernel::Message::SubkernelError(kernel::SubkernelStatus::OtherError),
Err(SubkernelError::Timeout) => kernel::SubkernelStatus::Timeout,
Err(SubkernelError::IncorrectState) => kernel::SubkernelStatus::IncorrectState,
Err(_) => kernel::SubkernelStatus::OtherError,
};
control.borrow_mut().tx.async_send(response).await;
control
.borrow_mut()
.tx
.async_send(kernel::Message::SubkernelAwaitFinishReply { status: status })
.await;
}
#[cfg(has_drtio)]
kernel::Message::SubkernelMsgSend { id, destination, data } => {
@ -459,23 +469,35 @@ async fn handle_run_kernel(
#[cfg(has_drtio)]
kernel::Message::SubkernelMsgRecvRequest { id, timeout, tags } => {
let message_received = subkernel::message_await(id as u32, timeout, timer).await;
let response = match message_received {
Ok(ref message) => kernel::Message::SubkernelMsgRecvReply { count: message.count },
Err(SubkernelError::Timeout) => kernel::Message::SubkernelError(kernel::SubkernelStatus::Timeout),
Err(SubkernelError::IncorrectState) => {
kernel::Message::SubkernelError(kernel::SubkernelStatus::IncorrectState)
}
Err(SubkernelError::CommLost) => kernel::Message::SubkernelError(kernel::SubkernelStatus::CommLost),
let (status, count) = match message_received {
Ok(ref message) => (kernel::SubkernelStatus::NoError, message.count),
Err(SubkernelError::Timeout) => (kernel::SubkernelStatus::Timeout, 0),
Err(SubkernelError::IncorrectState) => (kernel::SubkernelStatus::IncorrectState, 0),
Err(SubkernelError::CommLost) => (kernel::SubkernelStatus::CommLost, 0),
Err(SubkernelError::SubkernelException) => {
error!("Exception in subkernel");
// just retrieve the exception
let status = subkernel::await_finish(aux_mutex, routing_table, timer, id as u32, timeout)
.await
.unwrap();
kernel::Message::SubkernelError(kernel::SubkernelStatus::Exception(status.exception.unwrap()))
match stream {
None => (),
Some(stream) => {
write_chunk(stream, &status.exception.unwrap()).await?;
}
}
(kernel::SubkernelStatus::OtherError, 0)
}
Err(_) => kernel::Message::SubkernelError(kernel::SubkernelStatus::OtherError),
Err(_) => (kernel::SubkernelStatus::OtherError, 0),
};
control.borrow_mut().tx.async_send(response).await;
control
.borrow_mut()
.tx
.async_send(kernel::Message::SubkernelMsgRecvReply {
status: status,
count: count,
})
.await;
if let Ok(message) = message_received {
// receive code almost identical to RPC recv, except we are not reading from a stream
let mut reader = Cursor::new(message.data);
@ -507,7 +529,7 @@ async fn handle_run_kernel(
.async_send(kernel::Message::RpcRecvReply(Ok(0)))
.await;
i += 1;
if i < message.count {
if i < count {
current_tags = remaining_tags;
} else {
break;
@ -756,6 +778,7 @@ pub fn main(timer: GlobalTimer, cfg: Config) {
moninj::start(timer, &aux_mutex, &drtio_routing_table);
let control: Rc<RefCell<kernel::Control>> = Rc::new(RefCell::new(kernel::Control::start()));
let idle_kernel = Rc::new(cfg.read("idle_kernel").ok());
if let Ok(buffer) = cfg.read("startup_kernel") {
info!("Loading startup kernel...");
let routing_table = drtio_routing_table.borrow();
@ -781,27 +804,14 @@ pub fn main(timer: GlobalTimer, cfg: Config) {
error!("Error loading startup kernel!");
}
}
let cfg = Rc::new(cfg);
let restart_idle = Rc::new(Semaphore::new(1, 1));
mgmt::start(cfg.clone(), restart_idle.clone());
mgmt::start(cfg);
task::spawn(async move {
let connection = Rc::new(Semaphore::new(1, 1));
let terminate = Rc::new(Semaphore::new(0, 1));
let can_restart_idle = Rc::new(Semaphore::new(1, 1));
let restart_idle = restart_idle.clone();
loop {
let control = control.clone();
let mut maybe_stream = select_biased! {
s = (async {
TcpStream::accept(1381, 0x10_000, 0x10_000).await.unwrap()
}).fuse() => Some(s),
_ = (async {
restart_idle.async_wait().await;
can_restart_idle.async_wait().await;
}).fuse() => None
};
let mut stream = TcpStream::accept(1381, 0x10_000, 0x10_000).await.unwrap();
if connection.try_wait().is_none() {
// there is an existing connection
@ -809,58 +819,47 @@ pub fn main(timer: GlobalTimer, cfg: Config) {
connection.async_wait().await;
}
let maybe_idle_kernel = cfg.read("idle_kernel").ok();
if maybe_idle_kernel.is_none() && maybe_stream.is_none() {
control.borrow_mut().restart(); // terminate idle kernel if running
}
let control = control.clone();
let idle_kernel = idle_kernel.clone();
let connection = connection.clone();
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
let _ = terminate.try_wait();
let _ = can_restart_idle.try_wait();
task::spawn(async move {
let routing_table = routing_table.borrow();
select_biased! {
_ = (async {
if let Some(stream) = &mut maybe_stream {
let _ = handle_connection(stream, control.clone(), &up_destinations, &aux_mutex, &routing_table, timer)
.await
.map_err(|e| warn!("connection terminated: {}", e));
}
can_restart_idle.signal();
match maybe_idle_kernel {
Some(buffer) => {
loop {
info!("loading idle kernel");
match handle_flash_kernel(&buffer, &control, &up_destinations, &aux_mutex, &routing_table, timer).await {
Ok(_) => {
info!("running idle kernel");
match handle_run_kernel(None, &control, &up_destinations, &aux_mutex, &routing_table, timer).await {
Ok(_) => info!("idle kernel finished"),
Err(_) => warn!("idle kernel running error")
}
},
Err(_) => warn!("idle kernel loading error")
}
let _ = handle_connection(&mut stream, control.clone(), &up_destinations, &aux_mutex, &routing_table, timer)
.await
.map_err(|e| warn!("connection terminated: {}", e));
if let Some(buffer) = &*idle_kernel {
info!("Loading idle kernel");
let res = handle_flash_kernel(&buffer, &control, &up_destinations, &aux_mutex, &routing_table, timer)
.await;
match res {
#[cfg(has_drtio)]
Err(Error::DestinationDown) => {
let mut countdown = timer.countdown();
delay(&mut countdown, Milliseconds(500)).await;
}
},
None => info!("no idle kernel found")
Err(_) => warn!("error loading idle kernel"),
_ => (),
}
info!("Running idle kernel");
let _ = handle_run_kernel(None, &control, &up_destinations, &aux_mutex, &routing_table, timer)
.await.map_err(|_| warn!("error running idle kernel"));
info!("Idle kernel terminated");
}
}).fuse() => (),
_ = terminate.async_wait().fuse() => ()
}
connection.signal();
if let Some(stream) = maybe_stream {
let _ = stream.flush().await;
let _ = stream.abort().await;
}
let _ = stream.flush().await;
let _ = stream.abort().await;
});
}
});
@ -908,9 +907,8 @@ pub fn soft_panic_main(timer: GlobalTimer, cfg: Config) -> ! {
};
Sockets::init(32);
let dummy = Rc::new(Semaphore::new(0, 1));
mgmt::start(Rc::new(cfg), dummy);
mgmt::start(cfg);
// getting eth settings disables the LED as it resets GPIO
// need to re-enable it here

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

@ -6,7 +6,6 @@ use libasync::{smoltcp::TcpStream, task};
use libboard_artiq::logger::{BufferLogger, LogBufferRef};
use libboard_zynq::{slcr, smoltcp};
use libconfig::Config;
use libcortex_a9::semaphore::Semaphore;
use log::{self, debug, error, info, warn, LevelFilter};
use num_derive::FromPrimitive;
use num_traits::FromPrimitive;
@ -112,7 +111,7 @@ async fn read_key(stream: &mut TcpStream) -> Result<String> {
Ok(String::from_utf8(buffer).unwrap())
}
async fn handle_connection(stream: &mut TcpStream, pull_id: Rc<RefCell<u32>>, cfg: Rc<Config>, restart_idle: Rc<Semaphore>) -> Result<()> {
async fn handle_connection(stream: &mut TcpStream, pull_id: Rc<RefCell<u32>>, cfg: Rc<Config>) -> Result<()> {
if !expect(&stream, b"ARTIQ management\n").await? {
return Err(Error::UnexpectedPattern);
}
@ -201,9 +200,6 @@ async fn handle_connection(stream: &mut TcpStream, pull_id: Rc<RefCell<u32>>, cf
let value = cfg.write(&key, buffer);
if value.is_ok() {
debug!("write success");
if key == "idle_kernel" {
restart_idle.signal();
}
write_i8(stream, Reply::Success as i8).await?;
} else {
// this is an error because we do not expect write to fail
@ -217,9 +213,6 @@ async fn handle_connection(stream: &mut TcpStream, pull_id: Rc<RefCell<u32>>, cf
let value = cfg.remove(&key);
if value.is_ok() {
debug!("erase success");
if key == "idle_kernel" {
restart_idle.signal();
}
write_i8(stream, Reply::Success as i8).await?;
} else {
warn!("erase failed");
@ -236,17 +229,17 @@ async fn handle_connection(stream: &mut TcpStream, pull_id: Rc<RefCell<u32>>, cf
}
}
pub fn start(cfg: Rc<Config>, restart_idle: Rc<Semaphore>) {
pub fn start(cfg: Config) {
task::spawn(async move {
let pull_id = Rc::new(RefCell::new(0u32));
let cfg = Rc::new(cfg);
loop {
let mut stream = TcpStream::accept(1380, 2048, 2048).await.unwrap();
let pull_id = pull_id.clone();
let cfg = cfg.clone();
let restart_idle = restart_idle.clone();
task::spawn(async move {
info!("received connection");
let _ = handle_connection(&mut stream, pull_id, cfg, restart_idle)
let _ = handle_connection(&mut stream, pull_id, cfg)
.await
.map_err(|e| warn!("connection terminated: {:?}", e));
let _ = stream.flush().await;

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

@ -102,7 +102,7 @@ mod remote_moninj {
overrd: i8,
value: i8,
) {
let _lock = aux_mutex.async_lock().await;
let _lock = aux_mutex.lock();
drtioaux_async::send(
linkno,
&drtioaux_async::Packet::InjectionRequest {

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

@ -129,8 +129,6 @@ pub mod drtio {
| Packet::SubkernelLoadRunReply { destination, .. }
| Packet::SubkernelMessage { destination, .. }
| Packet::SubkernelMessageAck { destination, .. }
| Packet::SubkernelException { destination, .. }
| Packet::SubkernelExceptionRequest { destination, .. }
| Packet::DmaPlaybackStatus { destination, .. }
| Packet::SubkernelFinished { destination, .. } => {
if destination == 0 {
@ -185,7 +183,10 @@ pub mod drtio {
async fn drain_buffer(linkno: u8, draining_time: Milliseconds, timer: GlobalTimer) {
let max_time = timer.get_time() + draining_time;
while timer.get_time() < max_time {
loop {
if timer.get_time() > max_time {
return;
}
let _ = drtioaux_async::recv(linkno).await;
}
}
@ -834,19 +835,13 @@ pub mod drtio {
linkno,
routing_table,
&Packet::SubkernelExceptionRequest {
source: 0,
destination: destination,
},
timer,
)
.await?;
match reply {
Packet::SubkernelException {
destination: 0,
last,
length,
data,
} => {
Packet::SubkernelException { last, length, data } => {
remote_data.extend(&data[0..length as usize]);
if last {
return Ok(remote_data);

381
src/satman/src/main.rs
View File

@ -91,29 +91,13 @@ fn toggle_sed_spread(val: u8) {
#[cfg(has_drtio_routing)]
macro_rules! forward {
(
$router:expr,
$routing_table:expr,
$destination:expr,
$rank:expr,
$self_destination:expr,
$repeaters:expr,
$packet:expr,
$timer:expr
) => {{
($routing_table:expr, $destination:expr, $rank:expr, $repeaters:expr, $packet:expr, $timer:expr) => {{
let hop = $routing_table.0[$destination as usize][$rank as usize];
if hop != 0 {
let repno = (hop - 1) as usize;
if repno < $repeaters.len() {
if $packet.expects_response() {
return $repeaters[repno].aux_forward(
$packet,
$router,
$routing_table,
$rank,
$self_destination,
$timer,
);
return $repeaters[repno].aux_forward($packet, $timer);
} else {
return $repeaters[repno].aux_send($packet);
}
@ -126,16 +110,7 @@ macro_rules! forward {
#[cfg(not(has_drtio_routing))]
macro_rules! forward {
(
$router:expr,
$routing_table:expr,
$destination:expr,
$rank:expr,
$self_destination:expr,
$repeaters:expr,
$packet:expr,
$timer:expr
) => {};
($routing_table:expr, $destination:expr, $rank:expr, $repeaters:expr, $packet:expr, $timer:expr) => {};
}
fn process_aux_packet(
@ -216,10 +191,6 @@ fn process_aux_packet(
&drtioaux::Packet::DestinationStatusRequest {
destination: destination,
},
router,
_routing_table,
*rank,
*self_destination,
timer,
) {
Ok(()) => (),
@ -281,16 +252,7 @@ fn process_aux_packet(
channel,
probe,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
let value;
#[cfg(has_rtio_moninj)]
unsafe {
@ -312,16 +274,7 @@ fn process_aux_packet(
overrd,
value,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
#[cfg(has_rtio_moninj)]
unsafe {
csr::rtio_moninj::inj_chan_sel_write(channel as _);
@ -335,16 +288,7 @@ fn process_aux_packet(
channel,
overrd,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
let value;
#[cfg(has_rtio_moninj)]
unsafe {
@ -363,16 +307,7 @@ fn process_aux_packet(
destination: _destination,
busno: _busno,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
let succeeded = i2c.start().is_ok();
drtioaux::send(0, &drtioaux::Packet::I2cBasicReply { succeeded: succeeded })
}
@ -380,16 +315,7 @@ fn process_aux_packet(
destination: _destination,
busno: _busno,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
let succeeded = i2c.restart().is_ok();
drtioaux::send(0, &drtioaux::Packet::I2cBasicReply { succeeded: succeeded })
}
@ -397,16 +323,7 @@ fn process_aux_packet(
destination: _destination,
busno: _busno,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
let succeeded = i2c.stop().is_ok();
drtioaux::send(0, &drtioaux::Packet::I2cBasicReply { succeeded: succeeded })
}
@ -415,16 +332,7 @@ fn process_aux_packet(
busno: _busno,
data,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
match i2c.write(data) {
Ok(ack) => drtioaux::send(
0,
@ -447,16 +355,7 @@ fn process_aux_packet(
busno: _busno,
ack,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
match i2c.read(ack) {
Ok(data) => drtioaux::send(
0,
@ -480,16 +379,7 @@ fn process_aux_packet(
address,
mask,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
let ch = match mask {
//decode from mainline, PCA9548-centric API
0x00 => None,
@ -515,16 +405,7 @@ fn process_aux_packet(
div: _div,
cs: _cs,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
// todo: reimplement when/if SPI is available
//let succeeded = spi::set_config(busno, flags, length, div, cs).is_ok();
drtioaux::send(0, &drtioaux::Packet::SpiBasicReply { succeeded: false })
@ -534,16 +415,7 @@ fn process_aux_packet(
busno: _busno,
data: _data,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
// todo: reimplement when/if SPI is available
//let succeeded = spi::write(busno, data).is_ok();
drtioaux::send(0, &drtioaux::Packet::SpiBasicReply { succeeded: false })
@ -552,16 +424,7 @@ fn process_aux_packet(
destination: _destination,
busno: _busno,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
// todo: reimplement when/if SPI is available
// match spi::read(busno) {
// Ok(data) => drtioaux::send(0,
@ -581,16 +444,7 @@ fn process_aux_packet(
drtioaux::Packet::AnalyzerHeaderRequest {
destination: _destination,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
let header = analyzer.get_header();
drtioaux::send(
0,
@ -604,16 +458,7 @@ fn process_aux_packet(
drtioaux::Packet::AnalyzerDataRequest {
destination: _destination,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
let mut data_slice: [u8; SAT_PAYLOAD_MAX_SIZE] = [0; SAT_PAYLOAD_MAX_SIZE];
let meta = analyzer.get_data(&mut data_slice);
drtioaux::send(
@ -634,16 +479,7 @@ fn process_aux_packet(
length,
trace,
} => {
forward!(
router,
_routing_table,
destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, destination, *rank, _repeaters, &packet, timer);
*self_destination = destination;
let succeeded = dma_manager.add(source, id, status, &trace, length as usize).is_ok();
router.send(
@ -664,16 +500,7 @@ fn process_aux_packet(
id,
succeeded,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
dma_manager.ack_upload(
kernel_manager,
source,
@ -691,16 +518,7 @@ fn process_aux_packet(
destination: _destination,
id,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
let succeeded = dma_manager.erase(source, id).is_ok();
router.send(
drtioaux::Packet::DmaRemoveTraceReply {
@ -716,16 +534,7 @@ fn process_aux_packet(
destination: _destination,
succeeded: _,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
Ok(())
}
drtioaux::Packet::DmaPlaybackRequest {
@ -734,16 +543,7 @@ fn process_aux_packet(
id,
timestamp,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
let succeeded = if !kernel_manager.running() {
dma_manager.playback(source, id, timestamp).is_ok()
} else {
@ -763,16 +563,7 @@ fn process_aux_packet(
destination: _destination,
succeeded,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
if !succeeded {
kernel_manager.ddma_nack();
}
@ -786,16 +577,7 @@ fn process_aux_packet(
channel,
timestamp,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
dma_manager.remote_finished(kernel_manager, id, error, channel, timestamp);
Ok(())
}
@ -807,16 +589,7 @@ fn process_aux_packet(
length,
data,
} => {
forward!(
router,
_routing_table,
destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, destination, *rank, _repeaters, &packet, timer);
*self_destination = destination;
let succeeded = kernel_manager.add(id, status, &data, length as usize).is_ok();
drtioaux::send(0, &drtioaux::Packet::SubkernelAddDataReply { succeeded: succeeded })
@ -827,16 +600,7 @@ fn process_aux_packet(
id,
run,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
let mut succeeded = kernel_manager.load(id).is_ok();
// allow preloading a kernel with delayed run
if run {
@ -861,16 +625,7 @@ fn process_aux_packet(
destination: _destination,
succeeded,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
// received if local subkernel started another, remote subkernel
kernel_manager.subkernel_load_run_reply(succeeded);
Ok(())
@ -881,73 +636,25 @@ fn process_aux_packet(
with_exception,
exception_src,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
kernel_manager.remote_subkernel_finished(id, with_exception, exception_src);
Ok(())
}
drtioaux::Packet::SubkernelExceptionRequest {
source,
destination: _destination,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
let mut data_slice: [u8; MASTER_PAYLOAD_MAX_SIZE] = [0; MASTER_PAYLOAD_MAX_SIZE];
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
let mut data_slice: [u8; SAT_PAYLOAD_MAX_SIZE] = [0; SAT_PAYLOAD_MAX_SIZE];
let meta = kernel_manager.exception_get_slice(&mut data_slice);
router.send(
drtioaux::Packet::SubkernelException {
destination: source,
drtioaux::send(
0,
&drtioaux::Packet::SubkernelException {
last: meta.status.is_last(),
length: meta.len,
data: data_slice,
},
_routing_table,
*rank,
*self_destination,
)
}
drtioaux::Packet::SubkernelException {
destination: _destination,
last,
length,
data,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
kernel_manager.received_exception(
&data[..length as usize],
last,
router,
_routing_table,
*rank,
*self_destination,
);
Ok(())
}
drtioaux::Packet::SubkernelMessage {
source,
destination: _destination,
@ -956,16 +663,7 @@ fn process_aux_packet(
length,
data,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
kernel_manager.message_handle_incoming(status, id, length as usize, &data);
router.send(
drtioaux::Packet::SubkernelMessageAck { destination: source },
@ -977,16 +675,7 @@ fn process_aux_packet(
drtioaux::Packet::SubkernelMessageAck {
destination: _destination,
} => {
forward!(
router,
_routing_table,
_destination,
*rank,
*self_destination,
_repeaters,
&packet,
timer
);
forward!(_routing_table, _destination, *rank, _repeaters, &packet, timer);
if kernel_manager.message_ack_slice() {
let mut data_slice: [u8; MASTER_PAYLOAD_MAX_SIZE] = [0; MASTER_PAYLOAD_MAX_SIZE];
if let Some(meta) = kernel_manager.message_get_slice(&mut data_slice) {

33
src/satman/src/repeater.rs
View File

@ -208,37 +208,10 @@ impl Repeater {
}
}
pub fn aux_forward(
&self,
request: &drtioaux::Packet,
router: &mut Router,
routing_table: &drtio_routing::RoutingTable,
rank: u8,
self_destination: u8,
timer: &mut GlobalTimer,
) -> Result<(), drtioaux::Error> {
pub fn aux_forward(&self, request: &drtioaux::Packet, timer: &mut GlobalTimer) -> Result<(), drtioaux::Error> {
self.aux_send(request)?;
loop {
let reply = self.recv_aux_timeout(200, timer)?;
match reply {
// async/locally requested packets to be consumed or routed
// these may come while a packet would be forwarded
drtioaux::Packet::DmaPlaybackStatus { .. }
| drtioaux::Packet::SubkernelFinished { .. }
| drtioaux::Packet::SubkernelMessage { .. }
| drtioaux::Packet::SubkernelMessageAck { .. }
| drtioaux::Packet::SubkernelLoadRunReply { .. }
| drtioaux::Packet::SubkernelException { .. }
| drtioaux::Packet::DmaAddTraceReply { .. }
| drtioaux::Packet::DmaPlaybackReply { .. } => {
router.route(reply, routing_table, rank, self_destination);
}
_ => {
drtioaux::send(0, &reply).unwrap();
break;
}
}
}
let reply = self.recv_aux_timeout(200, timer)?;
drtioaux::send(0, &reply).unwrap();
Ok(())
}

3
src/satman/src/routing.rs
View File

@ -4,7 +4,7 @@ use core::cmp::min;
#[cfg(has_drtio_routing)]
use libboard_artiq::pl::csr;
use libboard_artiq::{drtio_routing, drtioaux,
drtioaux_proto::{PayloadStatus, MASTER_PAYLOAD_MAX_SIZE}};
drtioaux_proto::{PayloadStatus, MASTER_PAYLOAD_MAX_SIZE, SAT_PAYLOAD_MAX_SIZE}};
pub struct SliceMeta {
pub destination: u8,
@ -57,6 +57,7 @@ impl Sliceable {
self.data.extend(data);
}
get_slice_fn!(get_slice_sat, SAT_PAYLOAD_MAX_SIZE);
get_slice_fn!(get_slice_master, MASTER_PAYLOAD_MAX_SIZE);
}

146
src/satman/src/subkernel.rs
View File

@ -11,7 +11,7 @@ use io::{Cursor, ProtoWrite};
use ksupport::{eh_artiq, kernel, rpc};
use libboard_artiq::{drtio_routing::RoutingTable,
drtioaux,
drtioaux_proto::{PayloadStatus, MASTER_PAYLOAD_MAX_SIZE},
drtioaux_proto::{PayloadStatus, MASTER_PAYLOAD_MAX_SIZE, SAT_PAYLOAD_MAX_SIZE},
pl::csr};
use libboard_zynq::{time::Milliseconds, timer::GlobalTimer};
use libcortex_a9::sync_channel::Receiver;
@ -47,9 +47,6 @@ enum KernelState {
DmaAwait {
max_time: Milliseconds,
},
SubkernelRetrievingException {
destination: u8,
},
}
#[derive(Debug)]
@ -126,11 +123,10 @@ struct MessageManager {
struct Session {
id: u32,
kernel_state: KernelState,
last_exception: Option<Sliceable>, // exceptions raised locally
external_exception: Option<Vec<u8>>, // exceptions from sub-subkernels
last_exception: Option<Sliceable>,
messages: MessageManager,
source: u8, // which destination requested running the kernel
subkernels_finished: Vec<(u32, Option<u8>)>,
subkernels_finished: Vec<u32>,
}
impl Session {
@ -139,7 +135,6 @@ impl Session {
id: id,
kernel_state: KernelState::Absent,
last_exception: None,
external_exception: None,
messages: MessageManager::new(),
source: 0,
subkernels_finished: Vec::new(),
@ -415,9 +410,9 @@ impl<'a> Manager<'_> {
}
}
pub fn exception_get_slice(&mut self, data_slice: &mut [u8; MASTER_PAYLOAD_MAX_SIZE]) -> SliceMeta {
pub fn exception_get_slice(&mut self, data_slice: &mut [u8; SAT_PAYLOAD_MAX_SIZE]) -> SliceMeta {
match self.session.last_exception.as_mut() {
Some(exception) => exception.get_slice_master(data_slice),
Some(exception) => exception.get_slice_sat(data_slice),
None => SliceMeta {
destination: 0,
len: 0,
@ -545,7 +540,7 @@ impl<'a> Manager<'_> {
return;
}
match self.process_external_messages(router, routing_table, rank, destination, timer) {
match self.process_external_messages(timer) {
Ok(()) => (),
Err(Error::AwaitingMessage) => return, // kernel still waiting, do not process kernel messages
Err(Error::KernelException(exception)) => {
@ -601,41 +596,6 @@ impl<'a> Manager<'_> {
}
}
fn check_finished_kernels(
&mut self,
id: u32,
router: &mut Router,
routing_table: &RoutingTable,
rank: u8,
self_destination: u8,
) {
for (i, (status, exception_source)) in self.session.subkernels_finished.iter().enumerate() {
if *status == id {
if exception_source.is_none() {
self.control.tx.send(kernel::Message::SubkernelAwaitFinishReply);
self.session.kernel_state = KernelState::Running;
self.session.subkernels_finished.swap_remove(i);
} else {
let destination = exception_source.unwrap();
self.session.external_exception = Some(Vec::new());
self.session.kernel_state = KernelState::SubkernelRetrievingException {
destination: destination,
};
router.route(
drtioaux::Packet::SubkernelExceptionRequest {
source: self_destination,
destination: destination,
},
&routing_table,
rank,
self_destination,
);
}
break;
}
}
}
pub fn subkernel_load_run_reply(&mut self, succeeded: bool) {
if self.session.kernel_state == KernelState::SubkernelAwaitLoad {
self.control
@ -648,46 +608,16 @@ impl<'a> Manager<'_> {
}
pub fn remote_subkernel_finished(&mut self, id: u32, with_exception: bool, exception_source: u8) {
let exception_src = if with_exception { Some(exception_source) } else { None };
self.session.subkernels_finished.push((id, exception_src));
}
pub fn received_exception(
&mut self,
exception_data: &[u8],
last: bool,
router: &mut Router,
routing_table: &RoutingTable,
rank: u8,
self_destination: u8,
) {
if let KernelState::SubkernelRetrievingException { destination } = self.session.kernel_state {
self.session
.external_exception
.as_mut()
.unwrap()
.extend_from_slice(exception_data);
if last {
self.control
.tx
.send(kernel::Message::SubkernelError(kernel::SubkernelStatus::Exception(
self.session.external_exception.take().unwrap(),
)));
self.session.kernel_state = KernelState::Running;
} else {
/* fetch another slice */
router.route(
drtioaux::Packet::SubkernelExceptionRequest {
source: self_destination,
destination: destination,
},
routing_table,
rank,
self_destination,
);
}
if with_exception {
self.kernel_stop();
self.last_finished = Some(SubkernelFinished {
source: self.session.source,
id: self.session.id,
with_exception: true,
exception_source: exception_source,
})
} else {
warn!("Received unsolicited exception data");
self.session.subkernels_finished.push(id);
}
}
@ -850,35 +780,28 @@ impl<'a> Manager<'_> {
Ok(false)
}
fn process_external_messages(
&mut self,
router: &mut Router,
routing_table: &RoutingTable,
rank: u8,
self_destination: u8,
timer: &GlobalTimer,
) -> Result<(), Error> {
fn process_external_messages(&mut self, timer: &GlobalTimer) -> Result<(), Error> {
match &self.session.kernel_state {
KernelState::MsgAwait { max_time, id, tags } => {
if let Some(max_time) = *max_time {
if timer.get_time() > max_time {
self.control
.tx
.send(kernel::Message::SubkernelError(kernel::SubkernelStatus::Timeout));
self.control.tx.send(kernel::Message::SubkernelMsgRecvReply {
status: kernel::SubkernelStatus::Timeout,
count: 0,
});
self.session.kernel_state = KernelState::Running;
return Ok(());
}
}
if let Some(message) = self.session.messages.get_incoming(*id) {
self.control
.tx
.send(kernel::Message::SubkernelMsgRecvReply { count: message.count });
self.control.tx.send(kernel::Message::SubkernelMsgRecvReply {
status: kernel::SubkernelStatus::NoError,
count: message.count,
});
let tags = tags.clone();
self.session.kernel_state = KernelState::Running;
self.pass_message_to_kernel(&message, tags, timer)
} else {
let id = *id;
self.check_finished_kernels(id, router, routing_table, rank, self_destination);
Err(Error::AwaitingMessage)
}
}
@ -894,18 +817,27 @@ impl<'a> Manager<'_> {
KernelState::SubkernelAwaitFinish { max_time, id } => {
if let Some(max_time) = *max_time {
if timer.get_time() > max_time {
self.control
.tx
.send(kernel::Message::SubkernelError(kernel::SubkernelStatus::Timeout));
self.control.tx.send(kernel::Message::SubkernelAwaitFinishReply {
status: kernel::SubkernelStatus::Timeout,
});
self.session.kernel_state = KernelState::Running;
return Ok(());
}
}
let id = *id;
self.check_finished_kernels(id, router, routing_table, rank, self_destination);
let mut i = 0;
for status in &self.session.subkernels_finished {
if *status == *id {
self.control.tx.send(kernel::Message::SubkernelAwaitFinishReply {
status: kernel::SubkernelStatus::NoError,
});
self.session.kernel_state = KernelState::Running;
self.session.subkernels_finished.swap_remove(i);
break;
}
i += 1;
}
Ok(())
}
KernelState::SubkernelRetrievingException { .. } => Err(Error::AwaitingMessage),
KernelState::DmaAwait { max_time } | KernelState::DmaPendingAwait { max_time, .. } => {
if timer.get_time() > *max_time {
self.control.tx.send(kernel::Message::DmaAwaitRemoteReply {