persy 1.5.2

Transactional Persistence Engine
Documentation
#[cfg(unix)]
use crate::device::{is_free, PageOps};
use crate::{
    device::{Device, FreePage, Page, ReadPage, SizeTool, UpdateList},
    error::{OpenError, PERes},
};
use fs2::FileExt;
#[cfg(unix)]
use std::sync::Arc;
use std::{fs::File, sync::Mutex};

#[cfg(not(unix))]
use crate::device::common::{
    check_and_trim, create_page, create_page_raw, flush_free_page, flush_page, load_free_page, load_page,
    load_page_raw, mark_allocated, trim_or_free_page,
};

#[cfg(unix)]
pub struct FileDevice {
    file: File,
    size: Mutex<u64>,
}

#[cfg(unix)]
impl FileDevice {
    pub fn new_truncate(file: File) -> Result<FileDevice, OpenError> {
        file.try_lock_exclusive()?;
        file.set_len(0).map_err(OpenError::from)?;
        Ok(FileDevice {
            file,
            size: Mutex::new(0),
        })
    }
    pub fn new(file: File) -> Result<FileDevice, OpenError> {
        file.try_lock_exclusive()?;
        let len = file.len()?;
        Ok(FileDevice {
            file,
            size: Mutex::new(len),
        })
    }

    fn create_page_offset(&self, size: u64) -> PERes<u64> {
        let mut current = self.size.lock().expect("device file size lock not poisoned");
        let page = *current;
        *current += size;
        Ok(page)
    }

    fn trim_if_possible(&self, page: u64, size: u64) -> PERes<bool> {
        let mut current = self.size.lock().expect("device file size lock not poisoned");
        if page + size == *current {
            *current = page;
            self.file.set_len(page)?;
            Ok(true)
        } else {
            Ok(false)
        }
    }

    fn check_and_trim(&self, update_list: &mut dyn UpdateList, check_top_list: bool) -> PERes<bool> {
        use std::os::unix::prelude::FileExt;
        let mut lock = self.size.lock().expect("device file size lock not poisoned");
        let value = *lock;
        if value == 0 {
            return Ok(false);
        }
        let mut exp = [0u8];
        self.file.read_exact_at(&mut exp, value - 1)?;
        if exp[0] == 0 {
            return Ok(false);
        }
        let size = (1 << exp[0]) as u64;
        let mut header = [0u8; 18];
        let page = value - size;
        if page == 0 {
            return Ok(false);
        }
        self.file.read_exact_at(&mut header, page)?;
        if is_free(header[1]) {
            let fp = self.load_free_page(page)?;
            if fp.get_prev_free() == 0 {
                update_list.remove(exp[0], page, fp.get_next_free(), check_top_list)?;
            } else {
                let mut pfp = self.load_free_page(fp.get_prev_free())?;
                pfp.set_next_free(fp.get_next_free());
                self.flush_free_page(&pfp)?;
            }
            if fp.get_next_free() == 0 {
                update_list.remove_last(exp[0], page, fp.get_prev_free(), check_top_list)?;
            } else {
                let mut nfp = self.load_free_page(fp.get_next_free())?;
                nfp.set_prev_free(fp.get_prev_free());
                self.flush_free_page(&nfp)?;
            }
            *lock = page;
            self.file.set_len(page)?;
            Ok(true)
        } else {
            Ok(false)
        }
    }
}

#[cfg(unix)]
impl Device for FileDevice {
    fn load_free_page(&self, page: u64) -> PERes<FreePage> {
        use std::os::unix::prelude::FileExt;
        let mut data = [0u8; 32];
        self.file.read_exact_at(&mut data, page)?;
        Ok(FreePage::new(page, data))
    }

    fn flush_free_page(&self, page: &FreePage) -> PERes<()> {
        use std::os::unix::prelude::FileExt;
        self.file.write_all_at(&page.buff, page.get_index())?;
        Ok(())
    }

