perf(workers): strict priority pipeline across background workers
Workers now form a priority pipeline (thumbnails -> metadata -> embeddings -> tagging): a worker only claims jobs when no higher-priority tier has claimable work, so stages drain one at a time at full speed instead of all workers contending for CPU (shared rayon pool), disk, and the DB writer simultaneously. Each tier gate is a single read-only EXISTS query per idle tick using the same exclusion set the corresponding claim would (that worker''s paused folders plus actively-indexing folders), so paused or mid-scan work never blocks lower tiers, and failed jobs stop gating once they leave pending. Deferred workers wake within one backoff interval (250-750ms) of the higher tier draining. A watcher-driven trickle of new files briefly pauses lower tiers by design. A user-initiated tag run during a large embedding backlog waits for it; pausing embeddings per folder overrides if tags are wanted first.
This commit is contained in:
@@ -1152,6 +1152,57 @@ fn get_pending_thumbnail_jobs_excluding(
|
|||||||
Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
|
Ok(rows.collect::<rusqlite::Result<Vec<_>>>()?)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// True if any claimable (pending, non-excluded) jobs exist in `job_table`.
|
||||||
|
/// Used by lower-priority workers to defer to higher-priority queues.
|
||||||
|
/// `extra_predicate` narrows the image join (e.g. videos only for metadata).
|
||||||
|
fn has_claimable_jobs(
|
||||||
|
conn: &Connection,
|
||||||
|
job_table: &str,
|
||||||
|
extra_predicate: &str,
|
||||||
|
excluded_folder_ids: &std::collections::HashSet<i64>,
|
||||||
|
) -> Result<bool> {
|
||||||
|
let sql = format!(
|
||||||
|
"SELECT EXISTS(
|
||||||
|
SELECT 1
|
||||||
|
FROM {} j
|
||||||
|
JOIN images i ON i.id = j.image_id
|
||||||
|
WHERE j.status = 'pending'
|
||||||
|
{}
|
||||||
|
{}
|
||||||
|
)",
|
||||||
|
job_table,
|
||||||
|
extra_predicate,
|
||||||
|
folder_exclusion_clause("i", excluded_folder_ids)
|
||||||
|
);
|
||||||
|
Ok(conn.query_row(&sql, [], |row| row.get::<_, i64>(0))? != 0)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn has_claimable_thumbnail_jobs(
|
||||||
|
conn: &Connection,
|
||||||
|
excluded_folder_ids: &std::collections::HashSet<i64>,
|
||||||
|
) -> Result<bool> {
|
||||||
|
has_claimable_jobs(conn, "thumbnail_jobs", "", excluded_folder_ids)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn has_claimable_metadata_jobs(
|
||||||
|
conn: &Connection,
|
||||||
|
excluded_folder_ids: &std::collections::HashSet<i64>,
|
||||||
|
) -> Result<bool> {
|
||||||
|
has_claimable_jobs(
|
||||||
|
conn,
|
||||||
|
"metadata_jobs",
|
||||||
|
"AND i.media_kind = 'video'",
|
||||||
|
excluded_folder_ids,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn has_claimable_embedding_jobs(
|
||||||
|
conn: &Connection,
|
||||||
|
excluded_folder_ids: &std::collections::HashSet<i64>,
|
||||||
|
) -> Result<bool> {
|
||||||
|
has_claimable_jobs(conn, "embedding_jobs", "", excluded_folder_ids)
|
||||||
|
}
|
||||||
|
|
||||||
pub fn claim_thumbnail_jobs(
|
pub fn claim_thumbnail_jobs(
|
||||||
conn: &mut Connection,
|
conn: &mut Connection,
|
||||||
active_folder_ids: &std::collections::HashSet<i64>,
|
active_folder_ids: &std::collections::HashSet<i64>,
|
||||||
|
|||||||
@@ -121,6 +121,50 @@ static FOLDER_STORAGE_PROFILES: OnceLock<Mutex<HashMap<i64, RuntimeAdaptiveProfi
|
|||||||
static DB_WRITE_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
|
static DB_WRITE_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
|
||||||
const EMBEDDING_BATCH_SIZE: usize = 8;
|
const EMBEDDING_BATCH_SIZE: usize = 8;
|
||||||
|
|
||||||
|
/// Background workers form a strict priority pipeline: a worker only claims
|
||||||
|
/// jobs when no higher-priority claimable work exists, so stages drain one
|
||||||
|
/// at a time (thumbnails → metadata → embeddings → tagging) instead of all
|
||||||
|
/// workers contending for CPU, disk, and the DB writer at once.
|
||||||
|
#[derive(Clone, Copy, PartialEq, PartialOrd)]
|
||||||
|
enum WorkerTier {
|
||||||
|
Thumbnail = 0,
|
||||||
|
Metadata = 1,
|
||||||
|
Embedding = 2,
|
||||||
|
Tagging = 3,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// True if any tier above `own_tier` still has claimable work. Each check
|
||||||
|
/// uses the same exclusion set the corresponding claim would (that worker's
|
||||||
|
/// paused folders plus actively-indexing folders), so paused or mid-scan
|
||||||
|
/// work never gates lower tiers.
|
||||||
|
fn higher_priority_work_pending(pool: &DbPool, own_tier: WorkerTier) -> Result<bool> {
|
||||||
|
let conn = pool.get()?;
|
||||||
|
let active_folders = active_indexing_folders();
|
||||||
|
|
||||||
|
if own_tier > WorkerTier::Thumbnail {
|
||||||
|
let mut excluded = paused_folder_ids("thumbnail");
|
||||||
|
excluded.extend(active_folders.iter().copied());
|
||||||
|
if db::has_claimable_thumbnail_jobs(&conn, &excluded)? {
|
||||||
|
return Ok(true);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if own_tier > WorkerTier::Metadata {
|
||||||
|
let mut excluded = paused_folder_ids("metadata");
|
||||||
|
excluded.extend(active_folders.iter().copied());
|
||||||
|
if db::has_claimable_metadata_jobs(&conn, &excluded)? {
|
||||||
|
return Ok(true);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if own_tier > WorkerTier::Embedding {
|
||||||
|
let mut excluded = paused_folder_ids("embedding");
|
||||||
|
excluded.extend(active_folders.iter().copied());
|
||||||
|
if db::has_claimable_embedding_jobs(&conn, &excluded)? {
|
||||||
|
return Ok(true);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Ok(false)
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Clone, Serialize)]
|
#[derive(Clone, Serialize)]
|
||||||
pub struct IndexProgress {
|
pub struct IndexProgress {
|
||||||
pub folder_id: i64,
|
pub folder_id: i64,
|
||||||
@@ -702,6 +746,10 @@ fn process_metadata_batch(
|
|||||||
pool: &DbPool,
|
pool: &DbPool,
|
||||||
media_tools: &MediaTools,
|
media_tools: &MediaTools,
|
||||||
) -> Result<bool> {
|
) -> Result<bool> {
|
||||||
|
if higher_priority_work_pending(pool, WorkerTier::Metadata)? {
|
||||||
|
return Ok(false);
|
||||||
|
}
|
||||||
|
|
||||||
let jobs = {
|
let jobs = {
|
||||||
with_db_write_lock(|| {
|
with_db_write_lock(|| {
|
||||||
let mut conn = pool.get()?;
|
let mut conn = pool.get()?;
|
||||||
@@ -772,6 +820,10 @@ fn process_embedding_batch(
|
|||||||
pool: &DbPool,
|
pool: &DbPool,
|
||||||
embedder: &mut Option<ClipImageEmbedder>,
|
embedder: &mut Option<ClipImageEmbedder>,
|
||||||
) -> Result<bool> {
|
) -> Result<bool> {
|
||||||
|
if higher_priority_work_pending(pool, WorkerTier::Embedding)? {
|
||||||
|
return Ok(false);
|
||||||
|
}
|
||||||
|
|
||||||
let batch_started_at = Instant::now();
|
let batch_started_at = Instant::now();
|
||||||
let claim_started_at = Instant::now();
|
let claim_started_at = Instant::now();
|
||||||
// Exclude folders that are actively indexing (matching the thumbnail and
|
// Exclude folders that are actively indexing (matching the thumbnail and
|
||||||
@@ -1032,6 +1084,10 @@ fn process_tagging_batch(
|
|||||||
return Ok(false);
|
return Ok(false);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if higher_priority_work_pending(pool, WorkerTier::Tagging)? {
|
||||||
|
return Ok(false);
|
||||||
|
}
|
||||||
|
|
||||||
// Exclude actively-indexing folders for the same reason as the other
|
// Exclude actively-indexing folders for the same reason as the other
|
||||||
// workers: don't compete with a running scan.
|
// workers: don't compete with a running scan.
|
||||||
let mut excluded_folders = paused_folder_ids("tagging");
|
let mut excluded_folders = paused_folder_ids("tagging");
|
||||||
|
|||||||
Reference in New Issue
Block a user