snowykami/meilisearch
1
0
Fork 0
mirror of https://github.com/meilisearch/meilisearch.git synced 2025-09-05 20:26:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Optim O: Put instead of update

Browse Source
This commit is contained in:
Mubelotix
2025-08-12 13:47:40 +02:00
parent 915cd1f108
commit e1fa79e92a
1 changed files with 27 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
crates/index-scheduler/src/scheduler/process_batch.rs
View File

@ -1,5 +1,5 @@
use std::collections::{BTreeSet, HashMap, HashSet}; use std::collections::{BTreeSet, HashMap, HashSet};
use std::ops::{Range, RangeInclusive}; use std::ops::RangeInclusive;
use std::panic::{catch_unwind, AssertUnwindSafe}; use std::panic::{catch_unwind, AssertUnwindSafe};
use std::sync::atomic::Ordering; use std::sync::atomic::Ordering;
@ -805,29 +805,7 @@ impl IndexScheduler {
self.queue.batches.finished_at, self.queue.batches.finished_at,
)?; )?;
// 9. Remove tasks from indexes, statuses, and kinds // 9. Remove batches metadata from indexes, statuses, and kinds
progress.update_progress(TaskDeletionProgress::DeletingTasksMetadata);
let (atomic_progress, task_progress) = AtomicTaskStep::new(
(affected_indexes.len() + affected_statuses.len() + affected_kinds.len()) as u32,
);
progress.update_progress(task_progress);
for index in affected_indexes.iter() {
self.queue.tasks.update_index(wtxn, index, |bitmap| *bitmap -= &to_delete_tasks)?;
atomic_progress.fetch_add(1, Ordering::Relaxed);
}
for status in affected_statuses.iter() {
self.queue.tasks.update_status(wtxn, *status, |bitmap| *bitmap -= &to_delete_tasks)?;
atomic_progress.fetch_add(1, Ordering::Relaxed);
}
for kind in affected_kinds.iter() {
self.queue.tasks.update_kind(wtxn, *kind, |bitmap| *bitmap -= &to_delete_tasks)?;
atomic_progress.fetch_add(1, Ordering::Relaxed);
}
// 10. Remove batches metadata from indexes, statuses, and kinds
progress.update_progress(TaskDeletionProgress::DeletingBatchesMetadata); progress.update_progress(TaskDeletionProgress::DeletingBatchesMetadata);
for (index, batches) in to_remove_from_indexes { for (index, batches) in to_remove_from_indexes {
@ -848,6 +826,31 @@ impl IndexScheduler {
})?; })?;
} }
// 10. Remove tasks from indexes, statuses, and kinds
progress.update_progress(TaskDeletionProgress::DeletingTasksMetadata);
let (atomic_progress, task_progress) = AtomicTaskStep::new(
(affected_indexes.len() + affected_statuses.len() + affected_kinds.len()) as u32,
);
progress.update_progress(task_progress);
for (index, mut previous_tasks) in affected_indexes_tasks.into_iter() {
previous_tasks -= &to_delete_tasks;
self.queue.tasks.index_tasks.put(wtxn, index, &previous_tasks)?;
atomic_progress.fetch_add(1, Ordering::Relaxed);
}
for (status, mut previous_tasks) in affected_statuses_tasks.into_iter() {
previous_tasks -= &to_delete_tasks;
self.queue.tasks.status.put(wtxn, &status, &previous_tasks)?;
atomic_progress.fetch_add(1, Ordering::Relaxed);
}
for (kind, mut previous_tasks) in affected_kinds_tasks.into_iter() {
previous_tasks -= &to_delete_tasks;
self.queue.tasks.kind.put(wtxn, &kind, &previous_tasks)?;
atomic_progress.fetch_add(1, Ordering::Relaxed);
}
// 11. Delete tasks // 11. Delete tasks
progress.update_progress(TaskDeletionProgress::DeletingTasks); progress.update_progress(TaskDeletionProgress::DeletingTasks);
let (atomic_progress, task_progress) = let (atomic_progress, task_progress) =