use crate::{
address::segment::segment_page_iterator::SegmentPageIterator,
allocator::Allocator,
error::PERes,
id::{RecRef, SegmentId},
journal::{records::FreedPage, Journal, JournalId},
};
use std::{
cmp::Ordering,
collections::{hash_map::Entry, HashMap},
sync::Mutex,
};
pub type SnapshotId = u64;
#[derive(Clone, Debug)]
pub struct SnapshotEntry {
id: RecRef,
case: EntryCase,
}
#[derive(Clone, Debug, PartialEq)]
pub struct Change {
pub pos: u64,
pub version: u16,
}
#[derive(Clone, Debug, PartialEq)]
pub enum EntryCase {
Change(Change),
Insert,
}
impl SnapshotEntry {
pub fn change(id: &RecRef, pos: u64, version: u16) -> SnapshotEntry {
SnapshotEntry {
id: *id,
case: EntryCase::Change(Change { pos, version }),
}
}
pub fn insert(id: &RecRef) -> SnapshotEntry {
SnapshotEntry {
id: *id,
case: EntryCase::Insert,
}
}
}
#[derive(Clone, Debug)]
pub struct SegmentSnapshot {
name: String,
id: SegmentId,
first_page: u64,
}
impl SegmentSnapshot {
pub fn new(name: &str, id: SegmentId, first_page: u64) -> SegmentSnapshot {
SegmentSnapshot {
name: name.to_string(),
id,
first_page,
}
}
}
#[derive(Clone, Debug)]
pub struct SnapshotData {
snapshot_id: SnapshotId,
journal_id: Option<JournalId>,
entries: Option<Vec<SnapshotEntry>>,
freed_pages: Option<Vec<FreedPage>>,
segments: Option<HashMap<SegmentId, SegmentSnapshot>>,
segments_name: Option<HashMap<String, SegmentSnapshot>>,
reference_count: u32,
}
impl SnapshotData {
fn new(
id: SnapshotId,
journal_id: JournalId,
entries: Vec<SnapshotEntry>,
freed_pages: Vec<FreedPage>,
reference_count: u32,
) -> Self {
Self {
snapshot_id: id,
journal_id: Some(journal_id),
entries: Some(entries),
freed_pages: Some(freed_pages),
segments: None,
segments_name: None,
reference_count,
}
}
fn new_read(id: SnapshotId, reference_count: u32) -> Self {
Self {
snapshot_id: id,
journal_id: None,
entries: None,
freed_pages: None,
segments: None,
segments_name: None,
reference_count,
}
}
}
#[derive(Clone, Debug, PartialEq)]
pub struct RecordVersion {
snapshot_id: SnapshotId,
pub case: EntryCase,
}
#[derive(Debug)]
pub struct InternalSnapshots {
mapping: HashMap<RecRef, Vec<RecordVersion>>,
active_snapshots: Vec<SnapshotData>,
snapshot_sequence: u64,
}
impl InternalSnapshots {
fn next_snapshot_id(&mut self) -> SnapshotId {
let snapshot_id = self.snapshot_sequence;
if self.snapshot_sequence == u64::MAX {
self.snapshot_sequence = 0;
} else {
self.snapshot_sequence += 1;
}
snapshot_id
}
fn search(&self, snapshot_id: SnapshotId) -> Result<usize, usize> {
let snapshot_sequence = self.snapshot_sequence;
self.active_snapshots
.binary_search_by(|n| search(n.snapshot_id, snapshot_id, snapshot_sequence))
}
fn acquire_last_snapshot(&mut self) -> SnapshotId {
let last = self.snapshot_sequence - 1;
self.acquire_snapshot(last);
last
}
fn acquire_snapshot(&mut self, snapshot_id: SnapshotId) {
if let Ok(pos) = self.search(snapshot_id) {
if let Some(p) = self.active_snapshots.get_mut(pos) {
p.reference_count += 1;
} else {
unreachable!()
}
} else {
panic!("try to acquire a not existing snapshot")
}
}
fn clear_from(&mut self, snapshot_id: SnapshotId) -> PERes<(Option<Vec<FreedPage>>, Option<Vec<JournalId>>)> {
let snapshot_sequence = self.snapshot_sequence;
if let Ok(index) = self.search(snapshot_id) {
let size = index + 1;
let mut free_pages: Option<Vec<FreedPage>> = None;
let mut journal_ids: Option<Vec<JournalId>> = None;
let left_off = self.active_snapshots.split_off(size);
let old = std::mem::replace(&mut self.active_snapshots, left_off);
for tx in old {
if let Some(id) = tx.journal_id.clone() {
if let Some(ids) = &mut journal_ids {
ids.push(id.clone());
} else {
journal_ids = Some(vec![id.clone()]);
}
}
if let Some(entries) = tx.entries {
for record in entries {
if let Entry::Occupied(mut v) = self.mapping.entry(record.id) {
match v
.get()
.binary_search_by(|n| search(n.snapshot_id, snapshot_id, snapshot_sequence))
{
Ok(index) => {
v.get_mut().drain(..=index);
if v.get().is_empty() {
v.remove();
}
}
Err(_index) => {
v.remove();
}
}
}
}
}
if let Some(pages) = tx.freed_pages {
if let Some(free) = &mut free_pages {
free.extend(pages);
} else {
free_pages = Some(pages);
}
}
}
Ok((free_pages, journal_ids))
} else {
Ok((None, None))
}
}
fn release(&mut self, snapshot_id: SnapshotId) -> PERes<(Option<Vec<FreedPage>>, Option<Vec<JournalId>>)> {
let mut clear_id = None;
if let Ok(index) = self.search(snapshot_id) {
let mut loop_index = index;
while let Some(snap) = self.active_snapshots.get_mut(loop_index) {
snap.reference_count -= 1;
if snap.reference_count > 0 {
break;
}
clear_id = Some(snap.snapshot_id);
loop_index += 1;
}
}
if let Some(c_id) = clear_id {
self.clear_from(c_id)
} else {
Ok((None, None))
}
}
}
pub struct Snapshots {
lock: Mutex<InternalSnapshots>,
}
pub fn search(value: u64, value1: u64, top: u64) -> Ordering {
if value > top {
if value1 > top {
value.cmp(&value1)
} else {
Ordering::Less
}
} else if value1 > top {
Ordering::Greater
} else {
value.cmp(&value1)
}
}
impl Default for Snapshots {
fn default() -> Snapshots {
Self::new()
}
}
impl Snapshots {
pub fn new() -> Snapshots {
Snapshots {
lock: Mutex::new(InternalSnapshots {
mapping: HashMap::new(),
active_snapshots: Vec::new(),
snapshot_sequence: 0,
}),
}
}
pub fn acquire(&self, snapshot_id: SnapshotId) -> PERes<()> {
let mut lock = self.lock.lock()?;
lock.acquire_snapshot(snapshot_id);
Ok(())
}
pub fn current_snapshot(&self) -> PERes<SnapshotId> {
let mut lock = self.lock.lock()?;
if lock.active_snapshots.is_empty() {
let snapshot_id = lock.next_snapshot_id();
let reference_count = if lock.active_snapshots.is_empty() { 1 } else { 2 };
let snapshot = SnapshotData::new_read(snapshot_id, reference_count);
if let Err(index) = lock.search(snapshot_id) {
lock.active_snapshots.insert(index, snapshot);
}
Ok(snapshot_id)
} else {
Ok(lock.acquire_last_snapshot())
}
}
pub fn read_snapshot(&self) -> PERes<SnapshotId> {
let mut lock = self.lock.lock()?;
let snapshot_id = lock.next_snapshot_id();
let reference_count = if lock.active_snapshots.is_empty() { 1 } else { 2 };
let snapshot = SnapshotData::new_read(snapshot_id, reference_count);
if let Err(index) = lock.search(snapshot_id) {
lock.active_snapshots.insert(index, snapshot);
}
Ok(snapshot_id)
}
pub fn snapshot(
&self,
entries: Vec<SnapshotEntry>,
freed_pages: Vec<FreedPage>,
journal_id: JournalId,
) -> PERes<SnapshotId> {
let mut lock = self.lock.lock()?;
let snapshot_id = lock.next_snapshot_id();
for entry in &entries {
let to_add = RecordVersion {
snapshot_id,
case: entry.case.clone(),
};
let snapshot_sequence = lock.snapshot_sequence;
match lock.mapping.entry(entry.id) {
Entry::Occupied(mut v) => {
let vec = v.get_mut();
if let Err(index) = vec.binary_search_by(|n| search(n.snapshot_id, snapshot_id, snapshot_sequence))
{
vec.insert(index, to_add);
}
}
Entry::Vacant(e) => {
e.insert(vec![to_add]);
}
}
}
let reference_count = if lock.active_snapshots.is_empty() { 1 } else { 2 };
let snapshot = SnapshotData::new(snapshot_id, journal_id, entries, freed_pages, reference_count);
if let Err(index) = lock.search(snapshot_id) {
lock.active_snapshots.insert(index, snapshot);
}
Ok(snapshot_id)
}
pub fn fill_segments(&self, snapshot_id: SnapshotId, segments: &[SegmentSnapshot]) -> PERes<()> {
let mut segments_id = HashMap::new();
let mut segments_name = HashMap::new();
for segment in segments {
segments_id.insert(segment.id, segment.clone());
segments_name.insert(segment.name.clone(), segment.clone());
}
let mut lock = self.lock.lock()?;
if let Ok(index) = lock.search(snapshot_id) {
if let Some(snap) = lock.active_snapshots.get_mut(index) {
snap.segments = Some(segments_id);
snap.segments_name = Some(segments_name);
}
}
Ok(())
}
pub fn solve_segment_id(&self, snapshot_id: SnapshotId, segment: &str) -> PERes<Option<SegmentId>> {
let mut lock = self.lock.lock()?;
let res = if let Ok(index) = lock.search(snapshot_id) {
if let Some(snap) = lock.active_snapshots.get_mut(index) {
if let Some(segs) = &snap.segments_name {
segs.get(segment).map(|sd| sd.id)
} else {
None
}
} else {
None
}
} else {
None
};
Ok(res)
}
pub fn solve_segment_name(&self, snapshot_id: SnapshotId, segment_id: SegmentId) -> PERes<Option<String>> {
let mut lock = self.lock.lock()?;
let res = if let Ok(index) = lock.search(snapshot_id) {
if let Some(snap) = lock.active_snapshots.get_mut(index) {
if let Some(segs) = &snap.segments {
segs.get(&segment_id).map(|sd| sd.name.clone())
} else {
None
}
} else {
None
}
} else {
None
};
Ok(res)
}
pub fn scan(&self, snapshot_id: SnapshotId, segment_id: SegmentId) -> PERes<Option<SegmentPageIterator>> {
let mut lock = self.lock.lock()?;
let res = if let Ok(index) = lock.search(snapshot_id) {
if let Some(snap) = lock.active_snapshots.get_mut(index) {
if let Some(segs) = &snap.segments {
segs.get(&segment_id)
.map(|sd| SegmentPageIterator::snapshot(sd.first_page))
} else {
None
}
} else {
None
}
} else {
None
};
Ok(res)
}
pub fn list(&self, snapshot_id: SnapshotId) -> PERes<Vec<(String, SegmentId)>> {
let mut lock = self.lock.lock()?;
let res = if let Ok(index) = lock.search(snapshot_id) {
if let Some(snap) = lock.active_snapshots.get_mut(index) {
if let Some(segs) = &snap.segments {
segs.values()
.map(|data| (data.name.clone(), data.id))
.collect::<Vec<_>>()
} else {
Vec::new()
}
} else {
Vec::new()
}
} else {
Vec::new()
};
Ok(res)
}
pub fn read(&self, snapshot_id: SnapshotId, id: &RecRef) -> PERes<Option<RecordVersion>> {
let lock = self.lock.lock()?;
let snapshot_sequence = lock.snapshot_sequence;
Ok(if let Some(v) = lock.mapping.get(id) {
let index = match v.binary_search_by(|n| search(n.snapshot_id, snapshot_id, snapshot_sequence)) {
Ok(index) => index,
Err(index) => index,
};
v.get(index).cloned()
} else {
None
})
}
#[allow(unused)]
fn clear_from(&self, snapshot_id: SnapshotId) -> PERes<(Option<Vec<FreedPage>>, Option<Vec<JournalId>>)> {
self.lock.lock()?.clear_from(snapshot_id)
}
fn release(&self, snapshot_id: SnapshotId) -> PERes<(Option<Vec<FreedPage>>, Option<Vec<JournalId>>)> {
self.lock.lock()?.release(snapshot_id)
}
}
pub fn release_snapshot(id: SnapshotId, snapshots: &Snapshots, allocator: &Allocator, journal: &Journal) -> PERes<()> {
let (to_free, to_clean) = snapshots.release(id)?;
if let Some(to_free) = to_free {
allocator.free_pages(to_free.into_iter().map(|p| p.page))?;
}
if let Some(to_clean) = to_clean {
journal.finished_to_clean(&to_clean)?;
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::{search, Change, EntryCase, RecordVersion, SegmentSnapshot, SnapshotEntry, Snapshots};
use crate::{
id::{RecRef, SegmentId},
journal::{records::FreedPage, JournalId},
};
use std::cmp::Ordering;
#[test]
fn test_search() {
assert_eq!(search(10, 20, 40), Ordering::Less);
assert_eq!(search(20, 10, 40), Ordering::Greater);
assert_eq!(search(10, 30, 20), Ordering::Greater);
assert_eq!(search(30, 10, 20), Ordering::Less);
assert_eq!(search(20, 20, 20), Ordering::Equal);
assert_eq!(search(20, 19, 20), Ordering::Greater);
assert_eq!(search(20, 21, 20), Ordering::Greater);
assert_eq!(search(21, 21, 20), Ordering::Equal);
assert_eq!(search(19, 19, 20), Ordering::Equal);
}
#[test]
fn add_and_read() {
let snap = Snapshots::new();
let mut records = Vec::new();
let mut freed_pages = Vec::new();
for x in 0..3 {
records.push(SnapshotEntry::change(&RecRef::new(10, x), x as u64, x as u16));
freed_pages.push(FreedPage::new(x as u64));
}
let tx = snap
.snapshot(records.clone(), freed_pages.clone(), JournalId::new(0, 0))
.unwrap();
snap.clear_from(tx).unwrap();
let tx = snap.snapshot(records, freed_pages, JournalId::new(0, 0)).unwrap();
assert_eq!(
snap.read(tx - 1, &RecRef::new(10, 2)).unwrap(),
Some(RecordVersion {
snapshot_id: 1,
case: EntryCase::Change(Change { pos: 2, version: 2 }),
})
);
assert_eq!(
snap.read(tx - 1, &RecRef::new(10, 1)).unwrap(),
Some(RecordVersion {
snapshot_id: 1,
case: EntryCase::Change(Change { pos: 1, version: 1 }),
})
);
assert_eq!(
snap.read(tx, &RecRef::new(10, 2)).unwrap(),
Some(RecordVersion {
snapshot_id: 1,
case: EntryCase::Change(Change { pos: 2, version: 2 }),
})
);
assert_eq!(snap.read(tx + 1, &RecRef::new(10, 2)).unwrap(), None);
assert_eq!(snap.read(tx + 1, &RecRef::new(10, 10)).unwrap(), None);
}
#[test]
fn add_and_read_multiple_tx() {
let snap = Snapshots::new();
let mut txs = Vec::new();
for t in 0..5 {
let mut records = Vec::new();
let mut freed_pages = Vec::new();
for x in 0..10 {
if x != t {
records.push(SnapshotEntry::change(
&RecRef::new(10, x),
(10 * t + x) as u64,
(10 * t + x) as u16,
));
freed_pages.push(FreedPage::new((10 * t + x) as u64));
}
}
txs.push(snap.snapshot(records, freed_pages, JournalId::new(0, 0)).unwrap());
}
assert_eq!(
snap.read(txs[2], &RecRef::new(10, 2)).unwrap(),
Some(RecordVersion {
snapshot_id: txs[3],
case: EntryCase::Change(Change { pos: 32, version: 32 }),
})
);
assert_eq!(
snap.read(txs[3], &RecRef::new(10, 3)).unwrap(),
Some(RecordVersion {
snapshot_id: txs[4],
case: EntryCase::Change(Change { pos: 43, version: 43 }),
})
);
}
#[test]
fn test_snapshot_reference_count() {
let snap = Snapshots::new();
let mut records = Vec::new();
let mut freed_pages = Vec::new();
for x in 0..3 {
records.push(SnapshotEntry::change(&RecRef::new(10, x), x as u64, x as u16));
freed_pages.push(FreedPage::new(x as u64));
}
let first = snap.read_snapshot().unwrap();
let tx = snap
.snapshot(records.clone(), freed_pages, JournalId::new(0, 0))
.unwrap();
let last = snap.read_snapshot().unwrap();
snap.release(tx).unwrap();
assert_eq!(
snap.read(first, &RecRef::new(10, 2)).unwrap(),
Some(RecordVersion {
snapshot_id: 1,
case: EntryCase::Change(Change { pos: 2, version: 2 }),
})
);
snap.release(first).unwrap();
assert_eq!(snap.read(last, &RecRef::new(10, 2)).unwrap(), None);
}
#[test]
fn test_clanup_after_release() {
let snap = Snapshots::new();
let mut snapshots_id = Vec::new();
for s in 0..100 {
let mut records = Vec::new();
let mut freed_pages = Vec::new();
for x in 0..3 {
records.push(SnapshotEntry::change(&RecRef::new(s % 10, x), x as u64, x as u16));
freed_pages.push(FreedPage::new(x as u64));
}
let tx = snap
.snapshot(records.clone(), freed_pages, JournalId::new(0, 0))
.unwrap();
snapshots_id.push(tx);
}
for s in snapshots_id.into_iter().rev() {
snap.release(s).unwrap();
}
{
let data = snap.lock.lock().unwrap();
assert_eq!(data.active_snapshots.len(), 0);
assert_eq!(data.mapping.len(), 0);
}
let mut snapshots_id = Vec::new();
for s in 0..100 {
let mut records = Vec::new();
let mut freed_pages = Vec::new();
for x in 0..3 {
records.push(SnapshotEntry::change(&RecRef::new(s % 10, x), x as u64, x as u16));
freed_pages.push(FreedPage::new(x as u64));
}
let tx = snap
.snapshot(records.clone(), freed_pages, JournalId::new(0, 0))
.unwrap();
snapshots_id.push(tx);
}
for s in snapshots_id {
snap.release(s).unwrap();
}
let data = snap.lock.lock().unwrap();
assert_eq!(data.active_snapshots.len(), 0);
assert_eq!(data.mapping.len(), 0);
}
#[test]
fn test_snapshot_release_clear() {
let snap = Snapshots::new();
let mut snaps = Vec::new();
for i in 0..10 {
let id = snap.read_snapshot().unwrap();
let mut segments = Vec::new();
let seg_id = SegmentId::new(i);
segments.push(SegmentSnapshot::new("one", seg_id, i.into()));
segments.push(SegmentSnapshot::new("two", seg_id, i.into()));
snap.fill_segments(id, &segments).expect("fill works");
snaps.push(id);
}
snap.release(snaps[0]).expect("release works");
snap.release(snaps[1]).expect("release works");
snap.release(snaps[2]).expect("release works");
let solved = snap.solve_segment_id(snaps[3], "one").expect("solve id correctly");
assert_eq!(solved, Some(SegmentId::new(3)));
let data = snap.lock.lock().unwrap();
assert_eq!(data.active_snapshots.len(), 7);
}
#[test]
pub fn test_segment_snapshot_fill() {
let snap = Snapshots::new();
let id = snap.read_snapshot().unwrap();
let mut segments = Vec::new();
let seg_id = SegmentId::new(10);
segments.push(SegmentSnapshot::new("one", seg_id, 10));
snap.fill_segments(id, &segments).expect("fill works");
let solved = snap.solve_segment_id(id, "one").expect("solve id correctly");
assert_eq!(solved, Some(seg_id));
let solved = snap.solve_segment_name(id, seg_id).expect("solve id correctly");
assert_eq!(solved, Some("one".to_string()));
let solved = snap.solve_segment_id(id, "two").expect("solve id correctly");
assert_eq!(solved, None);
let solved = snap.scan(id, seg_id).expect("solve id correctly");
assert!(solved.is_some());
snap.release(id).expect("release works");
}
#[test]
pub fn test_segment_snapshot_fill_multiple() {
let snap = Snapshots::new();
let mut snaps = Vec::new();
for i in 0..10 {
let id = snap.read_snapshot().unwrap();
let mut segments = Vec::new();
let seg_id = SegmentId::new(i);
segments.push(SegmentSnapshot::new("one", seg_id, i.into()));
segments.push(SegmentSnapshot::new("two", seg_id, i.into()));
snap.fill_segments(id, &segments).expect("fill works");
snaps.push(id);
}
let solved = snap.solve_segment_id(snaps[3], "one").expect("solve id correctly");
assert_eq!(solved, Some(SegmentId::new(3)));
let solved = snap.solve_segment_id(snaps[3], "two").expect("solve id correctly");
assert_eq!(solved, Some(SegmentId::new(3)));
let solved = snap.solve_segment_id(snaps[6], "one").expect("solve id correctly");
assert_eq!(solved, Some(SegmentId::new(6)));
let solved = snap.solve_segment_id(snaps[6], "two").expect("solve id correctly");
assert_eq!(solved, Some(SegmentId::new(6)));
}
}