diff --git a/src/main.rs b/src/main.rs index 508eb5a..f0cfff2 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,256 +1,149 @@ #![allow(unused)] -use std::collections::LinkedList; +use std::collections::VecDeque; use std::sync::{Arc, Mutex, RwLock}; const ARRAY_SIZE: usize = 4096; #[derive(PartialEq, Debug, Clone, Copy)] struct Range { - // todo: also consider there might be an off the shelf solution that does ranges start: usize, end: usize, // exclusive } -#[derive(PartialEq, Debug)] -enum RangeLocation { - LeftEnd, - RightEnd, - Middle, - No, -} - impl Range { fn new(start: usize, end: usize) -> Self { Self { start, end } } - fn where_in_range(&self, index: usize) -> RangeLocation { - if index == self.start { - return RangeLocation::LeftEnd; - } - if index >= self.end { - return RangeLocation::No; - } - if index == self.end - 1 { - return RangeLocation::RightEnd; - } - RangeLocation::Middle - } - fn contains(&self, range: &Range) -> bool { range.start >= self.start && range.end <= self.end } - - fn is_empty(&self) -> bool { - self.start >= self.end - } -} - -#[derive(Debug)] -struct RangeNode { - ranges: Vec, // valid ranges - bytes: [u8; ARRAY_SIZE], - start: usize, // global - end: usize, // global - capacity: usize, -} - -impl RangeNode { - fn default() -> Self { - RangeNode::new(0, 0) - } - - fn new(start: usize, end: usize) -> Self { - Self { - ranges: Vec::new(), - bytes: [0; ARRAY_SIZE], - start, - end, - capacity: start + ARRAY_SIZE, - } - } - - fn read(&mut self, start: usize, len: usize) -> Result, BlobError> { - let mut v = Vec::new(); - let read_range = Range::new(start, start + len); - match self.ranges.last() { - Some(last) if read_range.end <= last.end => {} - _ => return Err(BlobError::InvalidRange), - } - - // check bytes are not already read - let mut contains = false; - for valid_range in self.ranges.iter() { - if valid_range.contains(&read_range) { - contains = true; - break; - } - if read_range.start >= valid_range.end { - return Err(BlobError::BytesAlreadyRead); - } - } - if !contains { - // Invalid range already caught above - return Err(BlobError::BytesAlreadyRead); - } - - let slice = &mut self.bytes[start..start + len]; - v.extend_from_slice(slice); - - // we copy existing entries into a new list to avoid iterator invalidation - // from inserting while iterating - // todo: might be a better way - let clear_range = Range { - start, - end: start + len, - }; - let mut new_ranges = Vec::new(); - let mut i = 0; - while i < self.ranges.len() { - let mut extra_range: Option = None; - let mut range = self.ranges[i]; - if range.where_in_range(clear_range.start) == RangeLocation::LeftEnd { - range.start = clear_range.end; - } else if range.where_in_range(clear_range.end - 1) == RangeLocation::RightEnd { - range.end = clear_range.end - } else if range.where_in_range(clear_range.start) == RangeLocation::Middle { - let end = range.end; - range.end = clear_range.start; - extra_range = Some(Range::new(clear_range.end, end)); - } - - if !range.is_empty() { - new_ranges.push(range); - } - - if let Some(er) = extra_range - && !er.is_empty() - { - new_ranges.push(er); - } - i += 1; - } - self.ranges = new_ranges; - Ok(v) - } - - fn append(&mut self, input: &[u8]) -> usize { - let new_len = input.len().min(self.capacity - self.end); - if new_len == 0 { - return 0; - } - let new_range = Range { - start: self.end, - end: self.end + new_len, - }; - - // we are appending so only have to check the last range - match self.ranges.last_mut() { - Some(range) => { - if range.where_in_range(new_range.start) != RangeLocation::No { - range.end = new_range.end - } else { - self.ranges.push(new_range); - } - } - _ => { - self.ranges.push(new_range); - } - } - self.bytes[new_range.start..new_range.end].copy_from_slice(&input[0..new_len]); - self.end += new_len; - new_len - } - - fn available_bytes(&self) -> usize { - self.bytes.len() - self.end - } - - fn contains(&self, index: usize) -> bool { - index >= self.start && index < self.end - } - - fn overlaps(&self, range: &Range) -> bool { - self.contains(range.start) || self.contains(range.end) - } - - fn all_read(&self) -> bool { - false - } } #[derive(Debug)] pub struct Blob { - // todo: consider fallible_vec to detect OOM on append - nodes: LinkedList, - end_index: usize, + // todo: consider fallible collections to detect OOM on append + + // space complexity O(n + m) + // where n is number of bytes + // and m is number of ranges + // ------------------------- + // also tried using LinkedList, but slow reading when ranges were inside nodes + // and indexing into LinkedList nodes doesn't seem to jive well with Rust's ownership model + // especially when considering multi-threaded access + arrays: VecDeque>, // complexity variable n + dropped: usize, + ranges: Vec, // valid, global // complexity variable m + end_index: usize, // global } impl Blob { fn new() -> Self { Self { - nodes: LinkedList::new(), + arrays: VecDeque::new(), + dropped: 0, + ranges: Vec::new(), end_index: 0, } } - fn append(&mut self, input: &[u8]) -> usize { - let mut remaining = input; - let mut written_total = 0; - while !remaining.is_empty() { - // add nodes if necessary - let mut new_node_start = 0; - let need_new = match self.nodes.back() { - Some(n) => { - new_node_start = n.end; - n.available_bytes() == 0 - } - None => true, - }; - if need_new { - self.nodes.push_back(RangeNode::default()); - } - // write to each node's array - let node = self.nodes.back_mut().unwrap(); - let written = node.append(remaining); - remaining = &remaining[written..]; - written_total += written; - } - self.end_index += input.len(); - written_total + // find array index - O(1) + fn locate(&self, global_off: usize) -> (usize, usize) { + ( + global_off / ARRAY_SIZE - self.dropped, + global_off % ARRAY_SIZE, + ) } + // appending complexity time, worst: O(n + a + m) + // where n is bytes written + // and a is arrays added + // and m is ranges added, usually 0 or 1, but reallocation could copy (m) + fn append(&mut self, input: &[u8]) -> usize { + if input.is_empty() { + return 0; + } + + let mut written = 0; + while written < input.len() { + // add new array if needed + let global_offset = self.end_index + written; + let within = global_offset % ARRAY_SIZE; + let physical = global_offset / ARRAY_SIZE - self.dropped; + if physical == self.arrays.len() { + // O(1) to push_back + self.arrays.push_back(Box::new([0u8; ARRAY_SIZE])); + } + // write to array - worst: O(n) + let take = (ARRAY_SIZE - within).min(input.len() - written); + let array = &mut self.arrays[physical]; + array[within..within + take].copy_from_slice(&input[written..written + take]); + written += take; + } + // update the ranges vector - O(1) + let new_range = Range::new(self.end_index, self.end_index + input.len()); + match self.ranges.last_mut() { + Some(last) if last.end == new_range.start => last.end = new_range.end, + _ => self.ranges.push(new_range), + } + + self.end_index += input.len(); + written + } + + // read time complexity, worst: O(n + m + d) + // where n is bytes read + // and m is ranges added. Usually O(1), but could be O(m) on re-allocation copy + // and d is arrays dropped fn read(&mut self, start: usize, len: usize) -> Result, BlobError> { - // early check for out-of-range + if len == 0 { + return Ok(Vec::new()); + } if start + len > self.end_index { + // early range check return Err(BlobError::InvalidRange); } - // read the bytes from each node - let mut remaining_start = start; - let mut remaining_len = len; - let mut read_vec = Vec::with_capacity(len); - let mut node_global_start = 0; - // works, but this is SLOW - for n in self.nodes.iter_mut() { - if read_vec.len() == len { - break; - } - // if !n.overlaps(&Range::new(remaining_start, remaining_start+remaining_len)) { - // continue; - // } + let read_range = Range::new(start, start + len); + let Some(idx) = self.ranges.iter().position(|r| r.contains(&read_range)) else { + return Err(BlobError::BytesAlreadyRead); + }; - // todo: this works, but is a bit too clunky - let node_global_end = node_global_start + n.end; - let want = start + read_vec.len(); // next global byte we still need - if want < node_global_end { - let local_start = want - node_global_start; - let take = (len - read_vec.len()).min(n.end - local_start); - let chunk = n.read(local_start, take)?; - read_vec.extend_from_slice(&chunk); - } - node_global_start = node_global_end; + // read the bytes - O(n) + let mut read_vec = Vec::with_capacity(len); + let mut copied = 0; + while copied < len { + let (physical, within) = self.locate(start + copied); + let take = (ARRAY_SIZE - within).min(len - copied); + read_vec.extend_from_slice(&self.arrays[physical][within..within + take]); + copied += take; + } + + // update valid ranges - worst case: O(m), usual case O(1) + let vr = self.ranges[idx]; + let mut remainder = Vec::new(); + if vr.start < read_range.start { + remainder.push(Range::new(vr.start, read_range.start)); + } + if read_range.end < vr.end { + remainder.push(Range::new(read_range.end, vr.end)); + } + self.ranges.splice(idx..idx + 1, remainder); + + // reclaim arrays below the lowest unread byte + // this will keep memory footprint relatively small + // at low cost O(1) or worst on re-allocation: O(number_of_arrays) + // compared to a Vec O(n) + // could also use a Vec>, but this would leave None slots + let min_unread = self + .ranges + .first() + .map(|r| r.start) + .unwrap_or(self.end_index); + while self.dropped < min_unread / ARRAY_SIZE { + self.arrays.pop_front(); + self.dropped += 1; } Ok(read_vec) } @@ -339,11 +232,6 @@ mod tests { let range2 = Range { start: 0, end: 3 }; assert_eq!(range, range2); // similar - assert_eq!(range.where_in_range(0), RangeLocation::LeftEnd); - assert_eq!(range.where_in_range(1), RangeLocation::Middle); - assert_eq!(range.where_in_range(2), RangeLocation::RightEnd); - assert_eq!(range.where_in_range(3), RangeLocation::No); - assert!(range.contains(&range2)); assert!(range2.contains(&range)); @@ -369,47 +257,6 @@ mod tests { let read = b.read(0, 1); // left read assert!(read.is_ok()); assert_eq!(read.unwrap(), vec![b'a']); - assert_eq!( - b.nodes.back().unwrap().ranges, - &[Range { start: 1, end: 3 }] - ); - - let read = b.read(0, 1); // dupe read - assert!(read.is_err()); - assert_eq!( - b.nodes.back().unwrap().ranges, - &[Range { start: 1, end: 3 }] - ); - - let read = b.read(1, 2); // right read - assert!(read.is_ok()); - assert_eq!(b.nodes.back().unwrap().ranges, &[]); - - let written = b.append(&[b'a', b'b', b'c']); - assert_eq!(written, 3); - assert_eq!( - b.nodes.back().unwrap().ranges, - &[Range { start: 3, end: 6 }] - ); - - let read = b.read(4, 1); // middle read - assert!(read.is_ok()); - assert_eq!( - b.nodes.back().unwrap().ranges, - &[Range::new(3, 4), Range::new(5, 6)] - ); - } - - #[test] - fn test_range_node() { - let mut b = RangeNode::default(); - let written = b.append(&[b'a', b'b', b'c']); - assert_eq!(written, 3); - assert_eq!(b.ranges, &[Range { start: 0, end: 3 }]); - - let read = b.read(0, 1); // left read - assert!(read.is_ok()); - assert_eq!(read.unwrap(), &[b'a']); assert_eq!(b.ranges, &[Range { start: 1, end: 3 }]); let read = b.read(0, 1); // dupe read @@ -418,7 +265,6 @@ mod tests { let read = b.read(1, 2); // right read assert!(read.is_ok()); - assert_eq!(read.unwrap(), &[b'b', b'c']); assert_eq!(b.ranges, &[]); let written = b.append(&[b'a', b'b', b'c']); @@ -427,7 +273,6 @@ mod tests { let read = b.read(4, 1); // middle read assert!(read.is_ok()); - assert_eq!(read.unwrap(), &[b'b']); assert_eq!(b.ranges, &[Range::new(3, 4), Range::new(5, 6)]); } @@ -436,12 +281,11 @@ mod tests { use std::thread; let bm = Arc::new(BlobManager::new()); - let handles: Vec<_> = (0..2usize) .map(|i| { let bm = Arc::clone(&bm); thread::spawn(move || { - let len = 1_000_000; // 100 MB test takes a while + let len = 10_000_000; // 100 MB test takes a while let data = vec![i as u8; len]; bm.create_blob(i).unwrap(); bm.append(i, &data).unwrap(); @@ -463,9 +307,7 @@ mod tests { use std::thread; let bm = Arc::new(BlobManager::new()); - let num_creates = 20_000; - let even = { let bm = Arc::clone(&bm); thread::spawn(move || { @@ -474,7 +316,6 @@ mod tests { } }) }; - let odd = { let bm = Arc::clone(&bm); thread::spawn(move || {