1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157

#![cfg_attr(target_family = "wasm", feature(wasi_ext))]

mod file_ext;
mod page;
mod page_no;
mod size_info;

pub use page::*;
pub use page_no::*;

use file_ext::FileExt;
use size_info::SizeInfo;
use std::convert::TryInto;
use std::fs::File;
use std::io::{Error, ErrorKind, Result};
use std::sync::atomic::{AtomicU32, Ordering};

use page_lock::RwLock;
use tokio::task::spawn_blocking;

/// **NOTE**: Caller must ensure that closure must not panic.
/// If it does, it will be undefined behavior.
macro_rules! into_async { [$body: expr] => { unsafe { spawn_blocking(move || $body).await.unwrap_unchecked() } }; }

pub struct Pages<N, const SIZE: usize> {
    /// Total Page number
    len: AtomicU32,
    rwlock: RwLock<N>,
    file: &'static File,
    metadata: [u8; SIZE],
}

impl<N: PageNo, const SIZE: usize> Pages<N, SIZE> {
    /// Create a new `Pages` instance.
    pub fn open(file: File) -> Result<Self> {
        let size_info = SizeInfo {
            block_size: SIZE as u32,
            pages_len_nbytes: N::SIZE as u8,
        };
        let file_len = file.metadata()?.len();
        // So that, There is no residue bytes.
        if file_len % SIZE as u64 != 0 {
            return Err(ErrorKind::InvalidData.into());
        }
        let mut metadata = [0; SIZE];
        let len = if file_len == 0 {
            // Is new file? set the length to 1.
            file.set_len(SIZE as u64)?;
            1
        } else {
            file.read_exact_at(&mut metadata, 0)?;
            let info = SizeInfo::from(metadata[0..4].try_into().unwrap());
            if info != size_info {
                return Err(Error::new(
                    ErrorKind::InvalidInput,
                    format!("Expected {:?}, but got: {:?}", info, size_info),
                ));
            }
            metadata[0..4].copy_from_slice(&size_info.to_bytes());
            file_len as u32 / SIZE as u32
        };
        Ok(Self {
            len: AtomicU32::new(len),
            file: Box::leak(Box::new(file)),
            rwlock: RwLock::new(),
            metadata,
        })
    }

    pub async fn sync_data(&self) -> Result<()> {
        let file = self.file;
        into_async!(file.sync_data())
    }

    pub async fn sync_all(&self) -> Result<()> {
        let file = self.file;
        into_async!(file.sync_all())
    }

    /// Read a page, without locking.
    pub async fn get(&self, num: N) -> Result<[u8; SIZE]> {
        debug_assert!(num.as_u32() < self.len());
        let num = num.as_u32() as u64;
        let file = self.file;
        into_async!({
            let mut buf = [0; SIZE];
            file.read_exact_at(&mut buf, SIZE as u64 * num)?;
            Ok(buf)
        })
    }

    pub async fn read(&self, num: N) -> Result<Reader<'_, N, [u8; SIZE]>> {
        let _guard = self.rwlock.read(num).await;
        let data = self.get(num).await?;
        Ok(Reader { data, _guard })
    }

    pub async fn goto(&self, num: N) -> Result<Page<'_, N, SIZE>> {
        let _lock = self.rwlock.write(num).await;
        let buf = self.get(num).await?;
        Ok(Page {
            _lock,
            num,
            buf,
            pages: self,
        })
    }

    pub async fn write(&self, num: N, buf: [u8; SIZE]) -> Result<()> {
        debug_assert!(num.as_u32() < self.len());
        let num = num.as_u32() as u64;
        let file = self.file;
        into_async!(file.write_all_at(&buf, SIZE as u64 * num))
    }

    /// Increase page number by `count`
    pub async fn alloc(&self, count: u32) -> Result<N> {
        let old_len = self.len.fetch_add(count, Ordering::SeqCst);
        let file = self.file;
        into_async!(file.set_len(SIZE as u64 * (old_len + count) as u64))?;
        Ok(N::new(old_len))
    }

    pub async fn create(&self, buf: [u8; SIZE]) -> Result<N> {
        let num = self.len.fetch_add(1, Ordering::SeqCst);
        self.write(PageNo::new(num), buf).await?;
        Ok(N::new(num))
    }

    pub fn alloc_sync(&self, count: u32) -> Result<N> {
        let old_len = self.len.fetch_add(count, Ordering::SeqCst);
        self.file.set_len(SIZE as u64 * (old_len + count) as u64)?;
        Ok(N::new(old_len))
    }

    #[inline]
    #[allow(clippy::len_without_is_empty)]
    pub fn len(&self) -> u32 {
        self.len.load(Ordering::Relaxed)
    }

    #[inline]
    pub fn inner(&self) -> &File {
        self.file
    }

    /// Get a mutable reference to the pages's metadata.
    pub fn metadata(&self) -> *mut [u8] {
        &self.metadata[4..] as *const _ as *mut _
    }
}

impl<N, const SIZE: usize> Drop for Pages<N, SIZE> {
    fn drop(&mut self) {
        self.file.write_all_at(&self.metadata, 0).unwrap();
    }
}