use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
use std::error;
use std::fs::File;
pub use std::fs::OpenOptions;
use std::io;
use std::io::{Read, Write};
use std::rc::Rc;
use std::sync;
use std::sync::Arc;
use address::Address;
use allocator::Allocator;
use config::Config;
use discref::DiscRef;
use fs2::FileExt;
use index::config::{IndexType, Indexes, ValueMode};
use index::keeper::{IndexKeeper, ValueChange};
use index::tree::{Index, Value};
use journal::Journal;
use journal::JOURNAL_PAGE_EXP;
use record_scanner::{RecordScanner, RecordScannerTx};
use std::collections::HashMap;
use std::fmt;
use std::str;
use transaction::TxSegCheck::{CREATED, DROPPED, NONE};
use transaction::{Transaction, TxRead};
const DEFAULT_PAGE_EXP: u8 = 10;
#[derive(PartialEq, Eq, PartialOrd, Ord, Hash, Clone, Debug)]
pub struct RecRef {
pub page: u64,
pub pos: u32,
}
pub struct PersyImpl {
config: Arc<Config>,
journal: Journal,
address: Address,
indexes: Indexes,
allocator: Arc<Allocator>,
}
pub struct TxFinalize {
transaction: Transaction,
}
#[derive(PartialEq, Debug)]
pub enum RecoverStatus {
Started,
PrepareCommit,
Rollback,
Commit,
}
#[derive(PartialEq, Debug)]
pub enum PersyError {
IO(String),
Err(String),
DecodingUTF(str::Utf8Error),
DecodingBASE64(data_encoding::DecodeError),
VersionNotLastest,
RecordNotFound(RecRef),
SegmentNotFound,
SegmentAlreadyExists,
CannotDropSegmentCreatedInTx,
Lock,
IndexMinElementsShouldBeAtLeastDoubleOfMax,
IndexNotFound,
IndexTypeMismatch(String),
IndexDuplicateKey(String, String),
}
pub type PRes<T> = Result<T, PersyError>;
impl PersyImpl {
pub fn create<P: Into<String>>(path: P) -> PRes<()> {
let f: File = OpenOptions::new()
.write(true)
.read(true)
.create_new(true)
.open(path.into())?;
PersyImpl::create_from_file(f)
}
pub fn create_from_file(f: File) -> PRes<()> {
f.try_lock_exclusive()?;
PersyImpl::init_file(f)?;
Ok(())
}
fn init_file(fl: File) -> PRes<()> {
let disc = DiscRef::new(fl);
let root_page = disc.create_page_raw(DEFAULT_PAGE_EXP)?;
let allocator_page = Allocator::init(&disc)?;
let allocator = &Allocator::new(disc, &Rc::new(Config::new()), allocator_page)?;
let address_page = Address::init(allocator)?;
let journal_page = Journal::init(allocator)?;
{
let mut root = allocator.disc().load_page_raw(root_page, DEFAULT_PAGE_EXP)?;
root.write_u16::<BigEndian>(0)?;
root.write_u64::<BigEndian>(address_page)?;
root.write_u64::<BigEndian>(journal_page)?;
root.write_u64::<BigEndian>(allocator_page)?;
allocator.flush_page(&mut root)?;
}
allocator.disc().sync()?;
Ok(())
}
fn new(file: File, config: Config) -> PRes<PersyImpl> {
let disc = DiscRef::new(file);
let address_page;
let journal_page;
let allocator_page;
{
let mut pg = disc.load_page_raw(0, DEFAULT_PAGE_EXP)?;
pg.read_u16::<BigEndian>()?;
address_page = pg.read_u64::<BigEndian>()?;
journal_page = pg.read_u64::<BigEndian>()?;
allocator_page = pg.read_u64::<BigEndian>()?;
}
let config = Arc::new(config);
let allocator = Arc::new(Allocator::new(disc, &config, allocator_page)?);
let address = Address::new(&allocator, &config, address_page)?;
let journal = Journal::new(&allocator, journal_page)?;
let indexes = Indexes::new(&config);
Ok(PersyImpl {
config: config.clone(),
journal,
address,
indexes,
allocator,
})
}
fn recover<C>(&self, check_if_commit: C) -> PRes<()>
where
C: Fn(&Vec<u8>) -> bool,
{
let mut last_id = None;
let mut commit_order = Vec::new();
let mut transactions = HashMap::new();
let journal = &self.journal;
let jp = journal.recover(|record, id| {
let tx = transactions
.entry(id.clone())
.or_insert_with(|| (RecoverStatus::Started, Transaction::recover(id.clone())));
let res = record.recover(&mut tx.1);
if res.is_err() {
tx.0 = RecoverStatus::Rollback;
} else {
match res.unwrap() {
RecoverStatus::Started => {
if tx.0 != RecoverStatus::Rollback {
tx.0 = RecoverStatus::Started;
}
}
RecoverStatus::PrepareCommit => {
if tx.0 != RecoverStatus::Rollback {
tx.0 = RecoverStatus::PrepareCommit;
commit_order.push(id.clone());
}
}
RecoverStatus::Rollback => {
tx.0 = RecoverStatus::Rollback;
}
RecoverStatus::Commit => {
if tx.0 != RecoverStatus::Rollback {
tx.0 = RecoverStatus::Commit;
}
}
}
}
})?;
let allocator = &self.allocator;
let address = &self.address;
let indexes = &self.indexes;
for id in commit_order {
if let Some(mut rec) = transactions.remove(&id) {
if rec.0 == RecoverStatus::PrepareCommit {
if check_if_commit(rec.1.meta_id()) {
rec.1.recover_prepare_commit(journal, address, allocator)?;
rec.1.recover_commit(journal, address, indexes, allocator)?;
last_id = Some(id.clone());
} else {
rec.1.recover_rollback(journal, address, allocator)?;
}
}
}
}
for p in jp {
allocator.remove_from_free(p, JOURNAL_PAGE_EXP)?;
}
for (_, rec) in transactions.iter_mut() {
rec.1.recover_rollback(journal, address, allocator)?;
}
if let Some(id) = last_id {
self.journal.clear(&id)?;
}
allocator.flush_free_list()?;
Ok(())
}
pub fn open<P: Into<String>>(path: P, config: Config) -> PRes<PersyImpl> {
PersyImpl::open_with_recover(path, config, |_| true)
}
pub fn open_from_file(f: File, config: Config) -> PRes<PersyImpl> {
PersyImpl::open_from_file_with_recover(f, config, |_| true)
}
pub fn open_with_recover<C, P: Into<String>>(path: P, config: Config, recover: C) -> PRes<PersyImpl>
where
C: Fn(&Vec<u8>) -> bool,
{
let f = OpenOptions::new()
.write(true)
.read(true)
.create(false)
.truncate(false)
.open(path.into())?;
PersyImpl::open_from_file_with_recover(f, config, recover)
}
pub fn open_from_file_with_recover<C>(f: File, config: Config, recover: C) -> PRes<PersyImpl>
where
C: Fn(&Vec<u8>) -> bool,
{
f.try_lock_exclusive()?;
let persy = PersyImpl::new(f, config)?;
persy.recover(recover)?;
Ok(persy)
}
pub fn begin_id(&self, meta_id: Vec<u8>) -> PRes<Transaction> {
let journal = &self.journal;
Ok(Transaction::new(journal, self.config.tx_strategy(), meta_id)?)
}
pub fn begin(&self) -> PRes<Transaction> {
self.begin_id(Vec::new())
}
pub fn create_segment(&self, tx: &mut Transaction, segment: &str) -> PRes<()> {
match tx.exists_segment(segment) {
DROPPED => {}
CREATED(_) => {
return Err(PersyError::SegmentAlreadyExists);
}
NONE => {
if self.address.exists_segment(&segment)? {
return Err(PersyError::SegmentAlreadyExists);
}
}
}
let segment_id = self.address.create_temp_segment(segment)?;
tx.add_create_segment(&self.journal, segment, segment_id)?;
Ok(())
}
pub fn drop_segment(&self, tx: &mut Transaction, segment: &str) -> PRes<()> {
let (_, segment_id) = self.check_segment_tx(tx, segment)?;
tx.add_drop_segment(&self.journal, segment, segment_id)?;
Ok(())
}
pub fn exists_segment(&self, segment: &str) -> PRes<bool> {
self.address.exists_segment(segment)
}
pub fn exists_segment_tx(&self, tx: &Transaction, segment: &str) -> PRes<bool> {
match tx.exists_segment(segment) {
DROPPED => Ok(false),
CREATED(_) => Ok(true),
NONE => self.address.exists_segment(segment),
}
}
fn check_segment_tx(&self, tx: &Transaction, segment: &str) -> PRes<(bool, u32)> {
match tx.exists_segment(segment) {
DROPPED => Err(PersyError::SegmentNotFound),
CREATED(segment_id) => Ok((true, segment_id)),
NONE => {
if let Some(id) = self.address.segment_id(segment)? {
Ok((false, id))
} else {
Err(PersyError::SegmentNotFound)
}
}
}
}
pub fn insert_record(&self, tx: &mut Transaction, segment: &str, rec: &[u8]) -> PRes<RecRef> {
let (in_tx, segment_id) = self.check_segment_tx(tx, segment)?;
let len = rec.len();
let allocation_exp = exp_from_content_size(len as u64);
let allocator = &self.allocator;
let address = &self.address;
let journal = &self.journal;
let page = allocator.allocate(allocation_exp)?;
let rec_ref = if in_tx {
address.allocate_temp(segment_id)?
} else {
address.allocate(segment_id)?
};
tx.add_insert(journal, segment_id, &rec_ref, page)?;
{
let mut pg = allocator.write_page(page)?;
pg.write_u64::<BigEndian>(len as u64)?;
pg.write_all(rec)?;
allocator.flush_page(&mut pg)?;
}
Ok(rec_ref)
}
fn read_ref_segment(&self, tx: &Transaction, segment_id: u32, rec_ref: &RecRef) -> PRes<Option<(u64, u16, u32)>> {
match tx.read(rec_ref) {
TxRead::RECORD(rec) => Ok(Some((rec.0, rec.1, segment_id))),
TxRead::DELETED => Ok(None),
TxRead::NONE => Ok(self
.address
.read(rec_ref, segment_id)?
.map(|(pos, version)| (pos, version, segment_id))),
}
}
fn read_ref(&self, tx: &Transaction, segment: &str, rec_ref: &RecRef) -> PRes<Option<(u64, u16, u32)>> {
let (_, segment_id) = self.check_segment_tx(tx, segment)?;
self.read_ref_segment(tx, segment_id, rec_ref)
}
fn read_page(&self, page: u64) -> PRes<Vec<u8>> {
let mut pg = self.allocator.load_page(page)?;
let len = pg.read_u64::<BigEndian>()?;
let mut buffer = Vec::<u8>::with_capacity(len as usize);
pg.take(len).read_to_end(&mut buffer)?;
Ok(buffer)
}
pub fn read_record_scan_tx(&self, tx: &Transaction, segment_id: u32, rec_ref: &RecRef) -> PRes<Option<Vec<u8>>> {
if let Some(page) = self.read_ref_segment(tx, segment_id, rec_ref)? {
Ok(Some(self.read_page(page.0)?))
} else {
Ok(None)
}
}
pub fn read_record_tx(&self, tx: &mut Transaction, segment: &str, rec_ref: &RecRef) -> PRes<Option<Vec<u8>>> {
if let Some(page) = self.read_ref(tx, &segment, rec_ref)? {
tx.add_read(&self.journal, page.2, rec_ref, page.1)?;
return Ok(Some(self.read_page(page.0)?));
}
Ok(None)
}
pub fn read_record(&self, segment: &str, rec_ref: &RecRef) -> PRes<Option<Vec<u8>>> {
if let Some(segment_id) = self.address.segment_id(segment)? {
self.read_record_scan(segment_id, rec_ref)
} else {
Err(PersyError::SegmentNotFound)
}
}
pub fn read_record_scan(&self, segment_id: u32, rec_ref: &RecRef) -> PRes<Option<Vec<u8>>> {
if let Some((page, _)) = self.address.read(rec_ref, segment_id)? {
Ok(Some(self.read_page(page)?))
} else {
Ok(None)
}
}
pub fn scan_records(&self, segment: &str) -> PRes<RecordScanner> {
let segment_id;
if let Some(id) = self.address.segment_id(segment)? {
segment_id = id;
} else {
return Err(PersyError::SegmentNotFound);
}
Ok(RecordScanner::new(&self, segment_id, self.address.scan(segment_id)?))
}
pub fn scan_records_tx<'a>(&'a self, tx: &'a Transaction, segment: &str) -> PRes<RecordScannerTx<'a>> {
let (_, segment_id) = self.check_segment_tx(tx, segment)?;
Ok(RecordScannerTx::<'a>::new(
&self,
&tx,
segment_id,
self.address.scan(segment_id)?,
))
}
pub fn update_record(&self, tx: &mut Transaction, segment: &str, rec_ref: &RecRef, rec: &[u8]) -> PRes<()> {
let allocator = &self.allocator;
let journal = &self.journal;
if let Some(old) = self.read_ref(tx, segment, rec_ref)? {
let len = rec.len();
let allocation_exp = exp_from_content_size(len as u64);
let page = allocator.allocate(allocation_exp)?;
tx.add_update(journal, old.2, &rec_ref, page, old.0, old.1)?;
{
let mut pg = allocator.write_page(page)?;
pg.write_u64::<BigEndian>(len as u64)?;
pg.write_all(rec)?;
allocator.flush_page(&mut pg)?;
}
return Ok(());
}
Err(PersyError::RecordNotFound(rec_ref.clone()))
}
pub fn delete_record(&self, tx: &mut Transaction, segment: &str, rec_ref: &RecRef) -> PRes<()> {
let journal = &self.journal;
if let Some(old) = self.read_ref(tx, segment, rec_ref)? {
tx.add_delete(journal, old.2, &rec_ref, old.0, old.1)?;
return Ok(());
}
Err(PersyError::RecordNotFound(rec_ref.clone()))
}
pub fn rollback(&self, mut tx: Transaction) -> PRes<()> {
let allocator = &self.allocator;
let journal = &self.journal;
let address = &self.address;
tx.rollback(journal, address, allocator)
}
pub fn prepare_commit(&self, mut tx: Transaction) -> PRes<TxFinalize> {
let indexes = &self.indexes;
let allocator = &self.allocator;
let journal = &self.journal;
let address = &self.address;
tx = tx.prepare_commit(journal, address, indexes, self, allocator)?;
Ok(TxFinalize { transaction: tx })
}
pub fn rollback_prepared(&self, finalizer: &mut TxFinalize) -> PRes<()> {
let allocator = &self.allocator;
let journal = &self.journal;
let address = &self.address;
let indexes = &self.indexes;
finalizer
.transaction
.rollback_prepared(journal, address, indexes, allocator)
}
pub fn commit(&self, finalizer: &mut TxFinalize) -> PRes<()> {
let allocator = &self.allocator;
let journal = &self.journal;
let indexes = &self.indexes;
let address = &self.address;
finalizer.transaction.commit(address, journal, indexes, allocator)
}
pub fn create_index<K, V>(&self, tx: &mut Transaction, index_name: &str, value_mode: ValueMode) -> PRes<()>
where
K: IndexType,
V: IndexType,
{
Indexes::create_index::<K, V>(self, tx, &index_name.to_string(), 32, 128, value_mode)
}
pub fn drop_index(&self, tx: &mut Transaction, index_name: &str) -> PRes<()> {
Indexes::drop_index(self, tx, &index_name.to_string())
}
pub fn put<K, V>(&self, tx: &mut Transaction, index_name: &str, k: K, v: V) -> PRes<()>
where
K: IndexType,
V: IndexType,
{
Indexes::check_and_get_index::<K, V>(self, Some(tx), index_name)?;
tx.add_put(&index_name.to_string(), k, v);
Ok(())
}
pub fn remove<K, V>(&self, tx: &mut Transaction, index_name: &str, k: K, v: Option<V>) -> PRes<()>
where
K: IndexType,
V: IndexType,
{
Indexes::check_and_get_index::<K, V>(self, Some(tx), index_name)?;
tx.add_remove(&index_name.to_string(), k, v);
Ok(())
}
pub fn get_tx<K, V>(&self, tx: &mut Transaction, index_name: &str, k: &K) -> PRes<Option<Value<V>>>
where
K: IndexType,
V: IndexType,
{
let changes = tx.get_changes::<K, V>(&index_name.to_string(), k);
let mut ik = Indexes::check_and_get_index_keeper::<K, V>(self, Some(tx), index_name)?;
self.indexes.read_lock(index_name.to_string())?;
let mut result = ik.get(k)?;
self.indexes.read_unlock(index_name.to_string())?;
if let Some(key_changes) = changes {
for change in key_changes {
match change {
ValueChange::ADD(add_value) => {
result = if let Some(s_result) = result {
match s_result {
Value::SINGLE(v) => match IndexKeeper::<K, V>::value_mode(&ik) {
ValueMode::REPLACE => Some(Value::SINGLE(add_value)),
ValueMode::EXCLUSIVE => {
if v == add_value {
Some(Value::SINGLE(v))
} else {
return Err(PersyError::IndexDuplicateKey(
index_name.to_string(),
format!("{}", k),
));
}
}
ValueMode::CLUSTER => {
if v == add_value {
Some(Value::SINGLE(v))
} else {
Some(Value::CLUSTER(vec![v, add_value]))
}
}
},
Value::CLUSTER(mut values) => {
if let Ok(pos) = values.binary_search(&add_value) {
values.insert(pos, add_value);
}
Some(Value::CLUSTER(values))
}
}
} else {
Some(Value::SINGLE(add_value))
}
}
ValueChange::REMOVE(rv) => {
if let Some(remove_value) = rv {
result = if let Some(s_result) = result {
match s_result {
Value::SINGLE(v) => {
if v == remove_value {
None
} else {
Some(Value::SINGLE(v))
}
}
Value::CLUSTER(mut values) => {
if let Ok(pos) = values.binary_search(&remove_value) {
values.remove(pos);
}
if values.len() == 1 {
Some(Value::SINGLE(values.pop().unwrap()))
} else {
Some(Value::CLUSTER(values))
}
}
}
} else {
None
}
} else {
result = None;
}
}
}
}
Ok(result)
} else {
Ok(result)
}
}
pub fn get<K, V>(&self, index_name: &str, k: &K) -> PRes<Option<Value<V>>>
where
K: IndexType,
V: IndexType,
{
let mut ik = Indexes::check_and_get_index_keeper::<K, V>(self, None, index_name)?;
self.indexes.read_lock(index_name.to_string())?;
let r = ik.get(k);
self.indexes.read_unlock(index_name.to_string())?;
r
}
}
fn exp_from_content_size(size: u64) -> u8 {
let final_size = size + 8 + 2;
let mut res: u8 = 1;
loop {
if final_size < (1 << res) {
return res;
}
res += 1;
}
}
impl From<io::Error> for PersyError {
fn from(erro: io::Error) -> PersyError {
PersyError::IO(format!("{}", erro).to_string())
}
}
impl<T> From<sync::PoisonError<T>> for PersyError {
fn from(_: sync::PoisonError<T>) -> PersyError {
PersyError::Lock
}
}
impl From<str::Utf8Error> for PersyError {
fn from(err: str::Utf8Error) -> PersyError {
PersyError::DecodingUTF(err)
}
}
impl From<data_encoding::DecodeError> for PersyError {
fn from(err: data_encoding::DecodeError) -> PersyError {
PersyError::DecodingBASE64(err)
}
}
impl error::Error for PersyError {}
impl fmt::Display for PersyError {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
PersyError::IO(m) => write!(f, "IO Error: {}", m),
PersyError::Err(g) => write!(f, "Generic Error: {}", g),
PersyError::DecodingUTF(e) => write!(f, "String decoding error: {}", e),
PersyError::DecodingBASE64(e) => write!(f, "BASE64 decoding error: {}", e),
PersyError::VersionNotLastest => write!(f, "The record version is not latest"),
PersyError::RecordNotFound(r) => write!(f, "Record not found: {}", r),
PersyError::SegmentNotFound => write!(f, "Segment not found"),
PersyError::SegmentAlreadyExists => write!(f, "Segment already exist"),
PersyError::CannotDropSegmentCreatedInTx => {
write!(f, "Create and drop of a segment in the same transaction is not allowed")
}
PersyError::Lock => write!(f, "Failure acquiring lock for poisoning"),
PersyError::IndexMinElementsShouldBeAtLeastDoubleOfMax => write!(
f,
"Index min page elements should be maximum half of the maximum elements"
),
PersyError::IndexNotFound => write!(f, "Index not found"),
PersyError::IndexTypeMismatch(m) => write!(f, "Index method type mismatch persistent types: {}", m),
PersyError::IndexDuplicateKey(i, k) => write!(f, "Found duplicate key:{} for index: {}", k, i),
}
}
}
impl RecRef {
pub fn new(page: u64, pos: u32) -> RecRef {
RecRef { page, pos }
}
}
impl fmt::Display for RecRef {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
use byteorder::{BigEndian, ByteOrder};
use data_encoding::BASE64URL;
let mut bytes = [0; 12];
BigEndian::write_u64(&mut bytes, self.page);
BigEndian::write_u32(&mut bytes[8..], self.pos);
write!(f, "{}", BASE64URL.encode(&bytes))
}
}
impl std::str::FromStr for RecRef {
type Err = PersyError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
use byteorder::{BigEndian, ReadBytesExt};
use data_encoding::BASE64URL;
use std::io::Cursor;
let mut bytes = Cursor::new(BASE64URL.decode(s.as_bytes())?);
let page = bytes.read_u64::<BigEndian>()?;
let pos = bytes.read_u32::<BigEndian>()?;
Ok(RecRef::new(page, pos))
}
}
#[cfg(test)]
mod tests {
use super::RecRef;
#[test()]
fn test_persy_id_string() {
let id = RecRef::new(20, 30);
let s = format!("{}", id);
let rid = s.parse::<RecRef>();
assert_eq!(rid, Ok(id));
}
#[test()]
fn test_persy_id_parse_failure() {
let s = "ACCC";
let rid = s.parse::<RecRef>();
assert!(rid.is_err());
}
}