snapshot.rs at dd0268a ⋅ tglman/persy

use crate::{
    allocator::Allocator,
    error::PRes,
    id::{RecRef, SegmentId},
    journal::{Journal, JournalId},
    segment::SnapshotSegmentPageIterator,
    transaction::FreedPage,
};
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.clone(),
            case: EntryCase::Change(Change { pos, version }),
        }
    }
    pub fn insert(id: &RecRef) -> SnapshotEntry {
        SnapshotEntry {
            id: id.clone(),
            case: EntryCase::Insert,
        }
    }
}

#[derive(Clone, Debug)]
pub struct SegmentSnapshop {
    name: String,
    id: SegmentId,
    first_page: u64,
}
impl SegmentSnapshop {
    pub fn new(name: &str, id: SegmentId, first_page: u64) -> SegmentSnapshop {
        SegmentSnapshop {
            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, SegmentSnapshop>>,
    segments_name: Option<HashMap<String, SegmentSnapshop>>,
    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,
}

pub struct InternalSnapshots {
    mapping: HashMap<RecRef, Vec<RecordVersion>>,
    active_snapshots: Vec<SnapshotData>,
    snapshot_sequence: u64,
}
impl InternalSnapshots {
    fn next_snapshot_id(&mut self) -> SnapshotId {
        //TODO: this may overflow and should start again from 0 double check overflow failures
        let snapshot_id = self.snapshot_sequence;
        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")
        }
    }
}

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) -> PRes<()> {
        let mut lock = self.lock.lock()?;
        lock.acquire_snapshot(snapshot_id);
        Ok(())
    }

    pub fn current_snapshot(&self) -> PRes<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) -> PRes<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,
    ) -> PRes<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.clone()) {
                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: &[SegmentSnapshop]) -> PRes<()> {
        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) -> PRes<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 {
                    if let Some(sd) = segs.get(segment) {
                        Some(sd.id)
                    } else {
                        None
                    }
                } else {
                    None
                }
            } else {
                None
            }
        } else {
            None
        };
        Ok(res)
    }

    pub fn solve_segment_name(&self, snapshot_id: SnapshotId, segment_id: SegmentId) -> PRes<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 {
                    if let Some(sd) = segs.get(&segment_id) {
                        Some(sd.name.clone())
                    } else {
                        None
                    }
                } else {
                    None
                }
            } else {
                None
            }
        } else {
            None
        };
        Ok(res)
    }

    pub fn scan(&self, snapshot_id: SnapshotId, segment_id: SegmentId) -> PRes<Option<SnapshotSegmentPageIterator>> {
        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 {
                    if let Some(sd) = segs.get(&segment_id) {
                        Some(SnapshotSegmentPageIterator::new(sd.first_page))
                    } else {
                        None
                    }
                } else {
                    None
                }
            } else {
                None
            }
        } else {
            None
        };
        Ok(res)
    }

    pub fn list(&self, snapshot_id: SnapshotId) -> PRes<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) -> PRes<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
        })
    }

    fn clear_from(&self, snapshot_id: SnapshotId) -> PRes<(Vec<FreedPage>, Vec<JournalId>)> {
        let mut lock = self.lock.lock()?;
        let snapshot_sequence = lock.snapshot_sequence;
        let mut free_pages = Vec::new();
        let mut journal_ids = Vec::new();
        if let Ok(index) = lock.search(snapshot_id) {
            let size = index + 1;
            let mut old = Vec::with_capacity(size);
            old.extend_from_slice(&lock.active_snapshots[0..size]);
            lock.active_snapshots = lock.active_snapshots.split_off(size);
            for tx in old {
                if let Some(id) = tx.journal_id.clone() {
                    journal_ids.push(id.clone());
                }
                if let Some(entries) = tx.entries {
                    for record in entries {
                        if let Some(v) = lock.mapping.get_mut(&record.id) {
                            if let Ok(index) =
                                v.binary_search_by(|n| search(n.snapshot_id, snapshot_id, snapshot_sequence))
                            {
                                v.truncate(index);
                            }
                        }
                    }
                }
                if let Some(pages) = tx.freed_pages {
                    free_pages.extend(pages);
                }
            }
        }
        Ok((free_pages, journal_ids))
    }

    fn release(&self, snapshot_id: SnapshotId) -> PRes<(Vec<FreedPage>, Vec<JournalId>)> {
        //TODO: This work fine but can cause problems if double release is called for the same id,
        //to refactor to something a bit more safe
        let mut clear_id = None;
        {
            let mut lock = self.lock.lock()?;
            if let Ok(index) = lock.search(snapshot_id) {
                let mut loop_index = index;
                while let Some(snap) = lock.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((Vec::new(), Vec::new()))
        }
    }
}

