use libc::{
c_uint,
c_void,
size_t,
}; use std::marker::PhantomData; use std::{
fmt,
mem,
ptr,
result,
slice,
};
use ffi;
use cursor::{
RoCursor,
RwCursor,
}; use database::Database; use environment::{
Environment,
Stat,
}; use error::{
lmdb_result,
Error,
Result,
}; use flags::{
DatabaseFlags,
EnvironmentFlags,
WriteFlags,
};
/// An LMDB transaction. /// /// All database operations require a transaction. pubtrait Transaction: Sized { /// Returns a raw pointer to the underlying LMDB transaction. /// /// The caller **must** ensure that the pointer is not used after the /// lifetime of the transaction. fn txn(&self) -> *mut ffi::MDB_txn;
/// Commits the transaction. /// /// Any pending operations will be saved. fn commit(self) -> Result<()> { unsafe { let result = lmdb_result(ffi::mdb_txn_commit(self.txn()));
mem::forget(self);
result
}
}
/// Aborts the transaction. /// /// Any pending operations will not be saved. fn abort(self) { // Abort should be performed in transaction destructors.
}
/// Opens a database in the transaction. /// /// If `name` is `None`, then the default database will be opened, otherwise /// a named database will be opened. The database handle will be private to /// the transaction until the transaction is successfully committed. If the /// transaction is aborted the returned database handle should no longer be /// used. /// /// Prefer using `Environment::open_db`. /// /// ## Safety /// /// This function (as well as `Environment::open_db`, /// `Environment::create_db`, and `Database::create`) **must not** be called /// from multiple concurrent transactions in the same environment. A /// transaction which uses this function must finish (either commit or /// abort) before any other transaction may use this function. unsafefn open_db(&self, name: Option<&str>) -> Result<Database> {
Database::new(self.txn(), name, 0)
}
/// Gets an item from a database. /// /// This function retrieves the data associated with the given key in the /// database. If the database supports duplicate keys /// (`DatabaseFlags::DUP_SORT`) then the first data item for the key will be /// returned. Retrieval of other items requires the use of /// `Transaction::cursor_get`. If the item is not in the database, then /// `Error::NotFound` will be returned. fn get<'txn, K>(&'txn self, database: Database, key: &K) -> Result<&'txn [u8]> where
K: AsRef<[u8]>,
{ let key = key.as_ref(); letmut key_val: ffi::MDB_val = ffi::MDB_val {
mv_size: key.len() as size_t,
mv_data: key.as_ptr() as *mut c_void,
}; letmut data_val: ffi::MDB_val = ffi::MDB_val {
mv_size: 0,
mv_data: ptr::null_mut(),
}; unsafe { match ffi::mdb_get(self.txn(), database.dbi(), &mut key_val, &an style='color:red'>mut data_val) {
ffi::MDB_SUCCESS => Ok(slice::from_raw_parts(data_val.mv_data as *const u8, data_val.mv_size as usize)),
err_code => Err(Error::from_err_code(err_code)),
}
}
}
/// Open a new read-only cursor on the given database. fn open_ro_cursor<'txn>(&'txn self, db: Database) -> Result<RoCursor<'txn>> {
RoCursor::new(self, db)
}
/// Gets the option flags for the given database in the transaction. fn db_flags(&self, db: Database) -> Result<DatabaseFlags> { letmut flags: c_uint = 0; unsafe {
lmdb_result(ffi::mdb_dbi_flags(self.txn(), db.dbi(), &mut flags))?;
}
Ok(DatabaseFlags::from_bits_truncate(flags))
}
impl<'env> Drop for RoTransaction<'env> { fn drop(&mutself) { unsafe { ffi::mdb_txn_abort(self.txn) }
}
}
impl<'env> RoTransaction<'env> { /// Creates a new read-only transaction in the given environment. Prefer /// using `Environment::begin_ro_txn`. pub(crate) fn new(env: &'env Environment) -> Result<RoTransaction<'env>> { letmut txn: *mut ffi::MDB_txn = ptr::null_mut(); unsafe {
lmdb_result(ffi::mdb_txn_begin(env.env(), ptr::null_mut(), ffi::MDB_RDONLY, &yle='color:red'>mut txn))?;
Ok(RoTransaction {
txn,
_marker: PhantomData,
})
}
}
/// Resets the read-only transaction. /// /// Abort the transaction like `Transaction::abort`, but keep the /// transaction handle. `InactiveTransaction::renew` may reuse the handle. /// This saves allocation overhead if the process will start a new read-only /// transaction soon, and also locking overhead if /// `EnvironmentFlags::NO_TLS` is in use. The reader table lock is released, /// but the table slot stays tied to its thread or transaction. Reader locks /// generally don't interfere with writers, but they keep old versions of /// database pages allocated. Thus they prevent the old pages from being /// reused when writers commit new data, and so under heavy load the /// database size may grow much more rapidly than otherwise. pubfn reset(self) -> InactiveTransaction<'env> { let txn = self.txn; unsafe {
mem::forget(self);
ffi::mdb_txn_reset(txn)
};
InactiveTransaction {
txn,
_marker: PhantomData,
}
}
}
impl<'env> Drop for InactiveTransaction<'env> { fn drop(&mutself) { unsafe { ffi::mdb_txn_abort(self.txn) }
}
}
impl<'env> InactiveTransaction<'env> { /// Renews the inactive transaction, returning an active read-only /// transaction. /// /// This acquires a new reader lock for a transaction handle that had been /// released by `RoTransaction::reset`. pubfn renew(self) -> Result<RoTransaction<'env>> { let txn = self.txn; unsafe {
mem::forget(self);
lmdb_result(ffi::mdb_txn_renew(txn))?
};
Ok(RoTransaction {
txn,
_marker: PhantomData,
})
}
}
impl<'env> Drop for RwTransaction<'env> { fn drop(&mutself) { unsafe { ffi::mdb_txn_abort(self.txn) }
}
}
impl<'env> RwTransaction<'env> { /// Creates a new read-write transaction in the given environment. Prefer /// using `Environment::begin_ro_txn`. pub(crate) fn new(env: &'env Environment) -> Result<RwTransaction<'env>> { letmut txn: *mut ffi::MDB_txn = ptr::null_mut(); unsafe {
lmdb_result(ffi::mdb_txn_begin(env.env(), ptr::null_mut(), EnvironmentFlags::empty().bits(), &mut txn))?;
Ok(RwTransaction {
txn,
_marker: PhantomData,
})
}
}
/// Opens a database in the provided transaction, creating it if necessary. /// /// If `name` is `None`, then the default database will be opened, otherwise /// a named database will be opened. The database handle will be private to /// the transaction until the transaction is successfully committed. If the /// transaction is aborted the returned database handle should no longer be /// used. /// /// Prefer using `Environment::create_db`. /// /// ## Safety /// /// This function (as well as `Environment::open_db`, /// `Environment::create_db`, and `Database::open`) **must not** be called /// from multiple concurrent transactions in the same environment. A /// transaction which uses this function must finish (either commit or /// abort) before any other transaction may use this function. pubunsafefn create_db(&self, name: Option<&str>, flags: DatabaseFlags) -> Result<Database> {
Database::new(self.txn(), name, flags.bits() | ffi::MDB_CREATE)
}
/// Opens a new read-write cursor on the given database and transaction. pubfn open_rw_cursor<'txn>(&'txn mutself, db: Database) -> Result<RwCursor<'txn>> {
RwCursor::new(self, db)
}
/// Stores an item into a database. /// /// This function stores key/data pairs in the database. The default /// behavior is to enter the new key/data pair, replacing any previously /// existing key if duplicates are disallowed, or adding a duplicate data /// item if duplicates are allowed (`DatabaseFlags::DUP_SORT`). pubfn put<K, D>(&mutself, database: Database, key: &K, data: &D, flags: WriteFlags) -> Result<()> where
K: AsRef<[u8]>,
D: AsRef<[u8]>,
{ let key = key.as_ref(); let data = data.as_ref(); letmut key_val: ffi::MDB_val = ffi::MDB_val {
mv_size: key.len() as size_t,
mv_data: key.as_ptr() as *mut c_void,
}; letmut data_val: ffi::MDB_val = ffi::MDB_val {
mv_size: data.len() as size_t,
mv_data: data.as_ptr() as *mut c_void,
}; unsafe { lmdb_result(ffi::mdb_put(self.txn(), database.dbi(), &>mut key_val, &mut data_val, flags.bits())) }
}
/// Returns a buffer which can be used to write a value into the item at the /// given key and with the given length. The buffer must be completely /// filled by the caller. pubfn reserve<'txn, K>(
&'txn mut self,
database: Database,
key: &K,
len: size_t,
flags: WriteFlags,
) -> Result<&'txn mut [u8]> where
K: AsRef<[u8]>,
{ let key = key.as_ref(); letmut key_val: ffi::MDB_val = ffi::MDB_val {
mv_size: key.len() as size_t,
mv_data: key.as_ptr() as *mut c_void,
}; letmut data_val: ffi::MDB_val = ffi::MDB_val {
mv_size: len,
mv_data: ptr::null_mut::<c_void>(),
}; unsafe {
lmdb_result(ffi::mdb_put( self.txn(),
database.dbi(),
&mut key_val,
&mut data_val,
flags.bits() | ffi::MDB_RESERVE,
))?;
Ok(slice::from_raw_parts_mut(data_val.mv_data as *mut u8, data_val.mv_size as usize))
}
}
/// Deletes an item from a database. /// /// This function removes key/data pairs from the database. If the database /// does not support sorted duplicate data items (`DatabaseFlags::DUP_SORT`) /// the data parameter is ignored. If the database supports sorted /// duplicates and the data parameter is `None`, all of the duplicate data /// items for the key will be deleted. Otherwise, if the data parameter is /// `Some` only the matching data item will be deleted. This function will /// return `Error::NotFound` if the specified key/data pair is not in the /// database. pubfn del<K>(&mutself, database: Database, key: &K, data: Option<&[u8]>) -> Result<()> where
K: AsRef<[u8]>,
{ let key = key.as_ref(); letmut key_val: ffi::MDB_val = ffi::MDB_val {
mv_size: key.len() as size_t,
mv_data: key.as_ptr() as *mut c_void,
}; let data_val: Option<ffi::MDB_val> = data.map(|data| ffi::MDB_val {
mv_size: data.len() as size_t,
mv_data: data.as_ptr() as *mut c_void,
});
/// Empties the given database. All items will be removed. pubfn clear_db(&mutself, db: Database) -> Result<()> { unsafe { lmdb_result(ffi::mdb_drop(self.txn(), db.dbi(), 0)) }
}
/// Drops the database from the environment. /// /// ## Safety /// /// This method is unsafe in the same ways as `Environment::close_db`, and /// should be used accordingly. pubunsafefn drop_db(&mutself, db: Database) -> Result<()> {
lmdb_result(ffi::mdb_drop(self.txn, db.dbi(), 1))
}
/// Begins a new nested transaction inside of this transaction. pubfn begin_nested_txn<'txn>(&'txn mutself) -> Result<RwTransaction<'txn>> { letmut nested: *mut ffi::MDB_txn = ptr::null_mut(); unsafe { let env: *mut ffi::MDB_env = ffi::mdb_txn_env(self.txn());
ffi::mdb_txn_begin(env, self.txn(), 0, &mut nested);
}
Ok(RwTransaction {
txn: nested,
_marker: PhantomData,
})
}
}
use std::io::Write; use std::sync::{
Arc,
Barrier,
}; use std::thread::{ self,
JoinHandle,
};
use tempdir::TempDir;
usesuper::*; use cursor::Cursor; use error::*; use flags::*;
#[test] fn test_put_get_del() { let dir = TempDir::new("test").unwrap(); let env = Environment::new().open(dir.path()).unwrap(); let db = env.open_db(None).unwrap();
#[test] fn test_put_get_del_multi() { let dir = TempDir::new("test").unwrap(); let env = Environment::new().open(dir.path()).unwrap(); let db = env.create_db(None, DatabaseFlags::DUP_SORT).unwrap();
let txn = env.begin_rw_txn().unwrap();
{ letmut cur = txn.open_ro_cursor(db).unwrap(); let iter = cur.iter_dup_of(b"key1"); let vals = iter.map(|x| x.unwrap()).map(|(_, x)| x).collect::<Vec<_>>();
assert_eq!(vals, vec![b"val1", b"val3"]);
let iter = cur.iter_dup_of(b"key2");
assert_eq!(0, iter.count());
}
txn.commit().unwrap();
}
#[test] fn test_reserve() { let dir = TempDir::new("test").unwrap(); let env = Environment::new().open(dir.path()).unwrap(); let db = env.open_db(None).unwrap();
#[test] fn test_inactive_txn() { let dir = TempDir::new("test").unwrap(); let env = Environment::new().open(dir.path()).unwrap(); let db = env.open_db(None).unwrap();
let txn = env.begin_ro_txn().unwrap(); let inactive = txn.reset(); let active = inactive.renew().unwrap();
assert!(active.get(db, b"key").is_ok());
}
#[test] fn test_nested_txn() { let dir = TempDir::new("test").unwrap(); let env = Environment::new().open(dir.path()).unwrap(); let db = env.open_db(None).unwrap();
#[test] fn test_clear_db() { let dir = TempDir::new("test").unwrap(); let env = Environment::new().open(dir.path()).unwrap(); let db = env.open_db(None).unwrap();
let txn = env.begin_ro_txn().unwrap();
assert_eq!(txn.get(db, b"key"), Err(Error::NotFound));
}
#[test] fn test_drop_db() { let dir = TempDir::new("test").unwrap(); let env = Environment::new().set_max_dbs(2).open(dir.path()).unwrap(); let db = env.create_db(Some("test"), DatabaseFlags::empty()).unwrap();
#[test] fn test_concurrent_readers_single_writer() { let dir = TempDir::new("test").unwrap(); let env: Arc<Environment> = Arc::new(Environment::new().open(dir.path()).unwrap());
let n = 10usize; // Number of concurrent readers let barrier = Arc::new(Barrier::new(n + 1)); letmut threads: Vec<JoinHandle<bool>> = Vec::with_capacity(n);
let key = b"key"; let val = b"val";
for _ in0..n { let reader_env = env.clone(); let reader_barrier = barrier.clone();
threads.push(thread::spawn(move || { let db = reader_env.open_db(None).unwrap();
{ let txn = reader_env.begin_ro_txn().unwrap();
assert_eq!(txn.get(db, key), Err(Error::NotFound));
txn.abort();
}
reader_barrier.wait();
reader_barrier.wait();
{ let txn = reader_env.begin_ro_txn().unwrap();
txn.get(db, key).unwrap() == val
}
}));
}
let db = env.open_db(None).unwrap(); letmut txn = env.begin_rw_txn().unwrap();
barrier.wait();
txn.put(db, key, val, WriteFlags::empty()).unwrap();
txn.commit().unwrap();
barrier.wait();
let db = env.open_db(None).unwrap(); let txn = env.begin_ro_txn().unwrap();
for i in0..n {
assert_eq!(format!("{}{}", val, i).as_bytes(), txn.get(db, &format!("{}{}", key, i)).unwrap());
}
}
#[test] fn test_stat() { let dir = TempDir::new("test").unwrap(); let env = Environment::new().open(dir.path()).unwrap(); let db = env.create_db(None, DatabaseFlags::empty()).unwrap();
{ let txn = env.begin_ro_txn().unwrap(); let stat = txn.stat(db).unwrap();
assert_eq!(stat.entries(), 4);
}
}
#[test] fn test_stat_dupsort() { let dir = TempDir::new("test").unwrap(); let env = Environment::new().open(dir.path()).unwrap(); let db = env.create_db(None, DatabaseFlags::DUP_SORT).unwrap();
Die Informationen auf dieser Webseite wurden
nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit,
noch Qualität der bereit gestellten Informationen zugesichert.
Bemerkung:
Die farbliche Syntaxdarstellung und die Messung sind noch experimentell.