made BlobManager and Blobs thread-safe

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<Option<Blob>>,
+    blobs: RwLock<Vec<Option<Arc<Mutex<Blob>>>>>,
+    // 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(());
+    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),
             }
-        Err(BlobError::BlobDNE)
+        };
+        blob.lock().unwrap().append(input);
+        Ok(())
     }
 
-    pub fn read(&mut self, id: usize, start: usize, len: usize) -> Result<Vec<u8>, BlobError> {
-        if let Some(Some(b)) = self.blobs.get_mut(id) {
+    pub fn read(&self, id: usize, start: usize, len: usize) -> Result<Vec<u8>, 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];
-            b.read(start, len, &mut read)?; // propagate error to caller
+        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)
-        } else {
-            Err(BlobError::BlobDNE)
-        }
     }
 }
 
@@ -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']);