pub fn release_snapshot(id: SnapshotId, snapshots: &Snapshots, allocator: &Allocator, journal: &Journal) -> PRes<()> {
    let (to_free, to_clean) = snapshots.release(id)?;

    for page in to_free {
        allocator.free(page.page)?;
    }
    journal.finished_to_clean(&to_clean)?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::{search, Change, EntryCase, RecordVersion, SegmentSnapshop, SnapshotEntry, Snapshots};
    use crate::{
        id::{RecRef, SegmentId},
        journal::JournalId,
        transaction::FreedPage,
    };
    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_mutliple_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 {
                // I skip a record for the specific tx because i need a missing record for the test
                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_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(SegmentSnapshop::new("one", seg_id, i.into()));
            segments.push(SegmentSnapshop::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)));
    }

    #[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(SegmentSnapshop::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(SegmentSnapshop::new("one", seg_id, i.into()));
            segments.push(SegmentSnapshop::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)));
    }
}

Read our documentation on viewing source code .

Navigation	Overlay
`t` Navigate files	`h` Toggle hits
`y` Change url to tip of branch	`m` Toggle misses
`b / v` Jump to prev/next hit line	`p` Toggle partial
`z / x` Jump to prev/next missed or partial line	`1..9` Toggle flags
`shift + o` Open current page in GitLab	`a` Toggle all on
`/ or ?` Show keyboard shortcuts dialog	`c` Toggle context lines or commits

