saymrwulf/xous-core
1
0
Fork 0
mirror of https://github.com/saymrwulf/xous-core.git synced 2026-11-08 04:56:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

add first draft of bulk serialize/deserialize functions

Browse source
This commit is contained in:
bunnie 2022-10-26 05:29:44 +08:00
parent 2e768dd19e
commit bffe92cab9
3 changed files with 391 additions and 44 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

41
services/pddb/src/api.rs
View file

@ -3,6 +3,7 @@ pub use rkyv_enum::*;
use bitfield::bitfield;
use std::num::NonZeroU32;
use core::ops::{Deref, DerefMut};
// on the "[allow(dead_code)]" directives: these constants are used to define the PDDB, and are
// sometimes used by both `bin` (main.rs) and `lib` (lib.rs) views, but also, sometimes used
@ -296,6 +297,8 @@ pub struct PddbDictRequest {
pub index: u32,
pub token: [u32; 4],
pub code: PddbRequestCode,
/// used only to specify a readback size limit for bulk-return requests
pub bulk_limit: Option<usize>,
}
/// A structure for requesting a token to access a particular key/value pair
@ -311,8 +314,6 @@ pub struct PddbKeyRequest {
pub alloc_hint: Option<u64>, // this is a usize but for IPC we must have defined memory sizes, so we pick the big option.
pub cb_sid: Option<[u32; 4]>,
pub result: PddbRequestCode,
/// used only to specify a readback size limit for bulk-return requests
pub bulk_limit: Option<u64>,
}
pub(crate) const MAX_PDDBKLISTLEN: usize = 4064;
@ -378,14 +379,46 @@ pub struct PddbKeyRecord {
/// Return codes for Read/Write API calls to the main server
#[repr(u32)]
#[derive(num_derive::FromPrimitive, num_derive::ToPrimitive, Debug, rkyv::Archive, rkyv::Serialize, rkyv::Deserialize)]
pub(crate) enum PddbBulkReadCode {
pub enum PddbBulkReadCode {
/// Uninitialized value
Uninit = 0,
/// call already in progress, try again later
/// call already in progress or timed out (e.g. token does not match what's on record)
Busy = 1,
/// last return data structure
Last = 2,
/// return data structure, with more data to come
Streaming = 3,
/// dictionary not found
NotFound = 4,
/// PDDB internal error
InternalError = 5,
}
#[derive(Default, Debug)]
#[repr(C)]
pub struct BulkReadHeader {
pub code: u32,
pub starting_key_index: u32,
pub len: u32,
pub total: u32,
pub token: [u32; 4],
}
impl Deref for BulkReadHeader {
type Target = [u8];
fn deref(&self) -> &[u8] {
unsafe {
core::slice::from_raw_parts(self as *const BulkReadHeader as *const u8, core::mem::size_of::<BulkReadHeader>())
as &[u8]
}
}
}
impl DerefMut for BulkReadHeader {
fn deref_mut(&mut self) -> &mut [u8] {
unsafe {
core::slice::from_raw_parts_mut(self as *mut BulkReadHeader as *mut u8, core::mem::size_of::<BulkReadHeader>())
as &mut [u8]
}
}
}
#[allow(dead_code)]

158
services/pddb/src/lib.rs
View file

@ -20,6 +20,17 @@ pub(crate) static REFCOUNT: AtomicU32 = AtomicU32::new(0);
pub(crate) static POLLER_REFCOUNT: AtomicU32 = AtomicU32::new(0);
use std::cell::RefCell;
use core::ops::DerefMut;
use core::mem::size_of;
use std::convert::TryInto;
use rkyv::{
archived_value,
de::deserializers::AllocDeserializer,
ser::{Serializer, serializers::WriteSerializer},
AlignedVec,
Deserialize,
};
#[derive(num_derive::FromPrimitive, num_derive::ToPrimitive, Debug)]
pub enum CbOp {
Change,
@ -390,7 +401,6 @@ impl Pddb {
result: PddbRequestCode::Uninit,
cb_sid,
alloc_hint: if let Some(a) = alloc_hint {Some(a as u64)} else {None},
bulk_limit: None,
};
let mut buf = Buffer::into_buf(request)
.or(Err(Error::new(ErrorKind::Other, "Xous internal error")))?;
@ -469,7 +479,6 @@ impl Pddb {
result: PddbRequestCode::Uninit,
cb_sid,
alloc_hint: None,
bulk_limit: None,
};
let mut buf = Buffer::into_buf(request)
.or(Err(Error::new(ErrorKind::Other, "Xous internal error")))?;
@ -516,7 +525,6 @@ impl Pddb {
result: PddbRequestCode::Uninit,
cb_sid,
alloc_hint: None,
bulk_limit: None,
};
let mut buf = Buffer::into_buf(request)
.or(Err(Error::new(ErrorKind::Other, "Xous internal error")))?;
@ -570,6 +578,7 @@ impl Pddb {
index: 0,
code: PddbRequestCode::Uninit,
token,
bulk_limit: None,
};
let mut buf = Buffer::into_buf(request)
.or(Err(Error::new(ErrorKind::Other, "Xous internal error")))?;
@ -651,6 +660,7 @@ impl Pddb {
index: 0,
code: PddbRequestCode::Uninit,
token,
bulk_limit: None,
};
let mut buf = Buffer::into_buf(request)
.or(Err(Error::new(ErrorKind::Other, "Xous internal error")))?;
@ -675,6 +685,7 @@ impl Pddb {
index,
code: PddbRequestCode::Uninit,
token,
bulk_limit: None,
};
let mut buf = Buffer::into_buf(request)
.or(Err(Error::new(ErrorKind::Other, "Xous internal error")))?;
@ -753,17 +764,12 @@ impl Pddb {
/// Retrieve an entire dictionary of data in a single call. Will return data records up to but not
/// over a total of `size_limit`. Keys that exceed the limit are still enumerated, but their
/// data sections are `None`, instead of `Some(Vec::<u8>)`.
/// Defaults to a size limit of up to 32k of bulk data returned, if it is not explicitly specified.
pub fn read_dict(&self, dict: &str, basis: Option<&str>, size_limit: Option<usize>) -> Result<Vec::<PddbKeyRecord>> {
use rkyv::{
archived_value,
de::deserializers::AllocDeserializer,
ser::{Serializer, serializers::WriteSerializer},
AlignedVec,
Archive,
Archived,
Deserialize,
Serialize,
};
// about the biggest we can move around in Precursor and not break heap.
const MAX_BUFLEN: usize = 128 * 1024;
// compromise between memory zeroing time and latency to send a message
const DEFAULT_LIMIT: usize = 32 * 1024;
/*
The operation proceeds in two phases. All operations use the DictBulkRead opcode.
@ -782,12 +788,15 @@ impl Pddb {
the allocated buffer, the data will always be returned as `None`; in case the data is larger than
the available space, the serialization stops, and the record is returned.
Subsequent calls from the client to the main server requries a `PddbDictRequest` structure
to be in the buffer, but only the "token" field is considered. The rest of the fields are disregarded;
changing the requested dictionary or other parameters has no impact on the call trajectory.
The PddbKeyRecords are manually packed into the memory messages with the following format:
MemoryMessage-custom fields:
- `offset` field of memory message: the current index-offset that the return data starts at. Used as a sanity check.
- `valid` field of the memory message: number of records expected inside the return data structure.
Data slice:
- `code`: u32 containing the current transaction code
- `starting_key_index` u32 of the current index-offset that the return data starts at. Used as a sanity check.
- `len` u32 of number of records expected inside the return data structure.
- `total`: u32 with total number of keys to be transmitted (shouldn't change over the life of the session)
- `token`: [u32; 4] confirming or establishing the session token (shouldn't change over the life of the session)
First record start at data[24]:
@ -803,61 +812,132 @@ impl Pddb {
The caller shall repeatedly call `DictBulkRead` until the `code` is `Last`, at which point the token
is deleted from the server, freeing it to service a new transaction. The caller should then collate the
results into the corresponding return vector.
If the caller fails to drain all the data, the server eventually times out and forgets the token.
The client will then get a "Busy" return code or "NotFound", depending on if they had fully initialized the
request structure for the subsequent calls.
*/
// allocate buffer, which is shuffled back and forth to the PDDB server
let alloc_target = if let Some(size_limit) = size_limit {
if size_limit < 4096 {
4096
} else {
// rounds *down* to the nearest page size, since it is a "guaranteed not to exceed" limit
size_limit & (4096 - 1)
if size_limit > MAX_BUFLEN {
MAX_BUFLEN
} else {
// rounds *down* to the nearest page size, since it is a "guaranteed not to exceed" limit
size_limit & (4096 - 1)
}
}
} else {
// compromise between memory zeroing time and latency to send a message
32 * 1024
DEFAULT_LIMIT
};
let msg_mem = xous::map_memory(
let mut msg_mem = xous::map_memory(
None,
None,
alloc_target,
MemoryFlags::R | MemoryFlags::W,
xous::MemoryFlags::R | xous::MemoryFlags::W,
)
.expect("Likely OOM in allocating read_dict buffer");
// setup the request arguments
let request = PddbDictRequest {
let mut request = PddbDictRequest {
basis_specified: basis.is_some(),
basis: xous_ipc::String::<BASIS_NAME_LEN>::from_str(basis.unwrap_or("")),
dict: xous_ipc::String::<DICT_NAME_LEN>::from_str(dict),
key: xous_ipc::String::<KEY_NAME_LEN>::new(),
index: 0,
token: [0u32; 4],
code: PddbRequestCode::BulkRead
code: PddbRequestCode::BulkRead,
bulk_limit: Some(size_limit.unwrap_or(DEFAULT_LIMIT)),
};
let mut serializer = WriteSerializer::new(AlignedVec::new());
let pos = serializer.serialize_value(&request).unwrap();
let buf = serializer.into_inner();
msg_mem.as_slice_mut()[..buf.len()].copy_from_slice(buf.as_slice());
// initiate the request
let msg = MemoryMessage {
id: Opcode::DictBulkRead.to_usize().unwrap(),
buf: msg_mem,
offset: pos,
valid: buf.len(),
};
// main loop
let mut ret = Vec::new();
let mut check_total_keys;
let mut check_key_index = 0;
loop {
let (pos, valid) = match xous::send_message(
let mut serializer = WriteSerializer::new(AlignedVec::new());
let pos = serializer.serialize_value(&request).unwrap();
let buf = serializer.into_inner();
msg_mem.as_slice_mut()[..buf.len()].copy_from_slice(buf.as_slice());
// initiate the request: assemble a MemoryMessage
let msg = xous::MemoryMessage {
id: Opcode::DictBulkRead.to_usize().unwrap(),
buf: msg_mem,
offset: xous::MemoryAddress::new(pos),
valid: xous::MemorySize::new(buf.len()),
};
// do a mutable lend to the server
let (_pos, _valid) = match xous::send_message(
self.conn,
msg
Message::MutableBorrow(msg)
) {
Ok(xous::Result::MemoryReturned(offset, valid)) => (offset, valid),
_ => return Err(Error::new(ErrorKind::Other, "Xous internal error"))
};
// unpack the return code. The result is not a single rkyv struct, it's hand-packed binary data. Unpack it.
let mut index = 0;
let mut header = BulkReadHeader::default();
header.deref_mut().copy_from_slice(
&msg_mem.as_slice()[index..index + size_of::<BulkReadHeader>()]
);
match FromPrimitive::from_u32(header.code).unwrap_or(PddbBulkReadCode::InternalError) {
PddbBulkReadCode::NotFound => {
return Err(Error::new(ErrorKind::NotFound, format!("Bulk Read: dictionary '{}' not found", dict)))
}
PddbBulkReadCode::Busy => {
return Err(Error::new(ErrorKind::TimedOut, "PDDB server busy or request timed out"))
}
PddbBulkReadCode::Streaming | PddbBulkReadCode::Last => {
// --- unpack the received data
// stash the token for future iterations
request.token = header.token;
check_total_keys = header.total;
if check_key_index != header.starting_key_index {
log::error!("local key index did not match remote key index: {}, {}", check_key_index, header.starting_key_index);
}
let mut key_count = 0;
while key_count < header.len {
if index + size_of::<u32>() * 2 > msg_mem.len() {
// quit if we don't have enough space to decode at least another two indices
break;
}
let size = u32::from_le_bytes(msg_mem.as_slice()[index..index + size_of::<u32>()].try_into().unwrap());
index += size_of::<u32>();
let pos = u32::from_le_bytes(msg_mem.as_slice()[index..index + size_of::<u32>()].try_into().unwrap());
index += size_of::<u32>();
if size != 0 && pos != 0 {
let archived = unsafe {
archived_value::<PddbKeyRecord>(&msg_mem.as_slice()[index..index + (size as usize)], pos as usize)
};
index += size as usize;
let key = archived.deserialize(&mut AllocDeserializer).unwrap();
ret.push(key);
key_count += 1;
check_key_index += 1;
} else {
// we encountered a nil field, stop decoding
break;
}
}
if key_count != header.len {
log::error!("key count did not match number in header {}, {}", key_count, header.len);
}
// if this is the last block, quit the loop
if header.code == (PddbBulkReadCode::Last as u32) {
break;
}
}
_ => {
return Err(Error::new(ErrorKind::Other, "Bulk read invalid return value"))
}
}
}
if check_total_keys != check_key_index {
log::error!("Number of keys read does not match expected value: {}, {}", check_total_keys, check_key_index);
}
Ok(Vec::new())
Ok(ret)
}
/// Triggers a dump of the PDDB to host disk

236
services/pddb/src/main.rs
View file

@ -398,6 +398,15 @@ use std::sync::Arc;
use locales::t;
use rkyv::{
de::deserializers::AllocDeserializer,
ser::{Serializer, serializers::WriteSerializer},
AlignedVec,
Deserialize,
};
use core::mem::size_of;
use core::ops::Deref;
#[cfg(feature="perfcounter")]
const FILE_ID_SERVICES_PDDB_SRC_MAIN: u32 = 0;
#[cfg(feature="perfcounter")]
@ -574,6 +583,7 @@ fn wrapped_main() -> ! {
let mut key_token: Option<[u32; 4]> = None;
let mut dict_list = Vec::<String>::new(); // storage for dict lists
let mut dict_token: Option<[u32; 4]> = None;
let mut bulkread_state: Option<BulkReadState> = None;
// the PDDB resets the hardware RTC to a new random starting point every time it is reformatted
// it is the only server capable of doing this.
@ -1394,8 +1404,205 @@ fn wrapped_main() -> ! {
pddb_os.perf_entry(FILE_ID_SERVICES_PDDB_SRC_MAIN, perflib::PERFMETA_ENDBLOCK, 4, std::line!());
}
// Optimized bulk data read handler. See lib.rs for documentation.
Opcode::DictBulkRead => {
const BULKREAD_TIMEOUT_MS: u64 = 5000;
let range = msg.body.memory_message_mut().unwrap();
let buf = unsafe { core::slice::from_raw_parts_mut(
range.buf.as_mut_ptr(),
range.buf.len(),
)};
// unpack the descriptor
let pos = range.offset.map(|o| o.get()).unwrap_or_default();
let r = unsafe { rkyv::archived_value::<PddbDictRequest>(buf, pos) };
let bulk_descriptor = r.deserialize(&mut AllocDeserializer).unwrap();
// check for a timeout; retire state if we did timeout
let mut timed_out = false;
if let Some(state) = &bulkread_state {
if tt.elapsed_ms() - state.last_time > BULKREAD_TIMEOUT_MS {
timed_out = true;
}
}
if timed_out {
bulkread_state = None;
}
// handle state initialization, if no call was previously initiated
let mut first_call = false;
if bulkread_state.is_none() {
first_call = true;
// confirm data exists; setup the tracking state
let key_list: Vec<String> = match basis_cache.key_list(
&mut pddb_os,
bulk_descriptor.dict.as_str().unwrap(),
if bulk_descriptor.basis_specified {
Some(bulk_descriptor.basis.as_str().unwrap())
} else {
None
}
) {
Ok(list) => list.into_iter().rev().collect(), // reverse order so Vec can just pop and get "first" item
Err(e) => {
match e.kind() {
std::io::ErrorKind::NotFound => {
buf[..4].copy_from_slice(&(PddbBulkReadCode::NotFound as u32).to_le_bytes())
}
_ => {
buf[..4].copy_from_slice(&(PddbBulkReadCode::InternalError as u32).to_le_bytes())
}
}
// this will cause the return record to just have the error code copied to it. The rest is invalid.
continue;
}
};
let state = BulkReadState {
token: [pddb_os.trng_u32(), pddb_os.trng_u32(), pddb_os.trng_u32(), pddb_os.trng_u32()],
is_basis_specified: bulk_descriptor.basis_specified,
basis: if bulk_descriptor.basis_specified{ bulk_descriptor.basis.to_string() } else { String::new() },
dict: bulk_descriptor.dict.to_string(),
total_keys: key_list.len(),
key_list,
buf_starting_key_index: 0,
last_time: tt.elapsed_ms(),
serialized: None,
read_limit: bulk_descriptor.bulk_limit.expect("bulk limit must be specified for bulk read calls"),
read_total: 0,
};
bulkread_state = Some(state);
}
// handle data packing into the structure
if let Some(state) = &mut bulkread_state {
// check that the token matches, if this isn't a first call to the function
if !first_call {
if state.token != bulk_descriptor.token {
buf[..4].copy_from_slice(&(PddbBulkReadCode::Busy as u32).to_le_bytes());
continue;
}
}
// start filling in the return structure
let mut header = BulkReadHeader::default();
header.total = state.total_keys as u32;
header.starting_key_index = state.buf_starting_key_index as u32;
header.token = state.token;
// now loop through and pack data into the slice
let mut index = size_of::<BulkReadHeader>();
let mut finished = false;
loop {
let (pos, sbuf) = if let Some((pos, sbuf)) = state.serialized.take() {
(pos, sbuf)
} else {
// no data was serialized, create it.
if let Some(key_name) = state.key_list.pop() {
let attr = basis_cache.key_attributes(&mut pddb_os,
&state.dict,
&key_name,
if state.is_basis_specified{Some(&state.basis)} else {None}
).expect("Key went missing during bulk read"); // could be a concurrent process mutating. We don't handle this; flag with a panic.
if attr.len < state.read_limit - state.read_total && // will it fit in the remaining read limit?
// will it fit even in a single buffer?
// computed as: 2x u32 for pos/len overhead
// size of the base key record, archived
// maximum length strings for basis name and key name
// plus header size
// 32 bytes "slop" to make sure we never fail to properly compute this limit
attr.len <
buf.len()
- (size_of::<u32>() * 2 + size_of::<ArchivedPddbKeyRecord>() + BASIS_NAME_LEN + KEY_NAME_LEN + size_of::<BulkReadHeader>() + 32)
{
let mut d = vec![0u8; attr.len];
match basis_cache.key_read(
&mut pddb_os,
&state.dict,
&key_name,
&mut d,
None,
Some(&attr.basis),
) {
Ok(readlen) => {
assert!(readlen == attr.len, "Bulk read key length did not match expected length");
let rec = PddbKeyRecord {
name: key_name,
len: attr.len,
reserved: attr.reserved,
age: attr.age,
index: attr.index,
basis: attr.basis,
data: Some(d)
};
state.read_total += attr.len; // commit the read length early
let mut serializer = WriteSerializer::new(AlignedVec::new());
let pos = serializer.serialize_value(&rec).unwrap();
(pos, serializer.into_inner())
}
Err(e) => {
panic!("Error reading previously attributed key {}: {:?}", &key_name, e);
}
}
} else {
// report the key, but with no data
let rec = PddbKeyRecord {
name: key_name,
len: attr.len,
reserved: attr.reserved,
age: attr.age,
index: attr.index,
basis: attr.basis,
data: None,
};
let mut serializer = WriteSerializer::new(AlignedVec::new());
let pos = serializer.serialize_value(&rec).unwrap();
(pos, serializer.into_inner())
}
} else {
// no more keys; we're done.
finished = true;
break;
}
};
// sbuf now contains some serialized data. will it fit? Need 2x`u32` as recordkeeeping overhead.
assert!(sbuf.len() == sbuf.as_slice().len(), "remove this assert if it doesn't fail on first run");
if sbuf.len() + size_of::<u32>() * 2 <= buf.len() - index {
// data can fit, copy it into the buffer.
state.buf_starting_key_index += 1;
header.len += 1;
// read length increment was handled when the data was copied into the serialization buffer.
buf[index..index + 4].copy_from_slice(
&(sbuf.len() as u32).to_le_bytes()
);
index += size_of::<u32>();
buf[index..index + 4].copy_from_slice(
&(pos as u32).to_le_bytes()
);
index += size_of::<u32>();
buf[index..index + sbuf.len()].copy_from_slice(
sbuf.as_slice()
);
index += sbuf.len();
} else {
assert!(
sbuf.len() + size_of::<u32>() * 2 > buf.len(),
// the remediation for this is to fix the "will it fit even in a single buffer?" test above
"Edge case not handled correctly: we have a serialized record that can never fit in a buffer"
);
// save the data for the next round, when we have an empty buffer
state.serialized = Some((pos, sbuf));
// data didn't fit, quit with finished = false;
break;
}
}
if finished {
header.code = PddbBulkReadCode::Last as u32;
} else {
header.code = PddbBulkReadCode::Streaming as u32;
}
buf[..size_of::<BulkReadHeader>()].copy_from_slice(
header.deref()
);
// update the last access time
state.last_time = tt.elapsed_ms();
} else {
log::warn!("This should be unreachable, the state should always be initialized by this point");
}
}
Opcode::SeekKeyStd => {
@ -1934,4 +2141,31 @@ pub(crate) fn heap_usage() -> usize {
0
},
}
}
}
struct BulkReadState {
/// API token
token: [u32; 4],
/// determines if the basis was specified
is_basis_specified: bool,
/// the current basis, if specified
basis: String,
/// the dictionary to read
dict: String,
/// list of keys in the dictionary. Entries are removed as they are serialized.
key_list: Vec::<String>,
/// total number of keys to send (e.g. initial key_list.len())
total_keys: usize,
/// Each buffer can hold multiple keys; a single buffer may be re-used several
/// times to send a large dictionary with many keys. This tracks the starting index
/// of the current buffer's set of keys.
buf_starting_key_index: usize,
/// Total limit of bulk read data to return
read_limit: usize,
/// Amount of data read so far
read_total: usize,
/// last interaction time, so we can timeout the state in the case that the client is misbehaved
last_time: u64,
/// the proposed data to send. Can be aborted if it is too big to fit in the remaining space.
serialized: Option<(usize, AlignedVec)>,
}