Query manager code

This commit is contained in:
micialware committed 2026-08-28 16:52:48 +03:00
1 parent 70aff6ed18
commit a01a29c492
2 files changed
+186 -39

No files matched your search

+39 -1
View File
@@ -1,3 +1,4 @@
use std::cmp::Ordering;
use crate::app::AppState; use crate::app::AppState;
use crate::core::srs::{FADE_PER_DAY, MAX_SCORE, OrderMode, WordOpenMode}; use crate::core::srs::{FADE_PER_DAY, MAX_SCORE, OrderMode, WordOpenMode};
use crate::data::{INVALID_ID, Id}; use crate::data::{INVALID_ID, Id};
@@ -178,7 +179,7 @@ impl CardStatistics {
} }
} }
#[derive(Clone, Copy)] #[derive(Copy)]
pub struct EntityHandle<E> { pub struct EntityHandle<E> {
pub(crate) id: Id, pub(crate) id: Id,
pub(crate) real: bool, pub(crate) real: bool,
@@ -187,6 +188,43 @@ pub struct EntityHandle<E> {
} }
impl<T> Clone for EntityHandle<T> {
fn clone(&self) -> Self {
EntityHandle{
id: self.id,
real: self.real,
index: self.index,
_phantom: Default::default(),
}
}
}
impl<T> Eq for EntityHandle<T> {}
impl<T> PartialEq<Self> for EntityHandle<T> {
fn eq(&self, other: &Self) -> bool {
self.real == other.real && self.id == other.id
}
}
impl<T> PartialOrd<Self> for EntityHandle<T> {
fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
Some(self.cmp(other))
}
}
impl<T> Ord for EntityHandle<T> {
fn cmp(&self, other: &Self) -> Ordering {
if self.real && !other.real {
return Ordering::Less
}
if !self.real && other.real{
return Ordering::Greater
}
self.id.cmp(&other.id)
}
}
pub trait EntityData{ pub trait EntityData{
fn id(&self) -> Id; fn id(&self) -> Id;
fn set_id(&mut self, id: Id); fn set_id(&mut self, id: Id);
+147 -38
View File
@@ -1,11 +1,13 @@
use crate::data::Id; use crate::data::Id;
use crate::data::entities::{EntityData, EntityHandle}; use crate::data::entities::{EntityData, EntityHandle};
use crate::notification::MaybeError; use crate::helpers::into_result;
use crate::notification::{MaybeError, NotificationData};
use parking_lot::{Mutex, RwLock}; use parking_lot::{Mutex, RwLock};
use rayon::iter::IndexedParallelIterator; use rayon::iter::IndexedParallelIterator;
use rayon::iter::IntoParallelRefIterator; use rayon::iter::IntoParallelRefIterator;
use rayon::iter::ParallelIterator; use rayon::iter::ParallelIterator;
use rusqlite::Connection; use rusqlite::Connection;
use std::cell::RefCell;
use std::marker::PhantomData; use std::marker::PhantomData;
use std::ops::Deref; use std::ops::Deref;
use std::rc::Weak; use std::rc::Weak;
@@ -31,20 +33,75 @@ where
R: Repository<T, D>, R: Repository<T, D>,
D: DbLayer<T>, D: DbLayer<T>,
{ {
queries: Mutex<Vec<Weak<Query<T, R, D>>>>, queries: Vec<Arc<Mutex<Query<T, R, D>>>>,
} }
impl<T, R, D> QueryManger<T, R, D>
where
R: Repository<T, D>,
D: DbLayer<T>,
T: EntityData,
{
pub(crate) fn handle_save(&self, old: EntityHandle<T>, new: &mut EntityHandle<T>, data: &T) {
if old.real && new.real {
return;
}
let affected = self.get_affected_queries(data);
for q in affected {
let mut query = self.queries[q].lock();
let index = query.cache.binary_search(&old).unwrap();
query.cache.remove(index);
let real_end = query.cache.iter().rposition(|r| r.real).unwrap() + 1;
query.cache.insert(real_end, new.clone());
}
}
pub(crate) fn handle_delete(&self, handle: &EntityHandle<T>, data: &T) {
let affected = self.get_affected_queries(data);
for q in affected {
let mut query = self.queries[q].lock();
let slice = query.cache.as_slice();
if let Ok(index) = slice.binary_search(&handle) {
query.cache.remove(index);
}
}
}
fn get_affected_queries(&self, data: &T) -> Vec<usize> {
self.queries
.iter()
.enumerate()
.filter(|(_, q)| {
let query = q.lock();
let rule = &query.rule;
return rule(data);
false
})
.map(|(i, _)| i)
.collect()
}
fn reload_queries(&mut self, data_repository: impl Repository<T, D>) {
for q in &self.queries {
let mut query = q.lock();
query.cache.clear();
}
}
}
pub struct Query<T, R, D> pub struct Query<T, R, D>
where where
R: Repository<T, D>, R: Repository<T, D>,
D: DbLayer<T>, D: DbLayer<T>,
{ {
cache: Vec<EntityHandle<T>>, cache: Vec<EntityHandle<T>>,
rule: Box<dyn Fn(&T) -> bool>, rule: Box<dyn Fn(&T) -> bool + Send + Sync>,
parallel: bool,
_marker: PhantomData<(R, D)>, _marker: PhantomData<(R, D)>,
} }
impl<T, R: Repository<T, D>, D: DbLayer<T>> Query<T, R, D> { impl<T: EntityData + Clone, R: Repository<T, D>, D: DbLayer<T>> Query<T, R, D> {
pub fn iter<'a>(&mut self, data_repository_inner: &'a R) -> impl Iterator<Item = &'a T> pub fn iter<'a>(&mut self, data_repository_inner: &'a R) -> impl Iterator<Item = &'a T>
where where
T: 'a, T: 'a,
@@ -64,6 +121,39 @@ impl<T, R: Repository<T, D>, D: DbLayer<T>> Query<T, R, D> {
} }
} }
} }
pub fn load(&mut self, data_repository: DataRepositoryInner<T, D>) {
self.cache.clear();
if self.parallel {
let mut cache: Vec<EntityHandle<T>> = data_repository
.cache
.par_iter()
.enumerate()
.filter(|(_, entity)| (self.rule)(entity))
.map(|(i, entity)| EntityHandle {
id: entity.id(),
real: true,
index: i as u32,
_phantom: PhantomData,
})
.collect();
cache.extend(
data_repository
.temporal
.iter()
.enumerate()
.filter(|(_, entity)| (self.rule)(entity))
.map(|(i, entity)| EntityHandle {
id: entity.id(),
real: false,
index: i as u32,
_phantom: PhantomData,
})
.collect::<Vec<_>>(),
);
}
}
} }
pub trait BatchedRepository<T, D> pub trait BatchedRepository<T, D>
@@ -135,29 +225,30 @@ impl<T: EntityData + Clone, D: DbLayer<T>> DataRepository<T, D> {
}) })
.collect(); .collect();
cache.extend(inner cache.extend(
.temporal inner
.iter() .temporal
.enumerate() .iter()
.filter(|(_, entity)| rule(entity)) .enumerate()
.map(|(i, entity)| EntityHandle { .filter(|(_, entity)| rule(entity))
id: entity.id(), .map(|(i, entity)| EntityHandle {
real: false, id: entity.id(),
index: i as u32, real: false,
_phantom: PhantomData, index: i as u32,
}).collect::<Vec<_>>()); _phantom: PhantomData,
})
.collect::<Vec<_>>(),
);
Query { Query {
cache, cache,
rule, rule,
parallel: true,
_marker: PhantomData, _marker: PhantomData,
} }
} }
pub fn query( pub fn query(&self, rule: Box<dyn Fn(&T) -> bool + Send + Sync>) -> Query<T, DataRepositoryInner<T, D>, D>
&self,
rule: Box<dyn Fn(&T) -> bool>,
) -> Query<T, DataRepositoryInner<T, D>, D>
where where
T: Send + Sync, T: Send + Sync,
{ {
@@ -176,21 +267,25 @@ impl<T: EntityData + Clone, D: DbLayer<T>> DataRepository<T, D> {
}) })
.collect(); .collect();
cache.extend(inner cache.extend(
.temporal inner
.iter() .temporal
.enumerate() .iter()
.filter(|(_, entity)| rule(entity)) .enumerate()
.map(|(i, entity)| EntityHandle { .filter(|(_, entity)| rule(entity))
id: entity.id(), .map(|(i, entity)| EntityHandle {
real: false, id: entity.id(),
index: i as u32, real: false,
_phantom: PhantomData, index: i as u32,
}).collect::<Vec<_>>()); _phantom: PhantomData,
})
.collect::<Vec<_>>(),
);
Query { Query {
cache, cache,
rule, rule,
parallel: false,
_marker: PhantomData, _marker: PhantomData,
} }
} }
@@ -251,19 +346,22 @@ impl<T: EntityData + Clone, D: DbLayer<T>> Repository<T, D> for DataRepositoryIn
} }
fn save(&mut self, entity: &mut EntityHandle<T>) -> MaybeError { fn save(&mut self, entity: &mut EntityHandle<T>) -> MaybeError {
let original_state = entity.clone();
if entity.real { if entity.real {
let data = self.get_ref(entity).unwrap(); let data = self.get_ref(entity).unwrap();
self.data_layer self.data_layer
.write_entity(data, &self.connection.lock())?; .write_entity(data, &self.connection.lock())?;
} else { } else {
let index = fast_position_of(self.temporal.as_slice(), entity).unwrap(); let index = fast_position_of(self.temporal.as_slice(), entity).unwrap();
let mut value = self.temporal.remove(index); let mut data = self.temporal.remove(index);
self.data_layer self.data_layer
.add_entity(&mut value, &self.connection.lock())?; .add_entity(&mut data, &self.connection.lock())?;
entity.real = true; entity.real = true;
entity.id = value.id(); entity.id = data.id();
entity.index = self.cache.len() as u32; entity.index = self.cache.len() as u32;
self.cache.push(value); self.query_manger.handle_save(original_state, entity, &data);
self.cache.push(data);
} }
Ok(()) Ok(())
@@ -278,8 +376,12 @@ impl<T: EntityData + Clone, D: DbLayer<T>> Repository<T, D> for DataRepositoryIn
&mut self.temporal &mut self.temporal
}; };
let index = fast_position_of(vec.as_slice(), &mut handle).unwrap(); let index = into_result(
vec.remove(index); fast_position_of(vec.as_slice(), &mut handle),
NotificationData::DbSaveError,
)?;
let value = vec.remove(index);
self.query_manger.handle_delete(&handle, &value);
Ok(()) Ok(())
} }
@@ -293,6 +395,7 @@ impl<T: EntityData + Clone, D: DbLayer<T>> Repository<T, D> for DataRepositoryIn
self.cache.clear(); self.cache.clear();
self.temporal.clear(); self.temporal.clear();
self.init()?; self.init()?;
Ok(()) Ok(())
} }
} }
@@ -331,15 +434,20 @@ impl<T: EntityData + Clone, D: BatchedDbLayer<T> + DbLayer<T>> BatchedRepository
.add_many_entity(&mut values, &self.connection.lock())?; .add_many_entity(&mut values, &self.connection.lock())?;
let start = self.cache.len(); let start = self.cache.len();
self.cache.extend(values);
for (i, handle) in temp.into_iter().enumerate() { for (i, (handle, data)) in temp.into_iter().zip(&values).enumerate() {
let old = handle.clone();
let h = handle; let h = handle;
h.id = self.cache[start + i].id(); h.id = self.cache[start + i].id();
h.index = (start + i) as u32; h.index = (start + i) as u32;
h.real = true; h.real = true;
self.query_manger.handle_save(old, h, data);
} }
self.cache.extend(values);
for handle in cached.into_iter() { for handle in cached.into_iter() {
let h = handle; let h = handle;
let idx = fast_position_of(self.cache.as_slice(), h).unwrap(); let idx = fast_position_of(self.cache.as_slice(), h).unwrap();
@@ -379,6 +487,7 @@ impl<T: EntityData + Clone, D: BatchedDbLayer<T> + DbLayer<T>> BatchedRepository
if remove { if remove {
cell += 1; cell += 1;
} }
!remove !remove
}); });