1		use crate::{
2		allocator::Allocator,
3		error::PRes,
4		id::{RecRef, SegmentId},
5		journal::{Journal, JournalId},
6		segment::SnapshotSegmentPageIterator,
7		transaction::FreedPage,
8		};
9		use std::{
10		cmp::Ordering,
11		collections::{hash_map::Entry, HashMap},
12		sync::Mutex,
13		};
14
15		pub type SnapshotId = u64;
16
17	1	#[derive(Clone, Debug)]
18		pub struct SnapshotEntry {
19	1	id: RecRef,
20	1	case: EntryCase,
21		}
22
23	1	#[derive(Clone, Debug, PartialEq)]
24		pub struct Change {
25	1	pub pos: u64,
26	1	pub version: u16,
27		}
28	1	#[derive(Clone, Debug, PartialEq)]
29		pub enum EntryCase {
30	1	Change(Change),
31		Insert,
32		}
33
34		impl SnapshotEntry {
35	1	pub fn change(id: &RecRef, pos: u64, version: u16) -> SnapshotEntry {
36	1	SnapshotEntry {
37	1	id: id.clone(),
38	1	case: EntryCase::Change(Change { pos, version }),
39		}
40	1	}
41	1	pub fn insert(id: &RecRef) -> SnapshotEntry {
42	1	SnapshotEntry {
43	1	id: id.clone(),
44	1	case: EntryCase::Insert,
45		}
46	1	}
47		}
48
49	1	#[derive(Clone, Debug)]
50		pub struct SegmentSnapshop {
51	1	name: String,
52	1	id: SegmentId,
53	1	first_page: u64,
54		}
55		impl SegmentSnapshop {
56	1	pub fn new(name: &str, id: SegmentId, first_page: u64) -> SegmentSnapshop {
57	1	SegmentSnapshop {
58	1	name: name.to_string(),
59		id,
60		first_page,
61		}
62	1	}
63		}
64
65	1	#[derive(Clone, Debug)]
66		pub struct SnapshotData {
67	1	snapshot_id: SnapshotId,
68	1	journal_id: Option<JournalId>,
69	1	entries: Option<Vec<SnapshotEntry>>,
70	1	freed_pages: Option<Vec<FreedPage>>,
71	1	segments: Option<HashMap<SegmentId, SegmentSnapshop>>,
72	1	segments_name: Option<HashMap<String, SegmentSnapshop>>,
73	1	reference_count: u32,
74		}
75
76		impl SnapshotData {
77	1	fn new(
78		id: SnapshotId,
79		journal_id: JournalId,
80		entries: Vec<SnapshotEntry>,
81		freed_pages: Vec<FreedPage>,
82		reference_count: u32,
83		) -> Self {
84	1	Self {
85		snapshot_id: id,
86	1	journal_id: Some(journal_id),
87	1	entries: Some(entries),
88	1	freed_pages: Some(freed_pages),
89	1	segments: None,
90	1	segments_name: None,
91		reference_count,
92		}
93	1	}
94	1	fn new_read(id: SnapshotId, reference_count: u32) -> Self {
95	1	Self {
96		snapshot_id: id,
97	1	journal_id: None,
98	1	entries: None,
99	1	freed_pages: None,
100	1	segments: None,
101	1	segments_name: None,
102		reference_count,
103		}
104	1	}
105		}
106
107	1	#[derive(Clone, Debug, PartialEq)]
108		pub struct RecordVersion {
109	1	snapshot_id: SnapshotId,
110	1	pub case: EntryCase,
111		}
112
113		pub struct InternalSnapshots {
114		mapping: HashMap<RecRef, Vec<RecordVersion>>,
115		active_snapshots: Vec<SnapshotData>,
116		snapshot_sequence: u64,
117		}
118		impl InternalSnapshots {
119	1	fn next_snapshot_id(&mut self) -> SnapshotId {
120		//TODO: this may overflow and should start again from 0 double check overflow failures
121	1	let snapshot_id = self.snapshot_sequence;
122	1	self.snapshot_sequence += 1;
123		snapshot_id
124	1	}
125	1	fn search(&self, snapshot_id: SnapshotId) -> Result<usize, usize> {
126	1	let snapshot_sequence = self.snapshot_sequence;
127	1	self.active_snapshots
128	1	.binary_search_by(\|n\| search(n.snapshot_id, snapshot_id, snapshot_sequence))
129	1	}
130	1	fn acquire_last_snapshot(&mut self) -> SnapshotId {
131	1	let last = self.snapshot_sequence - 1;
132	1	self.acquire_snapshot(last);
133		last
134	1	}
135	1	fn acquire_snapshot(&mut self, snapshot_id: SnapshotId) {
136	1	if let Ok(pos) = self.search(snapshot_id) {
137	1	if let Some(p) = self.active_snapshots.get_mut(pos) {
138	1	p.reference_count += 1;
139		} else {
140	0	unreachable!()
141		}
142		} else {
143	0	panic!("try to acquire a not existing snapshot")
144		}
145	1	}
146		}
147
148		pub struct Snapshots {
149		lock: Mutex<InternalSnapshots>,
150		}
151
152	1	pub fn search(value: u64, value1: u64, top: u64) -> Ordering {
153	1	if value > top {
154	1	if value1 > top {
155	1	value.cmp(&value1)
156		} else {
157	1	Ordering::Less
158		}
159	1	} else if value1 > top {
160	1	Ordering::Greater
161		} else {
162	1	value.cmp(&value1)
163		}
164	1	}
165
166		impl Default for Snapshots {
167		fn default() -> Snapshots {
168		Self::new()
169		}
170		}
171
172		impl Snapshots {
173	1	pub fn new() -> Snapshots {
174	1	Snapshots {
175	1	lock: Mutex::new(InternalSnapshots {
176	1	mapping: HashMap::new(),
177	1	active_snapshots: Vec::new(),
178		snapshot_sequence: 0,
179	0	}),
180		}
181	1	}
182
183	1	pub fn acquire(&self, snapshot_id: SnapshotId) -> PRes<()> {
184	1	let mut lock = self.lock.lock()?;
185	1	lock.acquire_snapshot(snapshot_id);
186	1	Ok(())
187	1	}
188
189	1	pub fn current_snapshot(&self) -> PRes<SnapshotId> {
190	1	let mut lock = self.lock.lock()?;
191	1	if lock.active_snapshots.is_empty() {
192	1	let snapshot_id = lock.next_snapshot_id();
193
194	1	let reference_count = if lock.active_snapshots.is_empty() { 1 } else { 2 };
195	1	let snapshot = SnapshotData::new_read(snapshot_id, reference_count);
196	1	if let Err(index) = lock.search(snapshot_id) {
197	1	lock.active_snapshots.insert(index, snapshot);
198		}
199	1	Ok(snapshot_id)
200	1	} else {
201	1	Ok(lock.acquire_last_snapshot())
202		}
203	1	}
204
205	1	pub fn read_snapshot(&self) -> PRes<SnapshotId> {
206	1	let mut lock = self.lock.lock()?;
207	1	let snapshot_id = lock.next_snapshot_id();
208
209	1	let reference_count = if lock.active_snapshots.is_empty() { 1 } else { 2 };
210	1	let snapshot = SnapshotData::new_read(snapshot_id, reference_count);
211	1	if let Err(index) = lock.search(snapshot_id) {
212	1	lock.active_snapshots.insert(index, snapshot);
213		}
214	1	Ok(snapshot_id)
215	1	}
216
217	1	pub fn snapshot(
218		&self,
219		entries: Vec<SnapshotEntry>,
220		freed_pages: Vec<FreedPage>,
221		journal_id: JournalId,
222		) -> PRes<SnapshotId> {
223	1	let mut lock = self.lock.lock()?;
224	1	let snapshot_id = lock.next_snapshot_id();
225	1	for entry in &entries {
226	1	let to_add = RecordVersion {
227	1	snapshot_id,
228	1	case: entry.case.clone(),
229		};
230
231	1	let snapshot_sequence = lock.snapshot_sequence;
232	1	match lock.mapping.entry(entry.id.clone()) {
233	1	Entry::Occupied(mut v) => {
234	1	let vec = v.get_mut();
235	1	if let Err(index) = vec.binary_search_by(\|n\| search(n.snapshot_id, snapshot_id, snapshot_sequence))
236		{
237	1	vec.insert(index, to_add);
238		}
239		}
240	1	Entry::Vacant(e) => {
241	1	e.insert(vec![to_add]);
242		}
243		}
244		}
245
246	1	let reference_count = if lock.active_snapshots.is_empty() { 1 } else { 2 };
247	1	let snapshot = SnapshotData::new(snapshot_id, journal_id, entries, freed_pages, reference_count);
248	1	if let Err(index) = lock.search(snapshot_id) {
249	1	lock.active_snapshots.insert(index, snapshot);
250		}
251	1	Ok(snapshot_id)
252	1	}
253
254	1	pub fn fill_segments(&self, snapshot_id: SnapshotId, segments: &[SegmentSnapshop]) -> PRes<()> {
255	1	let mut segments_id = HashMap::new();
256	1	let mut segments_name = HashMap::new();
257	1	for segment in segments {
258	1	segments_id.insert(segment.id, segment.clone());
259	1	segments_name.insert(segment.name.clone(), segment.clone());
260		}
261	1	let mut lock = self.lock.lock()?;
262	1	if let Ok(index) = lock.search(snapshot_id) {
263	1	if let Some(snap) = lock.active_snapshots.get_mut(index) {
264	1	snap.segments = Some(segments_id);
265	1	snap.segments_name = Some(segments_name);
266		}
267		}
268	1	Ok(())
269	1	}
270
271	1	pub fn solve_segment_id(&self, snapshot_id: SnapshotId, segment: &str) -> PRes<Option<SegmentId>> {
272	1	let mut lock = self.lock.lock()?;
273	1	let res = if let Ok(index) = lock.search(snapshot_id) {
274	1	if let Some(snap) = lock.active_snapshots.get_mut(index) {
275	1	if let Some(segs) = &snap.segments_name {
276	1	if let Some(sd) = segs.get(segment) {
277	1	Some(sd.id)
278		} else {
279	1	None
280		}
281		} else {
282	0	None
283		}
284		} else {
285	0	None
286		}
287		} else {
288	0	None
289		};
290	1	Ok(res)
291	1	}
292
293	1	pub fn solve_segment_name(&self, snapshot_id: SnapshotId, segment_id: SegmentId) -> PRes<Option<String>> {
294	1	let mut lock = self.lock.lock()?;
295	1	let res = if let Ok(index) = lock.search(snapshot_id) {
296	1	if let Some(snap) = lock.active_snapshots.get_mut(index) {
297	1	if let Some(segs) = &snap.segments {
298	1	if let Some(sd) = segs.get(&segment_id) {
299	1	Some(sd.name.clone())
300		} else {
301	0	None
302		}
303		} else {
304	0	None
305		}
306		} else {
307	0	None
308		}
309		} else {
310	0	None
311		};
312	1	Ok(res)
313	1	}
314
315	1	pub fn scan(&self, snapshot_id: SnapshotId, segment_id: SegmentId) -> PRes<Option<SnapshotSegmentPageIterator>> {
316	1	let mut lock = self.lock.lock()?;
317	1	let res = if let Ok(index) = lock.search(snapshot_id) {
318	1	if let Some(snap) = lock.active_snapshots.get_mut(index) {
319	1	if let Some(segs) = &snap.segments {
320	1	if let Some(sd) = segs.get(&segment_id) {
321	1	Some(SnapshotSegmentPageIterator::new(sd.first_page))
322		} else {
323	0	None
324		}
325		} else {
326	0	None
327		}
328		} else {
329	0	None
330		}
331		} else {
332	0	None
333		};
334	1	Ok(res)
335	1	}
336
337	1	pub fn list(&self, snapshot_id: SnapshotId) -> PRes<Vec<(String, SegmentId)>> {
338	1	let mut lock = self.lock.lock()?;
339	1	let res = if let Ok(index) = lock.search(snapshot_id) {
340	1	if let Some(snap) = lock.active_snapshots.get_mut(index) {
341	1	if let Some(segs) = &snap.segments {
342	1	segs.values()
343	1	.map(\|data\| (data.name.clone(), data.id))
344		.collect::<Vec<_>>()
345		} else {
346	0	Vec::new()
347		}
348		} else {
349	0	Vec::new()
350		}
351		} else {
352	0	Vec::new()
353		};
354	1	Ok(res)
355	1	}
356
357	1	pub fn read(&self, snapshot_id: SnapshotId, id: &RecRef) -> PRes<Option<RecordVersion>> {
358	1	let lock = self.lock.lock()?;
359	1	let snapshot_sequence = lock.snapshot_sequence;
360	1	Ok(if let Some(v) = lock.mapping.get(id) {
361	1	let index = match v.binary_search_by(\|n\| search(n.snapshot_id, snapshot_id, snapshot_sequence)) {
362	1	Ok(index) => index,
363	1	Err(index) => index,
364		};
365	1	v.get(index).cloned()
366		} else {
367	1	None
368		})
369	1	}
370
371	1	fn clear_from(&self, snapshot_id: SnapshotId) -> PRes<(Vec<FreedPage>, Vec<JournalId>)> {
372	1	let mut lock = self.lock.lock()?;
373	1	let snapshot_sequence = lock.snapshot_sequence;
374	1	let mut free_pages = Vec::new();
375	1	let mut journal_ids = Vec::new();
376	1	if let Ok(index) = lock.search(snapshot_id) {
377	1	let size = index + 1;
378	1	let mut old = Vec::with_capacity(size);
379	1	old.extend_from_slice(&lock.active_snapshots[0..size]);
380	1	lock.active_snapshots = lock.active_snapshots.split_off(size);
381	1	for tx in old {
382	1	if let Some(id) = tx.journal_id.clone() {
383	1	journal_ids.push(id.clone());
384		}
385	1	if let Some(entries) = tx.entries {
386	1	for record in entries {
387	1	if let Some(v) = lock.mapping.get_mut(&record.id) {
388	1	if let Ok(index) =
389	1	v.binary_search_by(\|n\| search(n.snapshot_id, snapshot_id, snapshot_sequence))
390		{
391	1	v.truncate(index);
392		}
393		}
394	1	}
395		}
396	1	if let Some(pages) = tx.freed_pages {
397	1	free_pages.extend(pages);
398		}
399	1	}
400	1	}
401	1	Ok((free_pages, journal_ids))
402	1	}
403
404	1	fn release(&self, snapshot_id: SnapshotId) -> PRes<(Vec<FreedPage>, Vec<JournalId>)> {
405		//TODO: This work fine but can cause problems if double release is called for the same id,
406		//to refactor to something a bit more safe
407	1	let mut clear_id = None;
408		{
409	1	let mut lock = self.lock.lock()?;
410	1	if let Ok(index) = lock.search(snapshot_id) {
411	1	let mut loop_index = index;
412	1	while let Some(snap) = lock.active_snapshots.get_mut(loop_index) {
413	1	snap.reference_count -= 1;
414	1	if snap.reference_count > 0 {
415	1	break;
416		}
417	1	clear_id = Some(snap.snapshot_id);
418	1	loop_index += 1;
419		}
420		}
421	1	}
422	1	if let Some(c_id) = clear_id {
423	1	self.clear_from(c_id)
424		} else {
425	1	Ok((Vec::new(), Vec::new()))
426		}
427	1	}
428		}
429
430	1	pub fn release_snapshot(id: SnapshotId, snapshots: &Snapshots, allocator: &Allocator, journal: &Journal) -> PRes<()> {
431	1	let (to_free, to_clean) = snapshots.release(id)?;
432
433	1	for page in to_free {
434	1	allocator.free(page.page)?;
435	1	}
436	1	journal.finished_to_clean(&to_clean)?;
437	1	Ok(())
438	1	}
439
440		#[cfg(test)]
441		mod tests {
442		use super::{search, Change, EntryCase, RecordVersion, SegmentSnapshop, SnapshotEntry, Snapshots};
443		use crate::{
444		id::{RecRef, SegmentId},
445		journal::JournalId,
446		transaction::FreedPage,
447		};
448		use std::cmp::Ordering;
449
450		#[test]
451	1	fn test_search() {
452	1	assert_eq!(search(10, 20, 40), Ordering::Less);
453	1	assert_eq!(search(20, 10, 40), Ordering::Greater);
454	1	assert_eq!(search(10, 30, 20), Ordering::Greater);
455	1	assert_eq!(search(30, 10, 20), Ordering::Less);
456	1	assert_eq!(search(20, 20, 20), Ordering::Equal);
457	1	assert_eq!(search(20, 19, 20), Ordering::Greater);
458	1	assert_eq!(search(20, 21, 20), Ordering::Greater);
459	1	assert_eq!(search(21, 21, 20), Ordering::Equal);
460	1	assert_eq!(search(19, 19, 20), Ordering::Equal);
461	1	}
462
463		#[test]
464	1	fn add_and_read() {
465	1	let snap = Snapshots::new();
466	1	let mut records = Vec::new();
467	1	let mut freed_pages = Vec::new();
468	1	for x in 0..3 {
469	1	records.push(SnapshotEntry::change(&RecRef::new(10, x), x as u64, x as u16));
470	1	freed_pages.push(FreedPage::new(x as u64));
471		}
472	1	let tx = snap
473	1	.snapshot(records.clone(), freed_pages.clone(), JournalId::new(0, 0))
474		.unwrap();
475	1	snap.clear_from(tx).unwrap();
476	1	let tx = snap.snapshot(records, freed_pages, JournalId::new(0, 0)).unwrap();
477	1	assert_eq!(
478	1	snap.read(tx - 1, &RecRef::new(10, 2)).unwrap(),
479		Some(RecordVersion {
480		snapshot_id: 1,
481		case: EntryCase::Change(Change { pos: 2, version: 2 }),
482		})
483		);
484
485	1	assert_eq!(
486	1	snap.read(tx - 1, &RecRef::new(10, 1)).unwrap(),
487		Some(RecordVersion {
488		snapshot_id: 1,
489		case: EntryCase::Change(Change { pos: 1, version: 1 }),
490		})
491		);
492
493	1	assert_eq!(
494	1	snap.read(tx, &RecRef::new(10, 2)).unwrap(),
495		Some(RecordVersion {
496		snapshot_id: 1,
497		case: EntryCase::Change(Change { pos: 2, version: 2 }),
498		})
499		);
500	1	assert_eq!(snap.read(tx + 1, &RecRef::new(10, 2)).unwrap(), None);
501	1	assert_eq!(snap.read(tx + 1, &RecRef::new(10, 10)).unwrap(), None);
502	1	}
503
504		#[test]
505	1	fn add_and_read_mutliple_tx() {
506	1	let snap = Snapshots::new();
507	1	let mut txs = Vec::new();
508	1	for t in 0..5 {
509	1	let mut records = Vec::new();
510	1	let mut freed_pages = Vec::new();
511	1	for x in 0..10 {
512		// I skip a record for the specific tx because i need a missing record for the test
513	1	if x != t {
514	1	records.push(SnapshotEntry::change(
515	1	&RecRef::new(10, x),
516	1	(10 * t + x) as u64,
517	1	(10 * t + x) as u16,
518		));
519	1	freed_pages.push(FreedPage::new((10 * t + x) as u64));
520		}
521		}
522	1	txs.push(snap.snapshot(records, freed_pages, JournalId::new(0, 0)).unwrap());
523	1	}
524	1	assert_eq!(
525	1	snap.read(txs[2], &RecRef::new(10, 2)).unwrap(),
526	1	Some(RecordVersion {
527	1	snapshot_id: txs[3],
528	1	case: EntryCase::Change(Change { pos: 32, version: 32 }),
529		})
530		);
531
532	1	assert_eq!(
533	1	snap.read(txs[3], &RecRef::new(10, 3)).unwrap(),
534	1	Some(RecordVersion {
535	1	snapshot_id: txs[4],
536	1	case: EntryCase::Change(Change { pos: 43, version: 43 }),
537		})
538		);
539	1	}
540
541		#[test]
542	1	fn test_snapshot_reference_count() {
543	1	let snap = Snapshots::new();
544	1	let mut records = Vec::new();
545	1	let mut freed_pages = Vec::new();
546	1	for x in 0..3 {
547	1	records.push(SnapshotEntry::change(&RecRef::new(10, x), x as u64, x as u16));
548	1	freed_pages.push(FreedPage::new(x as u64));
549		}
550	1	let first = snap.read_snapshot().unwrap();
551	1	let tx = snap
552	1	.snapshot(records.clone(), freed_pages, JournalId::new(0, 0))
553		.unwrap();
554	1	let last = snap.read_snapshot().unwrap();
555	1	snap.release(tx).unwrap();
556	1	assert_eq!(
557	1	snap.read(first, &RecRef::new(10, 2)).unwrap(),
558		Some(RecordVersion {
559		snapshot_id: 1,
560		case: EntryCase::Change(Change { pos: 2, version: 2 }),
561		})
562		);
563
564	1	snap.release(first).unwrap();
565	1	assert_eq!(snap.read(last, &RecRef::new(10, 2)).unwrap(), None);
566	1	}
567
568		#[test]
569	1	fn test_snapshot_release_clear() {
570	1	let snap = Snapshots::new();
571	1	let mut snaps = Vec::new();
572	1	for i in 0..10 {
573	1	let id = snap.read_snapshot().unwrap();
574	1	let mut segments = Vec::new();
575	1	let seg_id = SegmentId::new(i);
576	1	segments.push(SegmentSnapshop::new("one", seg_id, i.into()));
577	1	segments.push(SegmentSnapshop::new("two", seg_id, i.into()));
578	1	snap.fill_segments(id, &segments).expect("fill works");
579	1	snaps.push(id);
580	1	}
581	1	snap.release(snaps[0]).expect("release works");
582	1	snap.release(snaps[1]).expect("release works");
583	1	snap.release(snaps[2]).expect("release works");
584	1	let solved = snap.solve_segment_id(snaps[3], "one").expect("solve id correctly");
585	1	assert_eq!(solved, Some(SegmentId::new(3)));
586	1	}
587
588		#[test]
589	1	pub fn test_segment_snapshot_fill() {
590	1	let snap = Snapshots::new();
591	1	let id = snap.read_snapshot().unwrap();
592	1	let mut segments = Vec::new();
593	1	let seg_id = SegmentId::new(10);
594	1	segments.push(SegmentSnapshop::new("one", seg_id, 10));
595	1	snap.fill_segments(id, &segments).expect("fill works");
596	1	let solved = snap.solve_segment_id(id, "one").expect("solve id correctly");
597	1	assert_eq!(solved, Some(seg_id));
598	1	let solved = snap.solve_segment_name(id, seg_id).expect("solve id correctly");
599	1	assert_eq!(solved, Some("one".to_string()));
600	1	let solved = snap.solve_segment_id(id, "two").expect("solve id correctly");
601	1	assert_eq!(solved, None);
602	1	let solved = snap.scan(id, seg_id).expect("solve id correctly");
603	1	assert!(solved.is_some());
604	1	snap.release(id).expect("release works");
605	1	}
606
607		#[test]
608	1	pub fn test_segment_snapshot_fill_multiple() {
609	1	let snap = Snapshots::new();
610	1	let mut snaps = Vec::new();
611	1	for i in 0..10 {
612	1	let id = snap.read_snapshot().unwrap();
613	1	let mut segments = Vec::new();
614	1	let seg_id = SegmentId::new(i);
615	1	segments.push(SegmentSnapshop::new("one", seg_id, i.into()));
616	1	segments.push(SegmentSnapshop::new("two", seg_id, i.into()));
617	1	snap.fill_segments(id, &segments).expect("fill works");
618	1	snaps.push(id);
619	1	}
620	1	let solved = snap.solve_segment_id(snaps[3], "one").expect("solve id correctly");
621	1	assert_eq!(solved, Some(SegmentId::new(3)));
622	1	let solved = snap.solve_segment_id(snaps[3], "two").expect("solve id correctly");
623	1	assert_eq!(solved, Some(SegmentId::new(3)));
624	1	let solved = snap.solve_segment_id(snaps[6], "one").expect("solve id correctly");
625	1	assert_eq!(solved, Some(SegmentId::new(6)));
626	1	let solved = snap.solve_segment_id(snaps[6], "two").expect("solve id correctly");
627	1	assert_eq!(solved, Some(SegmentId::new(6)));
628	1	}
629		}