diff --git a/src/main.rs b/src/main.rs index 40377f8..0ce90ff 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,5 +1,7 @@ #![allow(unused)] +use std::sync::{Arc, Mutex, RwLock}; + #[derive(PartialEq, Debug, Clone, Copy)] struct Range { // todo: consider just using a map instead @@ -155,43 +157,58 @@ struct BlobManager { // fine if we assume: // * IDs are sequentially generated // * and IDs can be re-used - blobs: Vec>, + blobs: RwLock>>>>, + // Also note RwLock (create) starvation might be a concern if caller is + // append/read heavy due to platform-defined fairness + // todo: better approach? } +// todo: blob indexing is left up to the caller maybe add ID generator +// for cross-thread synchronization or change approach impl BlobManager { pub fn new() -> Self { - Self { blobs: Vec::new() } + Self { + blobs: RwLock::new(Vec::new()), + } } // todo: is usize appropriate datatype for ID? - pub fn create_blob(&mut self, id: usize) -> Result<(), BlobError> { - if let Some(Some(_)) = self.blobs.get_mut(id) { + pub fn create_blob(&self, id: usize) -> Result<(), BlobError> { + let mut blobs = self.blobs.write().unwrap(); + if let Some(Some(_)) = blobs.get(id) { return Err(BlobError::BlobExists); - } else if id >= self.blobs.len() { + } else if id >= blobs.len() { // todo: better strategy possible? - self.blobs.resize_with(id + 1, || None); + blobs.resize_with(id + 1, || None); } - self.blobs[id] = Some(Blob::new()); + blobs[id] = Some(Arc::new(Mutex::new(Blob::new()))); Ok(()) } - pub fn append(&mut self, id: usize, input: &[u8]) -> Result<(), BlobError> { - if let Some(Some(b)) = self.blobs.get_mut(id) { - b.append(input); - return Ok(()); - } - Err(BlobError::BlobDNE) + pub fn append(&self, id: usize, input: &[u8]) -> Result<(), BlobError> { + let blob = { + let blobs = self.blobs.read().unwrap(); + match blobs.get(id) { + Some(Some(b)) => Arc::clone(b), + _ => return Err(BlobError::BlobDNE), + } + }; + blob.lock().unwrap().append(input); + Ok(()) } - pub fn read(&mut self, id: usize, start: usize, len: usize) -> Result, BlobError> { - if let Some(Some(b)) = self.blobs.get_mut(id) { - let mut read = vec![0; len]; - b.read(start, len, &mut read)?; // propagate error to caller - // todo: if read clears a blob entirely should the blob be removed? - Ok(read) - } else { - Err(BlobError::BlobDNE) - } + pub fn read(&self, id: usize, start: usize, len: usize) -> Result, BlobError> { + let blob = { + let blobs = self.blobs.read().unwrap(); + match blobs.get(id) { + Some(Some(b)) => Arc::clone(b), + _ => return Err(BlobError::BlobDNE), + } + }; + let mut read = vec![0; len]; + blob.lock().unwrap().read(start, len, &mut read)?; // propagate error to caller + // todo: if read clears a blob entirely should the blob be removed? + Ok(read) } } @@ -257,9 +274,68 @@ mod tests { assert_eq!(b.valid_ranges, &[Range::new(3, 4), Range::new(5, 6)]); } + #[test] + fn test_blob_manager_threaded_read() { + 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 = 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(); + for j in 0..len as usize { + let byte = bm.read(i, j, 1).unwrap(); + assert_eq!(byte[0], i as u8); + } + }) + }) + .collect(); + + for handle in handles { + handle.join().unwrap(); + } + } + + #[test] + fn test_blob_manager_threaded_create() { + use std::thread; + + let bm = Arc::new(BlobManager::new()); + + let even = { + let bm = Arc::clone(&bm); + thread::spawn(move || { + for i in (0..20000usize).step_by(2) { + bm.create_blob(i).unwrap(); + } + }) + }; + + let odd = { + let bm = Arc::clone(&bm); + thread::spawn(move || { + for i in (1..20000usize).step_by(2) { + bm.create_blob(i).unwrap(); + } + }) + }; + + even.join().unwrap(); + odd.join().unwrap(); + + let blobs = bm.blobs.read().unwrap(); + assert_eq!(blobs.len(), 20000); + assert!(blobs.iter().all(|b| b.is_some())); + } + #[test] fn test_blob_manager() { - let mut bm = BlobManager::new(); + let bm = BlobManager::new(); bm.create_blob(0); bm.append(0, &[b'a', b'b', b'c']); let v = Vec::from([b'a', b'b']);