Skip to content

Reuse indexing worker threads across commits #2999

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
marcbachmann wants to merge 1 commit into
base: main
Choose a base branch
from
+135 −53
Open

Reuse indexing worker threads across commits #2999

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
173 changes: 120 additions & 53 deletions src/indexer/index_writer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,17 @@ fn error_in_index_worker_thread(context: &str) -> TantivyError {
))
}

const WORKER_FLUSH_FAILED: &str = "An indexing worker thread panicked before sealing its segments";

/// One commit epoch's work order for a persistent worker: drain `receiver`
/// until it closes (sealing a segment when the memory budget trips), then ack
/// through `flushed`. See [`IndexWriter::add_indexing_worker`].
struct EpochTask<D: Document> {
receiver: AddBatchReceiver<D>,
flushed: crossbeam_channel::Sender<crate::Result<()>>,
delete_cursor: DeleteCursor,
}

#[derive(Clone, bon::Builder)]
/// A builder for creating a new [IndexWriter] for an index.
pub struct IndexWriterOptions {
Expand Down Expand Up @@ -78,6 +89,12 @@ pub struct IndexWriter<D: Document = TantivyDocument> {
options: IndexWriterOptions,

workers_join_handle: Vec<JoinHandle<crate::Result<()>>>,
/// Delivers each worker its next [`EpochTask`]; dropping the senders tells
/// parked workers to exit.
worker_epoch_senders: Vec<crossbeam_channel::Sender<EpochTask<D>>>,
/// Flush acks for the in-flight epoch, waited on by `prepare_commit`. A
/// crossbeam receiver (not a `FutureResult`) keeps `IndexWriter: Sync`.
worker_flushes: Vec<crossbeam_channel::Receiver<crate::Result<()>>>,

index_writer_status: IndexWriterStatus<D>,
operation_sender: AddBatchSender<D>,
Expand Down Expand Up @@ -327,6 +344,8 @@ impl<D: Document> IndexWriter<D> {
segment_updater,

workers_join_handle: vec![],
worker_epoch_senders: vec![],
worker_flushes: vec![],

delete_queue,

Expand Down Expand Up @@ -355,6 +374,8 @@ impl<D: Document> IndexWriter<D> {
// this will stop the indexing thread,
// dropping the last reference to the segment_updater.
self.drop_sender();
// Close epoch channels so parked workers exit and can be joined.
self.worker_epoch_senders.clear();

let former_workers_handles = std::mem::take(&mut self.workers_join_handle);
for join_handle in former_workers_handles {
Expand Down Expand Up @@ -409,54 +430,84 @@ impl<D: Document> IndexWriter<D> {
})
}

/// Spawns a new worker thread for indexing.
/// The thread consumes documents from the pipeline.
/// Spawns a persistent indexing worker that lives for the writer's whole
/// life: each epoch it drains an [`EpochTask`]'s channel, acks through
/// `flushed`, and parks for the next. Replaces the former
/// join-then-respawn per commit, whose fixed cost dominated high-rate
/// commits.
fn add_indexing_worker(&mut self) -> crate::Result<()> {
let document_receiver_clone = self.operation_receiver()?;
let index_writer_bomb = self.index_writer_status.create_bomb();

let segment_updater = self.segment_updater.clone();

let mut delete_cursor = self.delete_queue.cursor();

let mem_budget = self.options.memory_budget_per_thread;
let index = self.index.clone();
let (epoch_sender, epoch_receiver) = crossbeam_channel::unbounded::<EpochTask<D>>();
// Seed the first epoch with the writer's current operation channel.
let (flushed_sender, flushed) = crossbeam_channel::bounded(1);
epoch_sender
.send(EpochTask {
receiver: self.operation_receiver()?,
flushed: flushed_sender,
delete_cursor: self.delete_queue.cursor(),
})
.expect("the worker's epoch receiver cannot be gone: we hold it");
let join_handle: JoinHandle<crate::Result<()>> = thread::Builder::new()
.name(format!("thrd-tantivy-index{}", self.worker_id))
.spawn(move || {
loop {
let mut document_iterator = document_receiver_clone
.clone()
.into_iter()
.filter(|batch| !batch.is_empty())
.peekable();

// The peeking here is to avoid creating a new segment's files
// if no document are available.
//
// This is a valid guarantee as the peeked document now belongs to
// our local iterator.
if let Some(batch) = document_iterator.peek() {
assert!(!batch.is_empty());
delete_cursor.skip_to(batch[0].opstamp);
} else {
// No more documents.
// It happens when there is a commit, or if the `IndexWriter`
// was dropped.
index_writer_bomb.defuse();
return Ok(());
// A closed epoch channel is the shutdown signal.
while let Ok(EpochTask {
receiver,
flushed,
mut delete_cursor,
}) = epoch_receiver.recv()
{
let epoch_result: crate::Result<()> = loop {
let mut document_iterator = receiver
.clone()
.into_iter()
.filter(|batch| !batch.is_empty())
.peekable();

// The peeking here is to avoid creating a new segment's files
// if no document are available.
//
// This is a valid guarantee as the peeked document now belongs to
// our local iterator.
if let Some(batch) = document_iterator.peek() {
assert!(!batch.is_empty());
delete_cursor.skip_to(batch[0].opstamp);
} else {
// Channel drained and closed: flush barrier reached.
break Ok(());
Comment on lines +480 to +482

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 14, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Advance delete cursors on empty epochs

In epochs containing only delete operations (or no add batches), this branch acknowledges the epoch without moving the persistent worker's delete_cursor. Before this change the worker exited on each commit and the replacement cursor started after the delete-only commit; now the cursor can remain before those deletes indefinitely. If delete_all_documents() later rewinds opstamps and new matching docs are added, a subsequent commit can apply the stale delete to the new segment, making documents disappear even though the delete belonged to an earlier cleared generation.

Useful? React with 👍 / 👎.

@marcbachmann marcbachmann Jul 15, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I checked this against the parent commit before concluding anything, and the premise here turns out to be wrong: the replacement cursor did not start after a delete-only commit. DeleteQueue::cursor() positions at the end of the last flushed block, and delete operations pushed since the last consumption sit unflushed in the writer vec. The old code respawned workers inside prepare_commit(), i.e. before the segment updater consumed that commit's deletes, so a respawned worker's fresh cursor iterated those pending deletes exactly like the persistent cursor does now.

The disappearing-documents scenario is real, but pre-existing rather than introduced by this PR. Repro (fails identically on the parent commit and on this branch):

writer.add_document(doc!(text => "hello"))?;
writer.commit()?;
writer.delete_term(Term::from_field_text(text, "hello")); // uncommitted
writer.delete_term(Term::from_field_text(text, "hello")); // uncommitted
writer.delete_all_documents()?;                            // reverts the stamper
writer.commit()?;
writer.add_document(doc!(text => "hello"))?;               // opstamp overlaps stale deletes
writer.commit()?;
// num_docs == 0 on both commits, expected 1

The root cause is that delete_all_documents() reverts the stamper but never clears the delete queue, so uncommitted deletes survive with opstamps above the reverted stamp. Only uncommitted deletes are hazardous (committed ones always sit below the reverted committed_opstamp and get skipped by skip_to), and uncommitted deletes are unflushed — ahead of any cursor, old or new. No cursor-position strategy can fix that; it needs the queue to be cleared on delete_all_documents(). I'll file that separately.

There was one real regression hiding behind this observation though: a parked worker's persistent cursor could pin the delete queue's block history across commits (memory retention, not correctness). Fixed by having each EpochTask carry a fresh delete_queue.cursor(), created at the same point in prepare_commit() where the old code respawned workers — restoring the former semantics exactly.

}

if let Err(err) = index_documents(
mem_budget,
index.new_segment(),
&mut document_iterator,
&segment_updater,
delete_cursor.clone(),
) {
break Err(err);
}
};
let failed = epoch_result.is_err();
// Forward the result verbatim; harmless if nobody waits.
let _ = flushed.send(epoch_result);
if failed {
// Leaving the bomb armed kills the writer.
return Err(error_in_index_worker_thread(
"An indexing worker thread stopped after an indexing error",
));
}

index_documents(
mem_budget,
index.new_segment(),
&mut document_iterator,
&segment_updater,
delete_cursor.clone(),
)?;
}
index_writer_bomb.defuse();
Ok(())
})?;
self.worker_id += 1;
self.worker_epoch_senders.push(epoch_sender);
self.worker_flushes.push(flushed);
self.workers_join_handle.push(join_handle);
Ok(())
}
Expand Down Expand Up @@ -542,15 +593,18 @@ impl<D: Document> IndexWriter<D> {
/// and replace all the channels by new ones.
///
/// The current workers will keep on indexing
/// the pending document and stop
/// the pending document and seal their segment
/// when no documents are remaining.
///
/// Returns the former segment_ready channel.
fn recreate_document_channel(&mut self) {
/// Returns the new receiver, which `prepare_commit` hands to the workers
/// as the next [`EpochTask`] after the current epoch flushes.
fn recreate_document_channel(&mut self) -> AddBatchReceiver<D> {
let (document_sender, document_receiver) =
crossbeam_channel::bounded(PIPELINE_MAX_SIZE_IN_DOCS);
self.operation_sender = document_sender;
self.index_writer_status = IndexWriterStatus::from(document_receiver);
self.index_writer_status
.replace_receiver(document_receiver.clone());
document_receiver
}

/// Rollback to the last commit
Expand Down Expand Up @@ -616,9 +670,9 @@ impl<D: Document> IndexWriter<D> {
/// using this API.
/// See [`PreparedCommit::set_payload()`].
pub fn prepare_commit(&mut self) -> crate::Result<PreparedCommit<'_, D>> {
// Here, because we join all of the worker threads,
// all of the segment update for this commit have been
// sent.
// Here, because we wait for all of the worker threads to reach the
// epoch's flush barrier, all of the segment updates for this commit
// have been sent.
//
// No document belonging to the next commit have been
// pushed too, because add_document can only happen
Expand All @@ -628,18 +682,29 @@ impl<D: Document> IndexWriter<D> {
// committed segments.
info!("Preparing commit");

// this will drop the current document channel
// and recreate a new one.
self.recreate_document_channel();

let former_workers_join_handle = std::mem::take(&mut self.workers_join_handle);

for worker_handle in former_workers_join_handle {
let indexing_worker_result = worker_handle
.join()
.map_err(|e| TantivyError::ErrorInThread(format!("{e:?}")))?;
indexing_worker_result?;
self.add_indexing_worker()?;
// Closing the current epoch channel makes workers drain it, seal their
// segments, ack, and park — replacing the old join-then-respawn per
// commit. Waiting on the acks is the barrier.
let document_receiver = self.recreate_document_channel();
for flushed in std::mem::take(&mut self.worker_flushes) {
// A disconnected channel means the worker panicked before acking.
flushed
.recv()
.map_err(|_| TantivyError::SystemError(WORKER_FLUSH_FAILED.to_string()))??;
}
// Hand every (parked) worker the next epoch's channel.
for epoch_sender in &self.worker_epoch_senders {
let (flushed_sender, flushed) = crossbeam_channel::bounded(1);
epoch_sender
.send(EpochTask {
receiver: document_receiver.clone(),
flushed: flushed_sender,
delete_cursor: self.delete_queue.cursor(),
})
.map_err(|_| {
error_in_index_worker_thread("An indexing worker thread is not running")
})?;
self.worker_flushes.push(flushed);
}

let commit_opstamp = self.stamper.stamp();
Expand Down Expand Up @@ -808,6 +873,8 @@ impl<D: Document> Drop for IndexWriter<D> {
fn drop(&mut self) {
self.segment_updater.kill();
self.drop_sender();
// Close epoch channels so parked workers exit and can be joined.
self.worker_epoch_senders.clear();
for work in self.workers_join_handle.drain(..) {
let _ = work.join();
}
Expand Down
15 changes: 15 additions & 0 deletions src/indexer/index_writer_status.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,21 @@ impl<D: Document> IndexWriterStatus<D> {
inner: Some(self.inner.clone()),
}
}

/// Replaces the operation receiver in place (a commit installs the next
/// epoch's channel). Keeps the same `Inner`, so the bombs handed to the
/// persistent indexing workers stay armed across commits. No-op if the
/// writer was killed.
pub(crate) fn replace_receiver(&self, receiver: AddBatchReceiver<D>) {
let mut wlock = self
.inner
.receive_channel
.write()
.expect("This lock should never be poisoned");
if self.inner.is_alive() {
*wlock = Some(receiver);
}
}
}

struct Inner<D: Document> {
Expand Down