Change connection type from Mutex to RwLock
This commit is contained in:
1 parent
4b17a5a4a7
commit
ea214d8ac0
1 file changed
+24
-29
+24
-29
@@ -43,6 +43,7 @@ where
|
||||
T: EntityData,
|
||||
{
|
||||
pub(crate) fn handle_save(&self, old: EntityHandle<T>, new: &mut EntityHandle<T>, data: &T) {
|
||||
return;
|
||||
if old.real() && new.real() {
|
||||
return;
|
||||
}
|
||||
@@ -57,6 +58,7 @@ where
|
||||
}
|
||||
}
|
||||
pub(crate) fn handle_delete(&self, handle: &EntityHandle<T>, data: &T) {
|
||||
return;
|
||||
let affected = self.get_affected_queries(data);
|
||||
for q in affected {
|
||||
let mut query = self.queries[q].lock();
|
||||
@@ -189,10 +191,10 @@ pub struct DataRepositoryInner<T: std::clone::Clone + EntityData, D: DbLayer<T>>
|
||||
cache: Vec<T>,
|
||||
temporal: Vec<T>,
|
||||
next_temporal: u32,
|
||||
connection: Arc<Mutex<Connection>>,
|
||||
connection: Arc<RwLock<Connection>>,
|
||||
data_layer: D,
|
||||
query_manger: QueryManger<T, Self, D>,
|
||||
version: u16
|
||||
version: u16,
|
||||
}
|
||||
|
||||
impl<T: std::clone::Clone + EntityData, D: DbLayer<T>> Deref for DataRepository<T, D> {
|
||||
@@ -203,7 +205,7 @@ impl<T: std::clone::Clone + EntityData, D: DbLayer<T>> Deref for DataRepository<
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: EntityData + Clone, D: DbLayer<T>> DataRepository<T, D> {
|
||||
impl<T: EntityData + Clone, D: DbLayer<T>> DataRepositoryInner<T, D> {
|
||||
pub fn query_par(
|
||||
&self,
|
||||
rule: Box<dyn Fn(&T) -> bool + Send + Sync>,
|
||||
@@ -211,9 +213,7 @@ impl<T: EntityData + Clone, D: DbLayer<T>> DataRepository<T, D> {
|
||||
where
|
||||
T: Send + Sync,
|
||||
{
|
||||
let inner = self.inner.read();
|
||||
|
||||
let mut cache: Vec<EntityHandle<T>> = inner
|
||||
let mut cache: Vec<EntityHandle<T>> = self
|
||||
.cache
|
||||
.par_iter()
|
||||
.enumerate()
|
||||
@@ -222,8 +222,7 @@ impl<T: EntityData + Clone, D: DbLayer<T>> DataRepository<T, D> {
|
||||
.collect();
|
||||
|
||||
cache.extend(
|
||||
inner
|
||||
.temporal
|
||||
self.temporal
|
||||
.iter()
|
||||
.enumerate()
|
||||
.filter(|(_, entity)| rule(entity))
|
||||
@@ -246,9 +245,7 @@ impl<T: EntityData + Clone, D: DbLayer<T>> DataRepository<T, D> {
|
||||
where
|
||||
T: Send + Sync,
|
||||
{
|
||||
let inner = self.inner.read();
|
||||
|
||||
let mut cache: Vec<EntityHandle<T>> = inner
|
||||
let mut cache: Vec<EntityHandle<T>> = self
|
||||
.cache
|
||||
.iter()
|
||||
.enumerate()
|
||||
@@ -257,8 +254,7 @@ impl<T: EntityData + Clone, D: DbLayer<T>> DataRepository<T, D> {
|
||||
.collect();
|
||||
|
||||
cache.extend(
|
||||
inner
|
||||
.temporal
|
||||
self.temporal
|
||||
.iter()
|
||||
.enumerate()
|
||||
.filter(|(_, entity)| rule(entity))
|
||||
@@ -324,12 +320,12 @@ impl<T: EntityData + Clone, D: DbLayer<T>> Repository<T, D> for DataRepositoryIn
|
||||
if entity.real() {
|
||||
let data = self.get_ref(entity).unwrap();
|
||||
self.data_layer
|
||||
.write_entity(data, &self.connection.lock())?;
|
||||
.write_entity(data, &self.connection.read())?;
|
||||
} else {
|
||||
let index = fast_position_of(self.temporal.as_slice(), entity).unwrap();
|
||||
let mut data = self.temporal.remove(index);
|
||||
self.data_layer
|
||||
.add_entity(&mut data, &self.connection.lock())?;
|
||||
.add_entity(&mut data, &self.connection.read())?;
|
||||
entity.set_real(true);
|
||||
entity.id = data.id();
|
||||
entity.set_index(self.cache.len());
|
||||
@@ -344,7 +340,7 @@ impl<T: EntityData + Clone, D: DbLayer<T>> Repository<T, D> for DataRepositoryIn
|
||||
fn delete(&mut self, mut handle: EntityHandle<T>) -> MaybeError {
|
||||
let vec = if handle.real() {
|
||||
self.data_layer
|
||||
.delete_entity(handle.id, &self.connection.lock())?;
|
||||
.delete_entity(handle.id, &self.connection.read())?;
|
||||
&mut self.cache
|
||||
} else {
|
||||
&mut self.temporal
|
||||
@@ -361,7 +357,7 @@ impl<T: EntityData + Clone, D: DbLayer<T>> Repository<T, D> for DataRepositoryIn
|
||||
|
||||
fn init(&mut self) -> MaybeError {
|
||||
self.data_layer
|
||||
.load_all(&mut self.cache, &self.connection.lock())?;
|
||||
.load_all(&mut self.cache, &self.connection.read())?;
|
||||
self.version += 1;
|
||||
Ok(())
|
||||
}
|
||||
@@ -388,7 +384,7 @@ impl<T: EntityData + Clone, D: BatchedDbLayer<T> + DbLayer<T>> BatchedRepository
|
||||
}
|
||||
});
|
||||
|
||||
let mut values = Vec::with_capacity(temp.len());
|
||||
let mut temp_values = Vec::with_capacity(temp.len());
|
||||
let mut cell = 0;
|
||||
let len = temp.len();
|
||||
self.temporal.retain(|x| {
|
||||
@@ -399,32 +395,31 @@ impl<T: EntityData + Clone, D: BatchedDbLayer<T> + DbLayer<T>> BatchedRepository
|
||||
let remove = x.id() == find;
|
||||
if remove {
|
||||
cell += 1;
|
||||
values.push(x.clone());
|
||||
temp_values.push(x.clone());
|
||||
}
|
||||
!remove
|
||||
});
|
||||
|
||||
self.data_layer
|
||||
.add_many_entity(&mut values, &self.connection.lock())?;
|
||||
.add_many_entity(&mut temp_values, &self.connection.read())?;
|
||||
|
||||
let start = self.cache.len();
|
||||
|
||||
for (i, (handle, data)) in temp.into_iter().zip(&values).enumerate() {
|
||||
for (i, (handle, data)) in temp.into_iter().zip(&mut temp_values).enumerate() {
|
||||
let old = handle.clone();
|
||||
let h = handle;
|
||||
h.id = self.cache[start + i].id();
|
||||
h.set_index(start + i);
|
||||
h.set_real(true);
|
||||
self.query_manger.handle_save(old, h, data);
|
||||
handle.id = data.id();
|
||||
handle.set_index(start + i);
|
||||
handle.set_real(true);
|
||||
self.query_manger.handle_save(old, handle, data);
|
||||
}
|
||||
|
||||
self.cache.extend(values);
|
||||
self.cache.extend(temp_values);
|
||||
|
||||
for handle in cached.into_iter() {
|
||||
let h = handle;
|
||||
let idx = fast_position_of(self.cache.as_slice(), h).unwrap();
|
||||
self.data_layer
|
||||
.write_entity(&self.cache[idx], &self.connection.lock())?;
|
||||
.write_entity(&self.cache[idx], &self.connection.read())?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
@@ -447,7 +442,7 @@ impl<T: EntityData + Clone, D: BatchedDbLayer<T> + DbLayer<T>> BatchedRepository
|
||||
!remove
|
||||
});
|
||||
self.data_layer
|
||||
.delete_many(cached.as_slice(), &self.connection.lock())?;
|
||||
.delete_many(cached.as_slice(), &self.connection.read())?;
|
||||
let len = cached.len();
|
||||
cell = 0;
|
||||
self.cache.retain(|x| {
|
||||
|
||||
Reference in new issue
Block a user