mirror of
https://github.com/quickwit-oss/tantivy.git
synced 2026-08-18 03:58:21 +00:00
Feature/reader (#517)
Adding IndexReader to the API. Making it possible to watch for changes. * Closes #500
This commit is contained in:
Paul Masurel
GitHub
@@ -0,0 +1,187 @@
|
||||
mod pool;
|
||||
|
||||
use self::pool::{LeasedItem, Pool};
|
||||
use core::Segment;
|
||||
use directory::Directory;
|
||||
use directory::WatchHandle;
|
||||
use directory::META_LOCK;
|
||||
use std::sync::Arc;
|
||||
use Index;
|
||||
use Result;
|
||||
use Searcher;
|
||||
use SegmentReader;
|
||||
|
||||
/// Defines when a new version of the index should be reloaded.
|
||||
///
|
||||
/// Regardless of whether you search and index in the same process, tantivy does not necessarily
|
||||
/// reflects the change that are commited to your index. `ReloadPolicy` precisely helps you define
|
||||
/// when you want your index to be reloaded.
|
||||
#[derive(Clone, Copy)]
|
||||
pub enum ReloadPolicy {
|
||||
/// The index is entirely reloaded manually.
|
||||
/// All updates of the index should be manual.
|
||||
///
|
||||
/// No change is reflected automatically. You are required to call `.load_seacher()` manually.
|
||||
Manual,
|
||||
/// The index is reloaded within milliseconds after a new commit is available.
|
||||
/// This is made possible by watching changes in the `meta.json` file.
|
||||
OnCommit, // TODO add NEAR_REAL_TIME(target_ms)
|
||||
}
|
||||
|
||||
/// `IndexReader` builder
|
||||
///
|
||||
/// It makes it possible to set the following values.
|
||||
///
|
||||
/// - `num_searchers` (by default, the number of detected CPU threads):
|
||||
///
|
||||
/// When `num_searchers` queries are requested at the same time, the `num_searchers` will block
|
||||
/// until the one of the searcher in-use gets released.
|
||||
/// - `reload_policy` (by default `ReloadPolicy::OnCommit`):
|
||||
///
|
||||
/// See [`ReloadPolicy`](./enum.ReloadPolicy.html) for more details.
|
||||
#[derive(Clone)]
|
||||
pub struct IndexReaderBuilder {
|
||||
num_searchers: usize,
|
||||
reload_policy: ReloadPolicy,
|
||||
index: Index,
|
||||
}
|
||||
|
||||
impl IndexReaderBuilder {
|
||||
pub(crate) fn new(index: Index) -> IndexReaderBuilder {
|
||||
IndexReaderBuilder {
|
||||
num_searchers: num_cpus::get(),
|
||||
reload_policy: ReloadPolicy::OnCommit,
|
||||
index,
|
||||
}
|
||||
}
|
||||
|
||||
/// Builds the reader.
|
||||
///
|
||||
/// Building the reader is a non-trivial operation that requires
|
||||
/// to open different segment readers. It may take hundreds of milliseconds
|
||||
/// of time and it may return an error.
|
||||
/// TODO(pmasurel) Use the `TryInto` trait once it is available in stable.
|
||||
pub fn try_into(self) -> Result<IndexReader> {
|
||||
let inner_reader = InnerIndexReader {
|
||||
index: self.index,
|
||||
num_searchers: self.num_searchers,
|
||||
searcher_pool: Pool::new(),
|
||||
};
|
||||
inner_reader.reload()?;
|
||||
let inner_reader_arc = Arc::new(inner_reader);
|
||||
let watch_handle_opt: Option<WatchHandle>;
|
||||
match self.reload_policy {
|
||||
ReloadPolicy::Manual => {
|
||||
// No need to set anything...
|
||||
watch_handle_opt = None;
|
||||
}
|
||||
ReloadPolicy::OnCommit => {
|
||||
let inner_reader_arc_clone = inner_reader_arc.clone();
|
||||
let callback = move || {
|
||||
if let Err(err) = inner_reader_arc_clone.reload() {
|
||||
error!(
|
||||
"Error while loading searcher after commit was detected. {:?}",
|
||||
err
|
||||
);
|
||||
}
|
||||
};
|
||||
let watch_handle = inner_reader_arc.index.directory().watch(Box::new(callback));
|
||||
watch_handle_opt = Some(watch_handle);
|
||||
}
|
||||
}
|
||||
Ok(IndexReader {
|
||||
inner: inner_reader_arc,
|
||||
watch_handle_opt,
|
||||
})
|
||||
}
|
||||
|
||||
/// Sets the reload_policy.
|
||||
///
|
||||
/// See [`ReloadPolicy`](./enum.ReloadPolicy.html) for more details.
|
||||
pub fn reload_policy(mut self, reload_policy: ReloadPolicy) -> IndexReaderBuilder {
|
||||
self.reload_policy = reload_policy;
|
||||
self
|
||||
}
|
||||
|
||||
/// Sets the number of `Searcher` in the searcher pool.
|
||||
pub fn num_searchers(mut self, num_searchers: usize) -> IndexReaderBuilder {
|
||||
self.num_searchers = num_searchers;
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
struct InnerIndexReader {
|
||||
num_searchers: usize,
|
||||
searcher_pool: Pool<Searcher>,
|
||||
index: Index,
|
||||
}
|
||||
|
||||
impl InnerIndexReader {
|
||||
fn reload(&self) -> Result<()> {
|
||||
let segment_readers: Vec<SegmentReader> = {
|
||||
let _meta_lock = self.index.directory().acquire_lock(&META_LOCK)?;
|
||||
let searchable_segments = self.searchable_segments()?;
|
||||
searchable_segments
|
||||
.iter()
|
||||
.map(SegmentReader::open)
|
||||
.collect::<Result<_>>()?
|
||||
};
|
||||
let schema = self.index.schema();
|
||||
let searchers = (0..self.num_searchers)
|
||||
.map(|_| Searcher::new(schema.clone(), self.index.clone(), segment_readers.clone()))
|
||||
.collect();
|
||||
self.searcher_pool.publish_new_generation(searchers);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Returns the list of segments that are searchable
|
||||
fn searchable_segments(&self) -> Result<Vec<Segment>> {
|
||||
self.index.searchable_segments()
|
||||
}
|
||||
|
||||
fn searcher(&self) -> LeasedItem<Searcher> {
|
||||
self.searcher_pool.acquire()
|
||||
}
|
||||
}
|
||||
|
||||
/// `IndexReader` is your entry point to read and search the index.
|
||||
///
|
||||
/// It controls when a new version of the index should be loaded and lends
|
||||
/// you instances of `Searcher` for the last loaded version.
|
||||
///
|
||||
/// `Clone` does not clone the different pool of searcher. `IndexReader`
|
||||
/// just wraps and `Arc`.
|
||||
#[derive(Clone)]
|
||||
pub struct IndexReader {
|
||||
inner: Arc<InnerIndexReader>,
|
||||
watch_handle_opt: Option<WatchHandle>,
|
||||
}
|
||||
|
||||
impl IndexReader {
|
||||
/// Update searchers so that they reflect the state of the last
|
||||
/// `.commit()`.
|
||||
///
|
||||
/// If you set up the `OnCommit` `ReloadPolicy` (which is the default)
|
||||
/// every commit should be rapidly reflected on your `IndexReader` and you should
|
||||
/// not need to call `reload()` at all.
|
||||
///
|
||||
/// This automatic reload can take 10s of milliseconds to kick in however, and in unit tests
|
||||
/// it can be nice to deterministically force the reload of searchers.
|
||||
pub fn reload(&self) -> Result<()> {
|
||||
self.inner.reload()
|
||||
}
|
||||
|
||||
/// Returns a searcher
|
||||
///
|
||||
/// This method should be called every single time a search
|
||||
/// query is performed.
|
||||
/// The searchers are taken from a pool of `num_searchers` searchers.
|
||||
/// If no searcher is available
|
||||
/// this may block.
|
||||
///
|
||||
/// The same searcher must be used for a given query, as it ensures
|
||||
/// the use of a consistent segment set.
|
||||
pub fn searcher(&self) -> LeasedItem<Searcher> {
|
||||
self.inner.searcher()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,225 @@
|
||||
use crossbeam::crossbeam_channel::unbounded;
|
||||
use crossbeam::{Receiver, RecvError, Sender};
|
||||
use std::ops::{Deref, DerefMut};
|
||||
use std::sync::atomic::AtomicUsize;
|
||||
use std::sync::atomic::Ordering;
|
||||
use std::sync::Arc;
|
||||
|
||||
pub struct GenerationItem<T> {
|
||||
generation: usize,
|
||||
item: T,
|
||||
}
|
||||
|
||||
/// Queue implementation for the Object Pool below
|
||||
/// Uses the unbounded Linked-List type queue from crossbeam-channel
|
||||
/// Splits the Queue into sender and receiver
|
||||
struct Queue<T> {
|
||||
sender: Sender<T>,
|
||||
receiver: Receiver<T>,
|
||||
}
|
||||
|
||||
impl<T> Queue<T> {
|
||||
fn new() -> Self {
|
||||
let (s, r) = unbounded();
|
||||
Queue {
|
||||
sender: s,
|
||||
receiver: r,
|
||||
}
|
||||
}
|
||||
|
||||
/// Sender trait returns a Result type, which is ignored.
|
||||
/// The Result is not handled at the moment
|
||||
fn push(&self, elem: T) {
|
||||
self.sender
|
||||
.send(elem)
|
||||
.expect("Sending an item to crossbeam-queue shouldn't fail");
|
||||
}
|
||||
|
||||
/// Relies on the underlying crossbeam-channel Receiver
|
||||
/// to block on empty queue
|
||||
fn pop(&self) -> Result<T, RecvError> {
|
||||
self.receiver.recv()
|
||||
}
|
||||
}
|
||||
|
||||
/// An object pool
|
||||
///
|
||||
/// This is used in tantivy to create a pool of `Searcher`.
|
||||
/// Object are wrapped in a `LeasedItem` wrapper and are
|
||||
/// released automatically back into the pool on `Drop`.
|
||||
pub struct Pool<T> {
|
||||
queue: Arc<Queue<GenerationItem<T>>>,
|
||||
freshest_generation: AtomicUsize,
|
||||
next_generation: AtomicUsize,
|
||||
}
|
||||
|
||||
impl<T> Pool<T> {
|
||||
pub fn new() -> Pool<T> {
|
||||
let queue = Arc::new(Queue::new());
|
||||
Pool {
|
||||
queue,
|
||||
freshest_generation: AtomicUsize::default(),
|
||||
next_generation: AtomicUsize::default(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Publishes a new generation of `Searcher`.
|
||||
///
|
||||
/// After publish, all new `Searcher` acquired will be
|
||||
/// of the new generation.
|
||||
pub fn publish_new_generation(&self, items: Vec<T>) {
|
||||
let next_generation = self.next_generation.fetch_add(1, Ordering::SeqCst) + 1;
|
||||
for item in items {
|
||||
let gen_item = GenerationItem {
|
||||
item,
|
||||
generation: next_generation,
|
||||
};
|
||||
self.queue.push(gen_item);
|
||||
}
|
||||
self.advertise_generation(next_generation);
|
||||
}
|
||||
|
||||
/// At the exit of this method,
|
||||
/// - freshest_generation has a value greater or equal than generation
|
||||
/// - freshest_generation has a value that has been advertised
|
||||
/// - freshest_generation has)
|
||||
fn advertise_generation(&self, generation: usize) {
|
||||
// not optimal at all but the easiest to read proof.
|
||||
loop {
|
||||
let former_generation = self.freshest_generation.load(Ordering::Acquire);
|
||||
if former_generation >= generation {
|
||||
break;
|
||||
}
|
||||
self.freshest_generation.compare_and_swap(
|
||||
former_generation,
|
||||
generation,
|
||||
Ordering::SeqCst,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
fn generation(&self) -> usize {
|
||||
self.freshest_generation.load(Ordering::Acquire)
|
||||
}
|
||||
|
||||
/// Acquires a new searcher.
|
||||
///
|
||||
/// If no searcher is available, this methods block until
|
||||
/// a searcher is released.
|
||||
pub fn acquire(&self) -> LeasedItem<T> {
|
||||
let generation = self.generation();
|
||||
loop {
|
||||
let gen_item = self.queue.pop().unwrap();
|
||||
if gen_item.generation >= generation {
|
||||
return LeasedItem {
|
||||
gen_item: Some(gen_item),
|
||||
recycle_queue: Arc::clone(&self.queue),
|
||||
};
|
||||
} else {
|
||||
// this searcher is obsolete,
|
||||
// removing it from the pool.
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct LeasedItem<T> {
|
||||
gen_item: Option<GenerationItem<T>>,
|
||||
recycle_queue: Arc<Queue<GenerationItem<T>>>,
|
||||
}
|
||||
|
||||
impl<T> Deref for LeasedItem<T> {
|
||||
type Target = T;
|
||||
|
||||
fn deref(&self) -> &T {
|
||||
&self
|
||||
.gen_item
|
||||
.as_ref()
|
||||
.expect("Unwrapping a leased item should never fail")
|
||||
.item // unwrap is safe here
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> DerefMut for LeasedItem<T> {
|
||||
fn deref_mut(&mut self) -> &mut T {
|
||||
&mut self
|
||||
.gen_item
|
||||
.as_mut()
|
||||
.expect("Unwrapping a mut leased item should never fail")
|
||||
.item // unwrap is safe here
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Drop for LeasedItem<T> {
|
||||
fn drop(&mut self) {
|
||||
if let Some(gen_item) = self.gen_item.take() {
|
||||
self.recycle_queue.push(gen_item);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
|
||||
use super::Pool;
|
||||
use super::Queue;
|
||||
use std::iter;
|
||||
|
||||
#[test]
|
||||
fn test_pool() {
|
||||
let items10: Vec<usize> = iter::repeat(10).take(10).collect();
|
||||
let pool = Pool::new();
|
||||
pool.publish_new_generation(items10);
|
||||
for _ in 0..20 {
|
||||
assert_eq!(*pool.acquire(), 10);
|
||||
}
|
||||
let items11: Vec<usize> = iter::repeat(11).take(10).collect();
|
||||
pool.publish_new_generation(items11);
|
||||
for _ in 0..20 {
|
||||
assert_eq!(*pool.acquire(), 11);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_queue() {
|
||||
let q = Queue::new();
|
||||
let elem = 5;
|
||||
q.push(elem);
|
||||
let res = q.pop();
|
||||
assert_eq!(res.unwrap(), elem);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_pool_dont_panic_on_empty_pop() {
|
||||
// When the object pool is exhausted, it shouldn't panic on pop()
|
||||
use std::sync::Arc;
|
||||
use std::{thread, time};
|
||||
|
||||
// Wrap the pool in an Arc, same way as its used in `core/index.rs`
|
||||
let pool = Arc::new(Pool::new());
|
||||
// clone pools outside the move scope of each new thread
|
||||
let pool1 = Arc::clone(&pool);
|
||||
let pool2 = Arc::clone(&pool);
|
||||
let elements_for_pool = vec![1, 2];
|
||||
pool.publish_new_generation(elements_for_pool);
|
||||
|
||||
let mut threads = vec![];
|
||||
let sleep_dur = time::Duration::from_millis(10);
|
||||
// spawn one more thread than there are elements in the pool
|
||||
threads.push(thread::spawn(move || {
|
||||
// leasing to make sure it's not dropped before sleep is called
|
||||
let _leased_searcher = &pool.acquire();
|
||||
thread::sleep(sleep_dur);
|
||||
}));
|
||||
threads.push(thread::spawn(move || {
|
||||
// leasing to make sure it's not dropped before sleep is called
|
||||
let _leased_searcher = &pool1.acquire();
|
||||
thread::sleep(sleep_dur);
|
||||
}));
|
||||
threads.push(thread::spawn(move || {
|
||||
// leasing to make sure it's not dropped before sleep is called
|
||||
let _leased_searcher = &pool2.acquire();
|
||||
thread::sleep(sleep_dur);
|
||||
}));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user