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

Merge #4861

Browse Source 
4861: Make sure the index scheduler never stops running r=irevoire a=irevoire

# Pull Request

## Related issue
Fixes https://github.com/meilisearch/meilisearch/issues/4748

## What does this PR do?
- Whatever happens, we always try to process tasks once every minute (if no tasks are enqueued that's practically free)

Co-authored-by: Tamo <tamo@meilisearch.com>
This commit is contained in:
meili-bors[bot] 2024-08-07 16:21:54 +00:00 committed by GitHub
commit 321639364f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 11 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
index-scheduler/src/lib.rs
View File

@ -35,6 +35,7 @@ pub type TaskId = u32;
use std::collections::{BTreeMap, HashMap}; use std::collections::{BTreeMap, HashMap};
use std::io::{self, BufReader, Read}; use std::io::{self, BufReader, Read};
use std::ops::{Bound, RangeBounds}; use std::ops::{Bound, RangeBounds};
use std::panic::{catch_unwind, AssertUnwindSafe};
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::sync::atomic::Ordering::{self, Relaxed}; use std::sync::atomic::Ordering::{self, Relaxed};
use std::sync::atomic::{AtomicBool, AtomicU32}; use std::sync::atomic::{AtomicBool, AtomicU32};
@ -612,19 +613,24 @@ impl IndexScheduler {
#[cfg(test)] #[cfg(test)]
run.breakpoint(Breakpoint::Init); run.breakpoint(Breakpoint::Init);
run.wake_up.wait(); run.wake_up.wait_timeout(std::time::Duration::from_secs(60));
loop { loop {
match run.tick() { let ret = catch_unwind(AssertUnwindSafe(|| run.tick()));
Ok(TickOutcome::TickAgain(_)) => (), match ret {
Ok(TickOutcome::WaitForSignal) => run.wake_up.wait(), Ok(Ok(TickOutcome::TickAgain(_))) => (),
Err(e) => { Ok(Ok(TickOutcome::WaitForSignal)) => run.wake_up.wait(),
Ok(Err(e)) => {
tracing::error!("{e}"); tracing::error!("{e}");
// Wait one second when an irrecoverable error occurs. // Wait one second when an irrecoverable error occurs.
if !e.is_recoverable() { if !e.is_recoverable() {
std::thread::sleep(Duration::from_secs(1)); std::thread::sleep(Duration::from_secs(1));
} }
} }
Err(_panic) => {
tracing::error!("Internal error: Unexpected panic in the `IndexScheduler::run` method.");
}
} }
} }
}) })