snowykami/meilisearch
1
0
Fork 0
mirror of https://github.com/meilisearch/meilisearch.git synced 2024-11-26 03:55:07 +08:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

finishes the global skelton of the auto-batcher

Browse Source
This commit is contained in:
Tamo 2022-09-13 14:59:03 +02:00 committed by Clément Renault
parent b3c9b128d9
commit a0588d6b94
No known key found for this signature in database
GPG Key ID: 92ADA4E935E71FA4
2 changed files with 118 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

178
index-scheduler/src/autobatcher.rs
View File

@ -1,3 +1,5 @@
use std::ops::ControlFlow;
use crate::{task::Kind, TaskId}; use crate::{task::Kind, TaskId};
pub enum BatchKind { pub enum BatchKind {
@ -68,136 +70,181 @@ impl BatchKind {
} }
/// Return true if you must stop. /// Return true if you must stop.
fn accumulate(&mut self, id: TaskId, kind: Kind) -> bool { fn accumulate(mut self, id: TaskId, kind: Kind) -> ControlFlow<Self, Self> {
match (self, kind) { match (self, kind) {
// must handle the deleteIndex // We don't batch any of these operations
(_, Kind::CreateIndex | Kind::RenameIndex | Kind::SwapIndex) => true, (this, Kind::CreateIndex | Kind::RenameIndex | Kind::SwapIndex) => {
ControlFlow::Break(this)
(BatchKind::ClearAll { ids }, Kind::ClearAllDocuments | Kind::DocumentDeletion) => { }
// The index deletion can batch with everything but must stop after
(
BatchKind::ClearAll { mut ids }
| BatchKind::DocumentAddition {
addition_ids: mut ids,
}
| BatchKind::DocumentDeletion {
deletion_ids: mut ids,
}
| BatchKind::Settings {
settings_ids: mut ids,
},
Kind::DeleteIndex,
) => {
ids.push(id); ids.push(id);
false ControlFlow::Break(BatchKind::DeleteIndex { ids })
} }
(BatchKind::ClearAll { .. }, Kind::DocumentAddition | Kind::Settings) => true, (
(BatchKind::DocumentAddition { addition_ids }, Kind::ClearAllDocuments) => { BatchKind::ClearAllAndSettings {
addition_ids.push(id); settings_ids: mut ids,
*self = BatchKind::ClearAll { mut other,
ids: addition_ids.clone(), }
}; | BatchKind::SettingsAndDocumentAddition {
false addition_ids: mut ids,
settings_ids: mut other,
},
Kind::DeleteIndex,
) => {
ids.push(id);
ids.append(&mut other);
ControlFlow::Break(BatchKind::DeleteIndex { ids })
} }
(BatchKind::DocumentAddition { addition_ids }, Kind::DocumentAddition) => { (BatchKind::ClearAll { mut ids }, Kind::ClearAllDocuments | Kind::DocumentDeletion) => {
addition_ids.push(id); ids.push(id);
false ControlFlow::Continue(BatchKind::ClearAll { ids })
}
(this @ BatchKind::ClearAll { .. }, Kind::DocumentAddition | Kind::Settings) => {
ControlFlow::Break(this)
}
(BatchKind::DocumentAddition { mut addition_ids }, Kind::ClearAllDocuments) => {
addition_ids.push(id);
ControlFlow::Continue(BatchKind::ClearAll { ids: addition_ids })
}
(BatchKind::DocumentAddition { mut addition_ids }, Kind::DocumentAddition) => {
addition_ids.push(id);
ControlFlow::Continue(BatchKind::DocumentAddition { addition_ids })
}
(this @ BatchKind::DocumentAddition { .. }, Kind::DocumentDeletion) => {
ControlFlow::Break(this)
} }
(BatchKind::DocumentAddition { .. }, Kind::DocumentDeletion) => true,
(BatchKind::DocumentAddition { addition_ids }, Kind::Settings) => { (BatchKind::DocumentAddition { addition_ids }, Kind::Settings) => {
*self = BatchKind::SettingsAndDocumentAddition { ControlFlow::Continue(BatchKind::SettingsAndDocumentAddition {
settings_ids: vec![id], settings_ids: vec![id],
addition_ids: addition_ids.clone(), addition_ids,
}; })
false
} }
(BatchKind::DocumentDeletion { deletion_ids }, Kind::ClearAllDocuments) => { (BatchKind::DocumentDeletion { mut deletion_ids }, Kind::ClearAllDocuments) => {
deletion_ids.push(id); deletion_ids.push(id);
*self = BatchKind::ClearAll { ControlFlow::Continue(BatchKind::ClearAll { ids: deletion_ids })
ids: deletion_ids.clone(),
};
false
} }
(BatchKind::DocumentDeletion { .. }, Kind::DocumentAddition) => true, (this @ BatchKind::DocumentDeletion { .. }, Kind::DocumentAddition) => {
(BatchKind::DocumentDeletion { deletion_ids }, Kind::DocumentDeletion) => { ControlFlow::Break(this)
}
(BatchKind::DocumentDeletion { mut deletion_ids }, Kind::DocumentDeletion) => {
deletion_ids.push(id); deletion_ids.push(id);
false ControlFlow::Continue(BatchKind::DocumentDeletion { deletion_ids })
} }
(BatchKind::DocumentDeletion { .. }, Kind::Settings) => true, (this @ BatchKind::DocumentDeletion { .. }, Kind::Settings) => ControlFlow::Break(this),
(BatchKind::Settings { settings_ids }, Kind::ClearAllDocuments) => { (BatchKind::Settings { settings_ids }, Kind::ClearAllDocuments) => {
*self = BatchKind::ClearAllAndSettings { ControlFlow::Continue(BatchKind::ClearAllAndSettings {
settings_ids: settings_ids.clone(), settings_ids: settings_ids.clone(),
other: vec![id], other: vec![id],
}; })
false
} }
(BatchKind::Settings { .. }, Kind::DocumentAddition) => true, (this @ BatchKind::Settings { .. }, Kind::DocumentAddition) => ControlFlow::Break(this),
(BatchKind::Settings { .. }, Kind::DocumentDeletion) => true, (this @ BatchKind::Settings { .. }, Kind::DocumentDeletion) => ControlFlow::Break(this),
(BatchKind::Settings { settings_ids }, Kind::Settings) => { (BatchKind::Settings { mut settings_ids }, Kind::Settings) => {
settings_ids.push(id); settings_ids.push(id);
false ControlFlow::Continue(BatchKind::Settings { settings_ids })
} }
( (
BatchKind::ClearAllAndSettings { BatchKind::ClearAllAndSettings {
other, mut other,
settings_ids, settings_ids,
}, },
Kind::ClearAllDocuments, Kind::ClearAllDocuments,
) => { ) => {
other.push(id); other.push(id);
false ControlFlow::Continue(BatchKind::ClearAllAndSettings {
other,
settings_ids,
})
}
(this @ BatchKind::ClearAllAndSettings { .. }, Kind::DocumentAddition) => {
ControlFlow::Break(this)
} }
(BatchKind::ClearAllAndSettings { .. }, Kind::DocumentAddition) => true,
( (
BatchKind::ClearAllAndSettings { BatchKind::ClearAllAndSettings {
other, mut other,
settings_ids, settings_ids,
}, },
Kind::DocumentDeletion, Kind::DocumentDeletion,
) => { ) => {
other.push(id); other.push(id);
false ControlFlow::Continue(BatchKind::ClearAllAndSettings {
other,
settings_ids,
})
} }
( (
BatchKind::ClearAllAndSettings { BatchKind::ClearAllAndSettings {
settings_ids, mut settings_ids,
other, other,
}, },
Kind::Settings, Kind::Settings,
) => { ) => {
settings_ids.push(id); settings_ids.push(id);
false ControlFlow::Continue(BatchKind::ClearAllAndSettings {
other,
settings_ids,
})
} }
( (
BatchKind::SettingsAndDocumentAddition { BatchKind::SettingsAndDocumentAddition {
settings_ids, settings_ids,
addition_ids, mut addition_ids,
}, },
Kind::ClearAllDocuments, Kind::ClearAllDocuments,
) => { ) => {
addition_ids.push(id); addition_ids.push(id);
*self = BatchKind::ClearAllAndSettings {
settings_ids: settings_ids.clone(), ControlFlow::Continue(BatchKind::ClearAllAndSettings {
other: addition_ids.clone(), settings_ids,
}; other: addition_ids,
false })
} }
( (
BatchKind::SettingsAndDocumentAddition { BatchKind::SettingsAndDocumentAddition {
mut addition_ids,
settings_ids, settings_ids,
addition_ids,
}, },
Kind::DocumentAddition, Kind::DocumentAddition,
) => { ) => {
addition_ids.push(id); addition_ids.push(id);
false ControlFlow::Continue(BatchKind::SettingsAndDocumentAddition {
addition_ids,
settings_ids,
})
}
(this @ BatchKind::SettingsAndDocumentAddition { .. }, Kind::DocumentDeletion) => {
ControlFlow::Break(this)
} }
( (
BatchKind::SettingsAndDocumentAddition { BatchKind::SettingsAndDocumentAddition {
settings_ids, mut settings_ids,
addition_ids,
},
Kind::DocumentDeletion,
) => true,
(
BatchKind::SettingsAndDocumentAddition {
settings_ids,
addition_ids, addition_ids,
}, },
Kind::Settings, Kind::Settings,
) => { ) => {
settings_ids.push(id); settings_ids.push(id);
false ControlFlow::Continue(BatchKind::SettingsAndDocumentAddition {
settings_ids,
addition_ids,
})
} }
(_, Kind::CancelTask | Kind::DumpExport | Kind::Snapshot) => unreachable!(), (_, Kind::CancelTask | Kind::DumpExport | Kind::Snapshot) => unreachable!(),
( (
@ -222,10 +269,11 @@ pub fn autobatch(enqueued: Vec<(TaskId, Kind)>) -> Option<BatchKind> {
} }
for (id, kind) in enqueued { for (id, kind) in enqueued {
if acc.accumulate(id, kind) { acc = match acc.accumulate(id, kind) {
break; ControlFlow::Continue(acc) => acc,
} ControlFlow::Break(acc) => return Some(acc),
};
} }
Some(acc) None
} }

10
index-scheduler/src/batch.rs
View File

@ -72,8 +72,8 @@ impl IndexScheduler {
}) })
.collect::<Result<Vec<_>>>()?; .collect::<Result<Vec<_>>>()?;
let primary_key = match document_addition_tasks[0].kind { let primary_key = match &document_addition_tasks[0].kind {
KindWithContent::DocumentAddition { primary_key, .. } => primary_key, KindWithContent::DocumentAddition { primary_key, .. } => primary_key.clone(),
_ => unreachable!(), _ => unreachable!(),
}; };
let content_files = document_addition_tasks let content_files = document_addition_tasks
@ -86,7 +86,7 @@ impl IndexScheduler {
let settings = settings_tasks let settings = settings_tasks
.iter() .iter()
.map(|task| match task.kind { .map(|task| match &task.kind {
KindWithContent::Settings { new_settings, .. } => new_settings.to_string(), KindWithContent::Settings { new_settings, .. } => new_settings.to_string(),
_ => unreachable!(), _ => unreachable!(),
}) })
@ -217,11 +217,11 @@ impl IndexScheduler {
let ret = index.update_documents( let ret = index.update_documents(
IndexDocumentsMethod::ReplaceDocuments, IndexDocumentsMethod::ReplaceDocuments,
primary_key, primary_key,
self.file_store, self.file_store.clone(),
content_files.into_iter(), content_files.into_iter(),
)?; )?;
for (ret, task) in ret.iter().zip(document_addition_tasks) { for (ret, mut task) in ret.iter().zip(document_addition_tasks.into_iter()) {
match ret { match ret {
Ok(ret) => task.info = Some(format!("{:?}", ret)), Ok(ret) => task.info = Some(format!("{:?}", ret)),
Err(err) => task.error = Some(err.to_string()), Err(err) => task.error = Some(err.to_string()),