week2 finished

mini-lsm-starter/src/manifest.rs

@@ -1,15 +1,14 @@
-#![allow(dead_code)] // REMOVE THIS LINE after fully implementing this functionality
-
 use std::fs::{File, OpenOptions};
 use std::io::{Read, Write};
 use std::path::Path;
 use std::sync::Arc;
 
-use crate::compact::CompactionTask;
-use anyhow::{Context, Result};
+use anyhow::{bail, Context, Result};
+use bytes::{Buf, BufMut};
 use parking_lot::{Mutex, MutexGuard};
 use serde::{Deserialize, Serialize};
-use serde_json::Deserializer;
+
+use crate::compact::CompactionTask;
 
 pub struct Manifest {
     file: Arc<Mutex<File>>,
@@ -44,10 +43,18 @@ impl Manifest {
             .context("failed to recover manifest")?;
         let mut buf = Vec::new();
         file.read_to_end(&mut buf)?;
-        let mut stream = Deserializer::from_slice(&buf).into_iter::<ManifestRecord>();
+        let mut buf_ptr = buf.as_slice();
         let mut records = Vec::new();
-        while let Some(x) = stream.next() {
-            records.push(x?);
+        while buf_ptr.has_remaining() {
+            let len = buf_ptr.get_u64();
+            let slice = &buf_ptr[..len as usize];
+            let json = serde_json::from_slice::<ManifestRecord>(slice)?;
+            buf_ptr.advance(len as usize);
+            let checksum = buf_ptr.get_u32();
+            if checksum != crc32fast::hash(slice) {
+                bail!("checksum mismatched!");
+            }
+            records.push(json);
         }
         Ok((
             Self {
@@ -67,7 +74,10 @@ impl Manifest {
 
     pub fn add_record_when_init(&self, record: ManifestRecord) -> Result<()> {
         let mut file = self.file.lock();
-        let buf = serde_json::to_vec(&record)?;
+        let mut buf = serde_json::to_vec(&record)?;
+        let hash = crc32fast::hash(&buf);
+        file.write_all(&(buf.len() as u64).to_be_bytes())?;
+        buf.put_u32(hash);
         file.write_all(&buf)?;
         file.sync_all()?;
         Ok(())

mini-lsm-starter/src/wal.rs

@@ -1,11 +1,10 @@
-#![allow(dead_code)] // REMOVE THIS LINE after fully implementing this functionality
-
 use std::fs::{File, OpenOptions};
+use std::hash::Hasher;
 use std::io::{Read, Write};
 use std::path::Path;
 use std::sync::Arc;
 
-use anyhow::{Context, Result};
+use anyhow::{bail, Context, Result};
 use bytes::{Buf, BufMut, Bytes};
 use crossbeam_skiplist::SkipMap;
 use parking_lot::Mutex;
@@ -39,12 +38,21 @@ impl Wal {
         file.read_to_end(&mut buf)?;
         let mut rbuf: &[u8] = buf.as_slice();
         while rbuf.has_remaining() {
+            let mut hasher = crc32fast::Hasher::new();
             let key_len = rbuf.get_u16() as usize;
+            hasher.write_u16(key_len as u16);
             let key = Bytes::copy_from_slice(&rbuf[..key_len]);
+            hasher.write(&key);
             rbuf.advance(key_len);
             let value_len = rbuf.get_u16() as usize;
+            hasher.write_u16(value_len as u16);
             let value = Bytes::copy_from_slice(&rbuf[..value_len]);
+            hasher.write(&value);
             rbuf.advance(value_len);
+            let checksum = rbuf.get_u32();
+            if hasher.finalize() != checksum {
+                bail!("checksum mismatch");
+            }
             skiplist.insert(key, value);
         }
         Ok(Self {
@@ -56,10 +64,17 @@ impl Wal {
         let mut file = self.file.lock();
         let mut buf: Vec<u8> =
             Vec::with_capacity(key.len() + value.len() + std::mem::size_of::<u16>());
+        let mut hasher = crc32fast::Hasher::new();
+        hasher.write_u16(key.len() as u16);
         buf.put_u16(key.len() as u16);
+        hasher.write(key);
         buf.put_slice(key);
+        hasher.write_u16(value.len() as u16);
         buf.put_u16(value.len() as u16);
         buf.put_slice(value);
+        hasher.write(value);
+        // add checksum: week 2 day 7
+        buf.put_u32(hasher.finalize());
         file.write_all(&buf)?;
         Ok(())
     }