    fn load_page(&self, page: u64) -> PERes<ReadPage> {
        use std::os::unix::prelude::FileExt;
        let mut exp = [0u8];
        self.file.read_exact_at(&mut exp, page)?;
        let size = 1 << exp[0]; //EXP - (size_exp+size_mitigator);
        let mut ve = vec![0u8; size];
        ve[0] = exp[0];
        self.file.read_exact_at(&mut ve[1..size], page + 1)?;
        // add 2 to skip the metadata
        Ok(ReadPage::new(Arc::new(ve), 2, page, exp[0]))
    }

    fn load_page_if_exists(&self, page: u64) -> PERes<Option<ReadPage>> {
        {
            if *self.size.lock().expect("device file size lock not poisoned") <= page {
                return Ok(None);
            }
        }
        Ok(Some(self.load_page(page)?))
    }

    fn load_page_raw(&self, page: u64, size_exp: u8) -> PERes<Page> {
        use std::os::unix::prelude::FileExt;
        let size = 1 << size_exp; //EXP - (size_exp+size_mitigator);
        let mut ve = vec![0u8; size];
        self.file.read_exact_at(&mut ve, page)?;
        Ok(Page::new(ve, 0, page, size_exp))
    }

    fn flush_page(&self, page: &Page) -> PERes<()> {
        use std::os::unix::prelude::FileExt;
        self.file.write_all_at(&page.buff, page.get_index())?;
        Ok(())
    }

    fn create_page_raw(&self, exp: u8) -> PERes<u64> {
        use std::os::unix::prelude::FileExt;
        let size = (1 << exp) as usize; //EXP - (size_exp+size_mitigator);
        let offset = self.create_page_offset(size as u64)?;
        let ve = vec![0u8; size];
        self.file.write_all_at(&ve, offset)?; //exp
        Ok(offset)
    }

    fn create_page(&self, exp: u8) -> PERes<Page> {
        use std::os::unix::prelude::FileExt;
        let size = (1 << exp) as usize; //EXP - (size_exp+size_mitigator);
        let offset = self.create_page_offset(size as u64)?;
        let mut ve = vec![0u8; size];
        ve[0] = exp;
        ve[size - 1] = exp;
        self.file.write_all_at(&ve, offset)?; //exp
        Ok(Page::new(ve, 2, offset, exp))
    }

    fn mark_allocated(&self, page: u64) -> PERes<u64> {
        let mut fp = self.load_free_page(page)?;
        debug_assert!(fp.is_free()?, "allocating: {} already allocated ", page);
        fp.set_free(false)?;
        let prev_page = fp.get_prev_free();
        fp.set_next_free(0);
        fp.set_prev_free(0);
        self.flush_free_page(&fp)?;
        if prev_page != 0 {
            let mut pfp = self.load_free_page(prev_page)?;
            pfp.set_next_free(0);
            self.flush_free_page(&pfp)?;
        }
        Ok(prev_page)
    }

    fn sync(&self) -> PERes<()> {
        self.file.sync_data()?;
        Ok(())
    }

    fn trim_or_free_page(&self, page: u64, update_list: &mut dyn UpdateList) -> PERes<()> {
        let mut tfp = self.load_free_page(page)?;
        let size = (1 << tfp.get_size_exp()) as u64;
        debug_assert!(!tfp.is_free()?, "freeing: {} already freed ", page);
        if !self.trim_if_possible(page, size)? {
            let next = update_list.update(tfp.get_size_exp(), page)?;
            tfp.set_free(true)?;
            tfp.set_prev_free(0);
            tfp.set_next_free(next);
            self.flush_free_page(&tfp)?;
            if next != 0 {
                let mut nfp = self.load_free_page(next)?;
                nfp.set_prev_free(page);
                self.flush_free_page(&nfp)?;
            }
        } else {
            while self.check_and_trim(update_list, true)? {}
        }
        Ok(())
    }

    fn trim_end_pages(&self, update_list: &mut dyn UpdateList) -> PERes<()> {
        while self.check_and_trim(update_list, false)? {}
        Ok(())
    }

