edgul/blob-storage
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

changed Blob's underlying data structure to be VecDeque

Browse Source
This commit is contained in:
Ed Guloien
2026-06-02 17:59:22 -04:00
parent 6047211b33
commit 44e6bfc0c3
1 changed files with 106 additions and 265 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main.rs
+105 -264
View File
@@ -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<Range>, // 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<Vec<u8>, 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<Range> = 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<RangeNode>,
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<Box<[u8; ARRAY_SIZE]>>, // complexity variable n
dropped: usize,
ranges: Vec<Range>, // 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<Vec<u8>, 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);
// 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;
}
node_global_start = node_global_end;
// 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<u8> O(n)
// could also use a Vec<Option<Box>>, 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 || {