diff --git a/services/pddb/src/api.rs b/services/pddb/src/api.rs index f4947e386..8ac62125d 100644 --- a/services/pddb/src/api.rs +++ b/services/pddb/src/api.rs @@ -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, } /// A structure for requesting a token to access a particular key/value pair @@ -311,8 +314,6 @@ pub struct PddbKeyRequest { pub alloc_hint: Option, // 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, } 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::()) + 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::()) + as &mut [u8] + } + } } #[allow(dead_code)] diff --git a/services/pddb/src/lib.rs b/services/pddb/src/lib.rs index d7e5dbf48..b7ae11d0c 100644 --- a/services/pddb/src/lib.rs +++ b/services/pddb/src/lib.rs @@ -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::)`. + /// 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) -> Result> { - 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::::from_str(basis.unwrap_or("")), dict: xous_ipc::String::::from_str(dict), key: xous_ipc::String::::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::()] + ); + 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::() * 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::()].try_into().unwrap()); + index += size_of::(); + let pos = u32::from_le_bytes(msg_mem.as_slice()[index..index + size_of::()].try_into().unwrap()); + index += size_of::(); + if size != 0 && pos != 0 { + let archived = unsafe { + archived_value::(&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 diff --git a/services/pddb/src/main.rs b/services/pddb/src/main.rs index b7634bcdc..42fc6e6c6 100644 --- a/services/pddb/src/main.rs +++ b/services/pddb/src/main.rs @@ -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::::new(); // storage for dict lists let mut dict_token: Option<[u32; 4]> = None; + let mut bulkread_state: Option = 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::(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 = 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::(); + 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::() * 2 + size_of::() + BASIS_NAME_LEN + KEY_NAME_LEN + size_of::() + 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::() * 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::(); + buf[index..index + 4].copy_from_slice( + &(pos as u32).to_le_bytes() + ); + index += size_of::(); + buf[index..index + sbuf.len()].copy_from_slice( + sbuf.as_slice() + ); + index += sbuf.len(); + } else { + assert!( + sbuf.len() + size_of::() * 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::()].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 }, } -} \ No newline at end of file +} + +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::, + /// 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)>, +}