-
-
Notifications
You must be signed in to change notification settings - Fork 35
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
add cache for decrypting dir entry names
add cache for reading meta for dir entries do some operations concurrently when creating a nod on encryption use block index as AAD to make sure blocks are not reordered add bech tests force unmount
- Loading branch information
1 parent
125c007
commit 73d66ea
Showing
13 changed files
with
1,599 additions
and
1,029 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,140 @@ | ||
| use std::io::{self, Read}; | ||
|
|
||
| use okaywal::{Entry, EntryId, LogManager, ReadChunkResult, SegmentReader, WriteAheadLog}; | ||
|
|
||
| fn main() -> io::Result<()> { | ||
| // begin rustme snippet: readme-example | ||
| // Open a log using an Checkpointer that echoes the information passed into each | ||
| // function that the Checkpointer trait defines. | ||
| let log = WriteAheadLog::recover("/tmp/my-log", LoggingCheckpointer)?; | ||
| log.checkpoint_active()?; | ||
|
|
||
| // Begin writing an entry to the log. | ||
| let mut writer = log.begin_entry()?; | ||
|
|
||
| // Each entry is one or more chunks of data. Each chunk can be individually | ||
| // addressed using its LogPosition. | ||
| for i in 0..10 { | ||
| writer.write_chunk(format!("this is the {} entry", i).as_bytes())?; | ||
| } | ||
| // let record = writer.write_chunk("this is the first entry".as_bytes())?; | ||
|
|
||
| // To fully flush all written bytes to disk and make the new entry | ||
| // resilliant to a crash, the writer must be committed. | ||
| writer.commit()?; | ||
| // end rustme snippet | ||
|
|
||
| // log.checkpoint_active()?; | ||
|
|
||
| // Let's reopen the log. During this process, | ||
| // LoggingCheckpointer::should_recover_segment will be invoked for each segment | ||
| // file that has not been checkpointed yet. In this example, it will be called | ||
| // once. Once the Checkpointer confirms the data should be recovered, | ||
| // LoggingCheckpointer::recover will be invoked once for each entry in the WAL | ||
| // that hasn't been previously checkpointed. | ||
| // drop(log); | ||
| // let log = WriteAheadLog::recover("/tmp/my-log", LoggingCheckpointer)?; | ||
|
|
||
| // We can use the previously returned DataRecord to read the original data. | ||
| // let mut reader = log.read_at(record.position)?; | ||
| // let mut buffer = vec![0; usize::try_from(record.length).unwrap()]; | ||
| // reader.read_exact(&mut buffer)?; | ||
| // println!( | ||
| // "Data read from log: {}", | ||
| // String::from_utf8(buffer).expect("invalid utf-8") | ||
| // ); | ||
|
|
||
| // Cleanup | ||
| // drop(reader); | ||
| // drop(log); | ||
| // std::fs::remove_dir_all("my-log")?; | ||
|
|
||
| log.shutdown()?; | ||
| // drop(log); | ||
|
|
||
| Ok(()) | ||
| } | ||
|
|
||
| #[derive(Debug)] | ||
| struct LoggingCheckpointer; | ||
|
|
||
| impl LogManager for LoggingCheckpointer { | ||
| fn recover(&mut self, entry: &mut Entry<'_>) -> io::Result<()> { | ||
| // This example uses read_all_chunks to load the entire entry into | ||
| // memory for simplicity. The crate also supports reading each chunk | ||
| // individually to minimize memory usage. | ||
| if let Some(all_chunks) = entry.read_all_chunks()? { | ||
| // Convert the Vec<u8>'s to Strings. | ||
| let all_chunks = all_chunks | ||
| .into_iter() | ||
| .map(String::from_utf8) | ||
| .collect::<Result<Vec<String>, _>>() | ||
| .expect("invalid utf-8"); | ||
| println!( | ||
| "LoggingCheckpointer::recover(entry_id: {:?}, data: {:?})", | ||
| entry.id(), | ||
| all_chunks, | ||
| ); | ||
| } else { | ||
| // This entry wasn't completely written. This could happen if a | ||
| // power outage or crash occurs while writing an entry. | ||
| } | ||
|
|
||
| Ok(()) | ||
| } | ||
|
|
||
| fn checkpoint_to( | ||
| &mut self, | ||
| last_checkpointed_id: EntryId, | ||
| _checkpointed_entries: &mut SegmentReader, | ||
| _wal: &WriteAheadLog, | ||
| ) -> io::Result<()> { | ||
| // checkpoint_to is called once enough data has been written to the | ||
| // WriteAheadLog. After this function returns, the log will recycle the | ||
| // file containing the entries being checkpointed. | ||
| // | ||
| // This function is where the entries must be persisted to the storage | ||
| // layer the WriteAheadLog is sitting in front of. To ensure ACID | ||
| // compliance of the combination of the WAL and the storage layer, the | ||
| // storage layer must be fully resilliant to losing any changes made by | ||
| // the checkpointed entries before this function returns. | ||
| println!("LoggingCheckpointer::checkpoint_to({last_checkpointed_id:?}"); | ||
| while let Some(mut entry) = _checkpointed_entries.read_entry()? { | ||
| println!( | ||
| "LoggingCheckpointer::checkpoint_to(entry_id: {:?})", | ||
| entry.id() | ||
| ); | ||
| while let chunk = entry.read_chunk() { | ||
| match chunk { | ||
| Ok(res) => match res { | ||
| ReadChunkResult::Chunk(chunk) => { | ||
| println!( | ||
| "LoggingCheckpointer::checkpoint_to(chunk bytes: {:?})", | ||
| chunk.bytes_remaining() | ||
| ); | ||
| } | ||
| ReadChunkResult::EndOfEntry => { | ||
| break; | ||
| } | ||
| ReadChunkResult::AbortedEntry => { | ||
| continue; | ||
| } | ||
| }, | ||
| Err(err) => {} | ||
| } | ||
| } | ||
| } | ||
| Ok(()) | ||
| } | ||
| } | ||
|
|
||
| #[test] | ||
| fn test() -> io::Result<()> { | ||
| // Clean up any previous runs of this example. | ||
| // let path = std::path::Path::new("/tmp/my-log"); | ||
| // if path.exists() { | ||
| // std::fs::remove_dir_all("/tmp/my-log")?; | ||
| // } | ||
|
|
||
| main() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.