Start9/patch-db
2
0
Fork 0
mirror of https://github.com/Start9Labs/patch-db.git synced 2026-03-26 02:11:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

looser locking

Browse Source
This commit is contained in:
Aiden McClelland
2021-07-19 11:31:40 -06:00
parent 43763539cd
commit 6c7e934b50
4 changed files with 55 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
patch-db/src/handle.rs
View File

@@ -14,7 +14,7 @@ use crate::{
use crate::{patch::DiffPatch, Error};
#[async_trait]
pub trait DbHandle: Sized + Send + Sync {
pub trait DbHandle: Send + Sync {
async fn begin<'a>(&'a mut self) -> Result<Transaction<&'a mut Self>, Error>;
fn rebase(&mut self) -> Result<(), Error>;
fn store(&self) -> Arc<RwLock<Store>>;
@@ -45,6 +45,7 @@ pub trait DbHandle: Sized + Send + Sync {
&mut self,
ptr: &JsonPointer<S, V>,
lock: LockType,
deep: bool,
) -> ();
async fn get<
T: for<'de> Deserialize<'de>,
@@ -65,7 +66,7 @@ pub trait DbHandle: Sized + Send + Sync {
) -> Result<(), Error>;
}
#[async_trait]
impl<Handle: DbHandle + Send + Sync> DbHandle for &mut Handle {
impl<Handle: DbHandle + ?Sized> DbHandle for &mut Handle {
async fn begin<'a>(&'a mut self) -> Result<Transaction<&'a mut Self>, Error> {
let Transaction {
locks,
@@ -127,8 +128,9 @@ impl<Handle: DbHandle + Send + Sync> DbHandle for &mut Handle {
&mut self,
ptr: &JsonPointer<S, V>,
lock: LockType,
deep: bool,
) {
(*self).lock(ptr, lock).await
(*self).lock(ptr, lock, deep).await
}
async fn get<
T: for<'de> Deserialize<'de>,
@@ -229,18 +231,19 @@ impl DbHandle for PatchDbHandle {
&mut self,
ptr: &JsonPointer<S, V>,
lock: LockType,
deep: bool,
) {
match lock {
LockType::Read => {
self.db
.locker
.add_read_lock(ptr, &mut self.locks, &mut [])
.add_read_lock(ptr, &mut self.locks, &mut [], deep)
.await;
}
LockType::Write => {
self.db
.locker
.add_write_lock(ptr, &mut self.locks, &mut [])
.add_write_lock(ptr, &mut self.locks, &mut [], deep)
.await;
}
LockType::None => (),

16
patch-db/src/locker.rs
View File

@@ -80,6 +80,7 @@ impl Locker {
pub async fn lock_read<S: AsRef<str>, V: SegList>(
&self,
ptr: &JsonPointer<S, V>,
deep: bool,
) -> ReadGuard<HashMap<String, Locker>> {
let mut lock = Some(self.0.clone().read().await.unwrap());
for seg in ptr.iter() {
@@ -94,7 +95,9 @@ impl Locker {
lock = Some(new_lock);
}
let res = lock.unwrap();
Self::lock_root_read(&res);
if deep {
Self::lock_root_read(&res);
}
res
}
pub(crate) async fn add_read_lock<S: AsRef<str> + Clone, V: SegList + Clone>(
@@ -102,6 +105,7 @@ impl Locker {
ptr: &JsonPointer<S, V>,
locks: &mut Vec<(JsonPointer, LockerGuard)>,
extra_locks: &mut [&mut [(JsonPointer, LockerGuard)]],
deep: bool,
) {
for lock in extra_locks
.iter()
@@ -114,7 +118,7 @@ impl Locker {
}
locks.push((
JsonPointer::to_owned(ptr.clone()),
LockerGuard::Read(self.lock_read(ptr).await.into()),
LockerGuard::Read(self.lock_read(ptr, deep).await.into()),
));
}
fn lock_root_write<'a>(guard: &'a WriteGuard<HashMap<String, Locker>>) -> BoxFuture<'a, ()> {
@@ -129,6 +133,7 @@ impl Locker {
pub async fn lock_write<S: AsRef<str>, V: SegList>(
&self,
ptr: &JsonPointer<S, V>,
deep: bool,
) -> WriteGuard<HashMap<String, Locker>> {
let mut lock = self.0.clone().write().await.unwrap();
for seg in ptr.iter() {
@@ -141,7 +146,9 @@ impl Locker {
lock = new_lock;
}
let res = lock;
Self::lock_root_write(&res);
if deep {
Self::lock_root_write(&res);
}
res
}
pub(crate) async fn add_write_lock<S: AsRef<str> + Clone, V: SegList + Clone>(
@@ -149,6 +156,7 @@ impl Locker {
ptr: &JsonPointer<S, V>,
locks: &mut Vec<(JsonPointer, LockerGuard)>, // tx locks
extra_locks: &mut [&mut [(JsonPointer, LockerGuard)]], // tx parent locks
deep: bool,
) {
let mut final_lock = None;
for lock in extra_locks
@@ -222,7 +230,7 @@ impl Locker {
if let Some(lock) = final_lock {
lock
} else {
LockerGuard::Write(self.lock_write(ptr).await.into())
LockerGuard::Write(self.lock_write(ptr, deep).await.into())
},
));
}

44
patch-db/src/model.rs
View File

@@ -64,11 +64,13 @@ where
T: Serialize + for<'de> Deserialize<'de>,
{
pub async fn lock<Db: DbHandle>(&self, db: &mut Db, lock: LockType) {
db.lock(&self.ptr, lock).await
db.lock(&self.ptr, lock, true).await
}
pub async fn get<Db: DbHandle>(&self, db: &mut Db) -> Result<ModelData<T>, Error> {
self.lock(db, LockType::Read).await;
pub async fn get<Db: DbHandle>(&self, db: &mut Db, lock: bool) -> Result<ModelData<T>, Error> {
if lock {
self.lock(db, LockType::Read).await;
}
Ok(ModelData(db.get(&self.ptr).await?))
}
@@ -202,11 +204,17 @@ impl<T: HasModel + Serialize + for<'de> Deserialize<'de>> HasModel for Box<T> {
pub struct OptionModel<T: HasModel + Serialize + for<'de> Deserialize<'de>>(T::Model);
impl<T: HasModel + Serialize + for<'de> Deserialize<'de>> OptionModel<T> {
pub async fn lock<Db: DbHandle>(&self, db: &mut Db, lock: LockType) {
db.lock(self.0.as_ref(), lock).await
db.lock(self.0.as_ref(), lock, true).await
}
pub async fn get<Db: DbHandle>(&self, db: &mut Db) -> Result<ModelData<Option<T>>, Error> {
self.lock(db, LockType::Read).await;
pub async fn get<Db: DbHandle>(
&self,
db: &mut Db,
lock: bool,
) -> Result<ModelData<Option<T>>, Error> {
if lock {
self.lock(db, LockType::Read).await;
}
Ok(ModelData(db.get(self.0.as_ref()).await?))
}
@@ -224,8 +232,10 @@ impl<T: HasModel + Serialize + for<'de> Deserialize<'de>> OptionModel<T> {
})
}
pub async fn exists<Db: DbHandle>(&self, db: &mut Db) -> Result<bool, Error> {
self.lock(db, LockType::Read).await;
pub async fn exists<Db: DbHandle>(&self, db: &mut Db, lock: bool) -> Result<bool, Error> {
if lock {
db.lock(self.0.as_ref(), LockType::Read, false).await;
}
Ok(db.exists(&self.as_ref(), None).await?)
}
@@ -252,7 +262,7 @@ impl<T: HasModel + Serialize + for<'de> Deserialize<'de>> OptionModel<T> {
}
pub async fn check<Db: DbHandle>(self, db: &mut Db) -> Result<Option<T::Model>, Error> {
Ok(if self.exists(db).await? {
Ok(if self.exists(db, true).await? {
Some(self.0)
} else {
None
@@ -260,7 +270,7 @@ impl<T: HasModel + Serialize + for<'de> Deserialize<'de>> OptionModel<T> {
}
pub async fn expect<Db: DbHandle>(self, db: &mut Db) -> Result<T::Model, Error> {
if self.exists(db).await? {
if self.exists(db, true).await? {
Ok(self.0)
} else {
Err(Error::NodeDoesNotExist(self.0.into()))
@@ -268,7 +278,7 @@ impl<T: HasModel + Serialize + for<'de> Deserialize<'de>> OptionModel<T> {
}
pub async fn delete<Db: DbHandle>(&self, db: &mut Db) -> Result<(), Error> {
db.lock(self.as_ref(), LockType::Write).await;
db.lock(self.as_ref(), LockType::Write, true).await;
db.put(self.as_ref(), &Value::Null).await
}
}
@@ -277,7 +287,7 @@ where
T: Serialize + for<'de> Deserialize<'de> + Send + Sync + HasModel,
{
pub async fn put<Db: DbHandle>(&self, db: &mut Db, value: &T) -> Result<(), Error> {
db.lock(self.as_ref(), LockType::Write).await;
db.lock(self.as_ref(), LockType::Write, true).await;
db.put(self.as_ref(), value).await
}
}
@@ -435,8 +445,14 @@ where
T::Key: Hash + Eq + for<'de> Deserialize<'de>,
T::Value: Serialize + for<'de> Deserialize<'de>,
{
pub async fn keys<Db: DbHandle>(&self, db: &mut Db) -> Result<IndexSet<T::Key>, Error> {
db.lock(self.as_ref(), LockType::Read).await;
pub async fn keys<Db: DbHandle>(
&self,
db: &mut Db,
lock: bool,
) -> Result<IndexSet<T::Key>, Error> {
if lock {
db.lock(self.as_ref(), LockType::Read, false).await;
}
let set = db.keys(self.as_ref(), None).await?;
Ok(set
.into_iter()

7
patch-db/src/transaction.rs
View File

@@ -159,17 +159,20 @@ impl<Parent: DbHandle + Send + Sync> DbHandle for Transaction<Parent> {
&mut self,
ptr: &JsonPointer<S, V>,
lock: LockType,
deep: bool,
) {
match lock {
LockType::None => (),
LockType::Read => {
let (locker, mut locks) = self.parent.locker_and_locks();
locker.add_read_lock(ptr, &mut self.locks, &mut locks).await
locker
.add_read_lock(ptr, &mut self.locks, &mut locks, deep)
.await
}
LockType::Write => {
let (locker, mut locks) = self.parent.locker_and_locks();
locker
.add_write_lock(ptr, &mut self.locks, &mut locks)
.add_write_lock(ptr, &mut self.locks, &mut locks, deep)
.await
}
}