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

subkernel: separate tags and data

Browse Source
This commit is contained in:
mwojcik 2023-10-18 11:56:38 +08:00
parent d3fbfd75b0
commit 0a3bfc9a61
8 changed files with 77 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -103,6 +103,7 @@ pub enum Message {
SubkernelMsgRecvRequest { SubkernelMsgRecvRequest {
id: u32, id: u32,
timeout: u64, timeout: u64,
tags: Vec<u8>,
}, },
#[cfg(has_drtio)] #[cfg(has_drtio)]
SubkernelMsgRecvReply { SubkernelMsgRecvReply {

2
src/libksupport/src/kernel/rpc.rs
View File

@ -10,7 +10,7 @@ use crate::{eh_artiq, rpc::send_args};
fn rpc_send_common(is_async: bool, service: u32, tag: &CSlice<u8>, data: *const *const ()) { fn rpc_send_common(is_async: bool, service: u32, tag: &CSlice<u8>, data: *const *const ()) {
let core1_tx = unsafe { KERNEL_CHANNEL_1TO0.as_mut().unwrap() }; let core1_tx = unsafe { KERNEL_CHANNEL_1TO0.as_mut().unwrap() };
let mut buffer = Vec::<u8>::new(); let mut buffer = Vec::<u8>::new();
send_args(&mut buffer, service, tag.as_ref(), data).expect("RPC encoding failed"); send_args(&mut buffer, service, tag.as_ref(), data, true).expect("RPC encoding failed");
core1_tx.send(Message::RpcSend { is_async, data: buffer }); core1_tx.send(Message::RpcSend { is_async, data: buffer });
} }

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

@ -53,7 +53,7 @@ pub extern "C" fn await_finish(id: u32, timeout: u64) {
pub extern "C" fn send_message(id: u32, count: u8, tag: &CSlice<u8>, data: *const *const ()) { pub extern "C" fn send_message(id: u32, count: u8, tag: &CSlice<u8>, data: *const *const ()) {
let mut buffer = Vec::<u8>::new(); let mut buffer = Vec::<u8>::new();
send_args(&mut buffer, 0, tag.as_ref(), data).expect("RPC encoding failed"); send_args(&mut buffer, 0, tag.as_ref(), data, false).expect("RPC encoding failed");
// overwrite service tag, include how many tags are in the message // overwrite service tag, include how many tags are in the message
buffer[3] = count; buffer[3] = count;
unsafe { unsafe {
@ -68,7 +68,7 @@ pub extern "C" fn send_message(id: u32, count: u8, tag: &CSlice<u8>, data: *cons
} }
} }
pub extern "C" fn await_message(id: u32, timeout: u64, min: u8, max: u8) { pub extern "C" fn await_message(id: u32, timeout: u64, tags: &CSlice<u8>, min: u8, max: u8) {
unsafe { unsafe {
KERNEL_CHANNEL_1TO0 KERNEL_CHANNEL_1TO0
.as_mut() .as_mut()
@ -76,6 +76,7 @@ pub extern "C" fn await_message(id: u32, timeout: u64, min: u8, max: u8) {
.send(Message::SubkernelMsgRecvRequest { .send(Message::SubkernelMsgRecvRequest {
id: id, id: id,
timeout: timeout, timeout: timeout,
tags: tags.as_ref().to_vec(),
}); });
} }
match unsafe { KERNEL_CHANNEL_0TO1.as_mut().unwrap() }.recv() { match unsafe { KERNEL_CHANNEL_0TO1.as_mut().unwrap() }.recv() {

63
src/libksupport/src/rpc.rs
View File

@ -175,7 +175,12 @@ where
} }
} }
pub fn recv_return<F, R>(reader: &mut R, tag_bytes: &[u8], data: *mut (), alloc: &mut F) -> Result<(), Error> pub fn recv_return<'a, F, R>(
reader: &mut R,
tag_bytes: &'a [u8],
data: *mut (),
alloc: &mut F,
) -> Result<&'a [u8], Error>
where where
F: FnMut(usize) -> *mut (), F: FnMut(usize) -> *mut (),
R: Read + ?Sized, R: Read + ?Sized,
@ -187,12 +192,22 @@ where
let mut data = data; let mut data = data;
unsafe { recv_value(reader, tag, &mut data, alloc)? }; unsafe { recv_value(reader, tag, &mut data, alloc)? };
Ok(()) Ok(it.data)
} }
unsafe fn send_elements<W>(writer: &mut W, elt_tag: Tag, length: usize, data: *const ()) -> Result<(), Error> unsafe fn send_elements<W>(
where W: Write + ?Sized { writer: &mut W,
elt_tag: Tag,
length: usize,
data: *const (),
write_tags: bool,
) -> Result<(), Error>
where
W: Write + ?Sized,
{
if write_tags {
writer.write_u8(elt_tag.as_u8())?; writer.write_u8(elt_tag.as_u8())?;
}
match elt_tag { match elt_tag {
// we cannot use NativeEndian::from_slice_i32 as the data is not mutable, // we cannot use NativeEndian::from_slice_i32 as the data is not mutable,
// and that is not needed as the data is already in native endian // and that is not needed as the data is already in native endian
@ -211,14 +226,14 @@ where W: Write + ?Sized {
_ => { _ => {
let mut data = data; let mut data = data;
for _ in 0..length { for _ in 0..length {
send_value(writer, elt_tag, &mut data)?; send_value(writer, elt_tag, &mut data, write_tags)?;
} }
} }
} }
Ok(()) Ok(())
} }
unsafe fn send_value<W>(writer: &mut W, tag: Tag, data: &mut *const ()) -> Result<(), Error> unsafe fn send_value<W>(writer: &mut W, tag: Tag, data: &mut *const (), write_tags: bool) -> Result<(), Error>
where W: Write + ?Sized { where W: Write + ?Sized {
macro_rules! consume_value { macro_rules! consume_value {
($ty:ty, | $ptr:ident | $map:expr) => {{ ($ty:ty, | $ptr:ident | $map:expr) => {{
@ -228,7 +243,9 @@ where W: Write + ?Sized {
}}; }};
} }
if write_tags {
writer.write_u8(tag.as_u8())?; writer.write_u8(tag.as_u8())?;
}
match tag { match tag {
Tag::None => Ok(()), Tag::None => Ok(()),
Tag::Bool => consume_value!(u8, |ptr| writer.write_u8(*ptr)), Tag::Bool => consume_value!(u8, |ptr| writer.write_u8(*ptr)),
@ -240,12 +257,14 @@ where W: Write + ?Sized {
Tag::Bytes | Tag::ByteArray => consume_value!(CSlice<u8>, |ptr| writer.write_bytes((*ptr).as_ref())), Tag::Bytes | Tag::ByteArray => consume_value!(CSlice<u8>, |ptr| writer.write_bytes((*ptr).as_ref())),
Tag::Tuple(it, arity) => { Tag::Tuple(it, arity) => {
let mut it = it.clone(); let mut it = it.clone();
if write_tags {
writer.write_u8(arity)?; writer.write_u8(arity)?;
}
let mut max_alignment = 0; let mut max_alignment = 0;
for _ in 0..arity { for _ in 0..arity {
let tag = it.next().expect("truncated tag"); let tag = it.next().expect("truncated tag");
max_alignment = core::cmp::max(max_alignment, tag.alignment()); max_alignment = core::cmp::max(max_alignment, tag.alignment());
send_value(writer, tag, data)? send_value(writer, tag, data, write_tags)?
} }
*data = round_up_const(*data, max_alignment); *data = round_up_const(*data, max_alignment);
Ok(()) Ok(())
@ -260,11 +279,13 @@ where W: Write + ?Sized {
let length = (**ptr).length as usize; let length = (**ptr).length as usize;
writer.write_u32((*ptr).length)?; writer.write_u32((*ptr).length)?;
let tag = it.clone().next().expect("truncated tag"); let tag = it.clone().next().expect("truncated tag");
send_elements(writer, tag, length, (**ptr).elements) send_elements(writer, tag, length, (**ptr).elements, write_tags)
}) })
} }
Tag::Array(it, num_dims) => { Tag::Array(it, num_dims) => {
if write_tags {
writer.write_u8(num_dims)?; writer.write_u8(num_dims)?;
}
consume_value!(*const (), |buffer| { consume_value!(*const (), |buffer| {
let elt_tag = it.clone().next().expect("truncated tag"); let elt_tag = it.clone().next().expect("truncated tag");
@ -276,14 +297,14 @@ where W: Write + ?Sized {
}) })
} }
let length = total_len as usize; let length = total_len as usize;
send_elements(writer, elt_tag, length, *buffer) send_elements(writer, elt_tag, length, *buffer, write_tags)
}) })
} }
Tag::Range(it) => { Tag::Range(it) => {
let tag = it.clone().next().expect("truncated tag"); let tag = it.clone().next().expect("truncated tag");
send_value(writer, tag, data)?; send_value(writer, tag, data, write_tags)?;
send_value(writer, tag, data)?; send_value(writer, tag, data, write_tags)?;
send_value(writer, tag, data)?; send_value(writer, tag, data, write_tags)?;
Ok(()) Ok(())
} }
Tag::Keyword(it) => { Tag::Keyword(it) => {
@ -295,7 +316,7 @@ where W: Write + ?Sized {
writer.write_string(str::from_utf8((*ptr).name.as_ref()).unwrap())?; writer.write_string(str::from_utf8((*ptr).name.as_ref()).unwrap())?;
let tag = it.clone().next().expect("truncated tag"); let tag = it.clone().next().expect("truncated tag");
let mut data = ptr.offset(1) as *const (); let mut data = ptr.offset(1) as *const ();
send_value(writer, tag, &mut data) send_value(writer, tag, &mut data, write_tags)
}) })
// Tag::Keyword never appears in composite types, so we don't have // Tag::Keyword never appears in composite types, so we don't have
// to accurately advance data. // to accurately advance data.
@ -310,8 +331,16 @@ where W: Write + ?Sized {
} }
} }
pub fn send_args<W>(writer: &mut W, service: u32, tag_bytes: &[u8], data: *const *const ()) -> Result<(), Error> pub fn send_args<W>(
where W: Write + ?Sized { writer: &mut W,
service: u32,
tag_bytes: &[u8],
data: *const *const (),
write_tags: bool,
) -> Result<(), Error>
where
W: Write + ?Sized,
{
let (arg_tags_bytes, return_tag_bytes) = split_tag(tag_bytes); let (arg_tags_bytes, return_tag_bytes) = split_tag(tag_bytes);
let mut args_it = TagIterator::new(arg_tags_bytes); let mut args_it = TagIterator::new(arg_tags_bytes);
@ -322,7 +351,7 @@ where W: Write + ?Sized {
for index in 0.. { for index in 0.. {
if let Some(arg_tag) = args_it.next() { if let Some(arg_tag) = args_it.next() {
let mut data = unsafe { *data.offset(index) }; let mut data = unsafe { *data.offset(index) };
unsafe { send_value(writer, arg_tag, &mut data)? }; unsafe { send_value(writer, arg_tag, &mut data, write_tags)? };
} else { } else {
break; break;
} }
@ -450,7 +479,7 @@ pub mod tag {
#[derive(Debug, Clone, Copy)] #[derive(Debug, Clone, Copy)]
pub struct TagIterator<'a> { pub struct TagIterator<'a> {
data: &'a [u8], pub data: &'a [u8],
} }
impl<'a> TagIterator<'a> { impl<'a> TagIterator<'a> {

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

@ -5,7 +5,7 @@ use core_io::Error as IoError;
use cslice::CSlice; use cslice::CSlice;
use futures::{future::FutureExt, select_biased}; use futures::{future::FutureExt, select_biased};
#[cfg(has_drtio)] #[cfg(has_drtio)]
use io::{Cursor, ProtoRead}; use io::Cursor;
#[cfg(has_drtio)] #[cfg(has_drtio)]
use ksupport::rpc; use ksupport::rpc;
use ksupport::{kernel, resolve_channel_name}; use ksupport::{kernel, resolve_channel_name};
@ -451,7 +451,7 @@ async fn handle_run_kernel(
.await; .await;
} }
#[cfg(has_drtio)] #[cfg(has_drtio)]
kernel::Message::SubkernelMsgRecvRequest { id, timeout } => { kernel::Message::SubkernelMsgRecvRequest { id, timeout, tags } => {
let message_received = subkernel::message_await(id, timeout, timer).await; let message_received = subkernel::message_await(id, timeout, timer).await;
let (status, count) = match message_received { let (status, count) = match message_received {
Ok(ref message) => (kernel::SubkernelStatus::NoError, message.count), Ok(ref message) => (kernel::SubkernelStatus::NoError, message.count),
@ -471,7 +471,7 @@ async fn handle_run_kernel(
if let Ok(message) = message_received { if let Ok(message) = message_received {
// receive code almost identical to RPC recv, except we are not reading from a stream // receive code almost identical to RPC recv, except we are not reading from a stream
let mut reader = Cursor::new(message.data); let mut reader = Cursor::new(message.data);
let mut tag: [u8; 1] = [message.tag]; let mut current_tags: &[u8] = &tags;
let mut i = 0; let mut i = 0;
loop { loop {
// kernel has to consume all arguments in the whole message // kernel has to consume all arguments in the whole message
@ -479,7 +479,7 @@ async fn handle_run_kernel(
kernel::Message::RpcRecvRequest(slot) => slot, kernel::Message::RpcRecvRequest(slot) => slot,
other => panic!("expected root value slot from core1, not {:?}", other), other => panic!("expected root value slot from core1, not {:?}", other),
}; };
rpc::recv_return(&mut reader, &tag, slot, &mut |size| { let remaining_tags = rpc::recv_return(&mut reader, &current_tags, slot, &mut |size| {
if size == 0 { if size == 0 {
0 as *mut () 0 as *mut ()
} else { } else {
@ -500,7 +500,7 @@ async fn handle_run_kernel(
.await; .await;
i += 1; i += 1;
if i < count { if i < count {
tag[0] = reader.read_u8()?; current_tags = remaining_tags;
} else { } else {
break; break;
} }

1
src/runtime/src/main.rs
View File

@ -62,6 +62,7 @@ mod grabber {
use libasync::delay; use libasync::delay;
use libboard_artiq::grabber; use libboard_artiq::grabber;
use libboard_zynq::time::Milliseconds; use libboard_zynq::time::Milliseconds;
use crate::GlobalTimer; use crate::GlobalTimer;
pub async fn grabber_thread(timer: GlobalTimer) { pub async fn grabber_thread(timer: GlobalTimer) {
let mut countdown = timer.countdown(); let mut countdown = timer.countdown();

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

@ -212,7 +212,6 @@ pub async fn await_finish(
pub struct Message { pub struct Message {
from_id: u32, from_id: u32,
pub count: u8, pub count: u8,
pub tag: u8,
pub data: Vec<u8>, pub data: Vec<u8>,
} }
@ -236,8 +235,7 @@ pub async fn message_handle_incoming(id: u32, last: bool, length: usize, data: &
Message { Message {
from_id: id, from_id: id,
count: data[0], count: data[0],
tag: data[1], data: data[1..length].to_vec(),
data: data[2..length].to_vec(),
}, },
); );
} }

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

@ -6,7 +6,7 @@ use core::{cmp::min, option::NoneError, slice, str};
use core_io::{Error as IoError, Write}; use core_io::{Error as IoError, Write};
use cslice::AsCSlice; use cslice::AsCSlice;
use io::{Cursor, ProtoRead, ProtoWrite}; use io::{Cursor, ProtoWrite};
use ksupport::{eh_artiq, kernel, rpc}; use ksupport::{eh_artiq, kernel, rpc};
use libboard_artiq::{drtioaux_proto::{MASTER_PAYLOAD_MAX_SIZE, SAT_PAYLOAD_MAX_SIZE}, use libboard_artiq::{drtioaux_proto::{MASTER_PAYLOAD_MAX_SIZE, SAT_PAYLOAD_MAX_SIZE},
pl::csr}; pl::csr};
@ -14,12 +14,12 @@ use libboard_zynq::{time::Milliseconds, timer::GlobalTimer};
use libcortex_a9::sync_channel::Receiver; use libcortex_a9::sync_channel::Receiver;
use log::warn; use log::warn;
#[derive(Debug, Clone, Copy, PartialEq)] #[derive(Debug, Clone, PartialEq)]
enum KernelState { enum KernelState {
Absent, Absent,
Loaded, Loaded,
Running, Running,
MsgAwait(Milliseconds), MsgAwait(Milliseconds, Vec<u8>),
MsgSending, MsgSending,
} }
@ -66,7 +66,6 @@ pub struct Sliceable {
/* represents interkernel messages */ /* represents interkernel messages */
struct Message { struct Message {
count: u8, count: u8,
tag: u8,
data: Vec<u8>, data: Vec<u8>,
} }
@ -183,8 +182,7 @@ impl MessageManager {
None => { None => {
self.in_buffer = Some(Message { self.in_buffer = Some(Message {
count: data[0], count: data[0],
tag: data[1], data: data[1..length].to_vec(),
data: data[2..length].to_vec(),
}); });
} }
}; };
@ -509,9 +507,9 @@ impl<'a> Manager<'_> {
self.session.messages.accept_outgoing(data)?; self.session.messages.accept_outgoing(data)?;
self.session.kernel_state = KernelState::MsgSending; self.session.kernel_state = KernelState::MsgSending;
} }
kernel::Message::SubkernelMsgRecvRequest { id: _, timeout } => { kernel::Message::SubkernelMsgRecvRequest { id: _, timeout, tags } => {
let max_time = timer.get_time() + Milliseconds(timeout); let max_time = timer.get_time() + Milliseconds(timeout);
self.session.kernel_state = KernelState::MsgAwait(max_time); self.session.kernel_state = KernelState::MsgAwait(max_time, tags);
} }
kernel::Message::UpDestinationsRequest(destination) => { kernel::Message::UpDestinationsRequest(destination) => {
self.control self.control
@ -526,9 +524,9 @@ impl<'a> Manager<'_> {
} }
fn process_external_messages(&mut self, timer: GlobalTimer) -> Result<(), Error> { fn process_external_messages(&mut self, timer: GlobalTimer) -> Result<(), Error> {
match self.session.kernel_state { match &self.session.kernel_state {
KernelState::MsgAwait(timeout) => { KernelState::MsgAwait(timeout, tags) => {
if timer.get_time() > timeout { if timer.get_time() > *timeout {
self.control.tx.send(kernel::Message::SubkernelMsgRecvReply { self.control.tx.send(kernel::Message::SubkernelMsgRecvReply {
status: kernel::SubkernelStatus::Timeout, status: kernel::SubkernelStatus::Timeout,
count: 0, count: 0,
@ -541,8 +539,9 @@ impl<'a> Manager<'_> {
status: kernel::SubkernelStatus::NoError, status: kernel::SubkernelStatus::NoError,
count: message.count, count: message.count,
}); });
let tags = tags.clone();
self.session.kernel_state = KernelState::Running; self.session.kernel_state = KernelState::Running;
self.pass_message_to_kernel(&message, timer) self.pass_message_to_kernel(&message, tags, timer)
} else { } else {
Err(Error::AwaitingMessage) Err(Error::AwaitingMessage)
} }
@ -560,9 +559,9 @@ impl<'a> Manager<'_> {
} }
} }
fn pass_message_to_kernel(&mut self, message: &Message, timer: GlobalTimer) -> Result<(), Error> { fn pass_message_to_kernel(&mut self, message: &Message, tags: Vec<u8>, timer: GlobalTimer) -> Result<(), Error> {
let mut reader = Cursor::new(&message.data); let mut reader = Cursor::new(&message.data);
let mut tag: [u8; 1] = [message.tag]; let mut current_tags: &[u8] = &tags;
let mut i = message.count; let mut i = message.count;
loop { loop {
let slot = match recv_w_timeout(&mut self.control.rx, timer, 100)? { let slot = match recv_w_timeout(&mut self.control.rx, timer, 100)? {
@ -571,7 +570,7 @@ impl<'a> Manager<'_> {
}; };
let mut exception: Option<Sliceable> = None; let mut exception: Option<Sliceable> = None;
let mut unexpected: Option<String> = None; let mut unexpected: Option<String> = None;
rpc::recv_return(&mut reader, &tag, slot, &mut |size| { let remaining_tags = rpc::recv_return(&mut reader, current_tags, slot, &mut |size| {
if size == 0 { if size == 0 {
0 as *mut () 0 as *mut ()
} else { } else {
@ -610,8 +609,7 @@ impl<'a> Manager<'_> {
if i == 0 { if i == 0 {
break; break;
} else { } else {
// update the tag for next read current_tags = remaining_tags;
tag[0] = reader.read_u8()?;
} }
} }
Ok(()) Ok(())