    #[cfg(test)]
    fn release_file_lock(&self) -> PERes<()> {
        self.file.unlock()?;
        Ok(())
    }
}

#[cfg(not(unix))]
pub struct FileHandler {
    file: File,
    metadata_changed: bool,
}

#[cfg(not(unix))]
pub struct FileDevice {
    file: Mutex<FileHandler>,
}

#[cfg(not(unix))]
impl FileDevice {
    pub fn new_truncate(file: File) -> Result<FileDevice, OpenError> {
        file.try_lock_exclusive()?;
        file.set_len(0).map_err(OpenError::from)?;
        Ok(FileDevice {
            file: Mutex::new(FileHandler {
                file,
                metadata_changed: false,
            }),
        })
    }
    pub fn new(file: File) -> Result<FileDevice, OpenError> {
        file.try_lock_exclusive()?;
        Ok(FileDevice {
            file: Mutex::new(FileHandler {
                file,
                metadata_changed: false,
            }),
        })
    }
}

#[cfg(not(unix))]
impl Device for FileDevice {
    fn load_free_page(&self, page: u64) -> PERes<FreePage> {
        load_free_page(&mut self.file.lock().expect("device file lock not poisoned").file, page)
    }

    fn flush_free_page(&self, page: &FreePage) -> PERes<()> {
        flush_free_page(&mut self.file.lock().expect("device file lock not poisoned").file, page)
    }

    fn load_page(&self, page: u64) -> PERes<ReadPage> {
        load_page(&mut self.file.lock().expect("device file lock not poisoned").file, page)
    }

    fn load_page_if_exists(&self, page: u64) -> PERes<Option<ReadPage>> {
        let mut lock = self.file.lock().expect("device file lock not poisoned");
        if lock.file.len()? <= page {
            return Ok(None);
        }
        Ok(Some(load_page(&mut lock.file, page)?))
    }

    fn load_page_raw(&self, page: u64, size_exp: u8) -> PERes<Page> {
        load_page_raw(
            &mut self.file.lock().expect("device file lock not poisoned").file,
            page,
            size_exp,
        )
    }

    fn flush_page(&self, page: &Page) -> PERes<()> {
        flush_page(&mut self.file.lock().expect("device file lock not poisoned").file, page)
    }

    fn create_page_raw(&self, exp: u8) -> PERes<u64> {
        let lock = &mut self.file.lock().expect("device file lock not poisoned");
        lock.metadata_changed = true;
        create_page_raw(&mut lock.file, exp, None)
    }

    fn create_page(&self, exp: u8) -> PERes<Page> {
        let lock = &mut self.file.lock().expect("device file lock not poisoned");
        lock.metadata_changed = true;
        create_page(&mut lock.file, exp, None)
    }

    fn mark_allocated(&self, page: u64) -> PERes<u64> {
        mark_allocated(&mut self.file.lock().expect("device file lock not poisoned").file, page)
    }

    fn sync(&self) -> PERes<()> {
        let to_sync;
        let metadata_changed;
        {
            let mut lock = self.file.lock().expect("device file lock not poisoned");
            to_sync = lock.file.try_clone()?;
            metadata_changed = lock.metadata_changed;
            lock.metadata_changed = false;
        }
        if metadata_changed {
            to_sync.sync_all()?;
        } else {
            to_sync.sync_data()?;
        }
        Ok(())
    }

    fn trim_or_free_page(&self, page: u64, update_list: &mut dyn UpdateList) -> PERes<()> {
        trim_or_free_page(
            &mut self.file.lock().expect("device file lock not poisoned").file,
            page,
            update_list,
        )
    }

    fn trim_end_pages(&self, update_list: &mut dyn UpdateList) -> PERes<()> {
        while check_and_trim(
            &mut self.file.lock().expect("device file lock not poisoned").file,
            update_list,
            false,
        )? {}
        Ok(())
    }

    #[cfg(test)]
    fn release_file_lock(&self) -> PERes<()> {
        Ok(self.file.lock().expect("device file lock not poisoned").file.unlock()?)
    }
}