persistor

Cargo.toml

@@ -40,7 +40,7 @@ rust-stemmers = "1.2"
 downcast-rs = { version="1.0" }
 tantivy-query-grammar = { version="0.12", path="./query-grammar" }
 bitpacking = {version="0.8", default-features = false, features=["bitpacker4x"]}
-census = "0.4"
+census =  {path="../census"}
 fnv = "1.0.6"
 owned-read = "0.4"
 failure = "0.1"

TODO.md

@@ -0,0 +1,9 @@
+- segment writer serializes if already flush
+- flush if exceed max doc. 
+- unit tests
+- reader from `IndexWriter`
+- configurable limit to flush 
+- bundle directory
+- add index worker when exceeds some doc limit
+- flush before prepare commit.
+- segment_writer should not receive a segment 

src/common/composite_file.rs

@@ -190,11 +190,12 @@ mod test {
     use crate::schema::Field;
     use std::io::Write;
     use std::path::Path;
+    
 
     #[test]
     fn test_composite_file() {
         let path = Path::new("test_path");
-        let mut directory = RAMDirectory::create();
+        let mut directory = RAMDirectory::default();
         {
             let w = directory.open_write(path).unwrap();
             let mut composite_write = CompositeWrite::wrap(w);

src/common/mod.rs

@@ -2,6 +2,7 @@ pub mod bitpacker;
 mod bitset;
 mod composite_file;
 mod counting_writer;
+mod mutable_enum;
 mod serialize;
 mod vint;
 
@@ -9,6 +10,7 @@ pub use self::bitset::BitSet;
 pub(crate) use self::bitset::TinySet;
 pub(crate) use self::composite_file::{CompositeFile, CompositeWrite};
 pub use self::counting_writer::CountingWriter;
+pub(crate) use self::mutable_enum::MutableEnum;
 pub use self::serialize::{BinarySerializable, FixedSize};
 pub use self::vint::{read_u32_vint, serialize_vint_u32, write_u32_vint, VInt};
 pub use byteorder::LittleEndian as Endianness;
@@ -220,10 +222,12 @@ pub(crate) mod test {
         assert_eq!(minmax(vals.into_iter()), None);
     }
 
+    #[test]
     fn test_minmax_one() {
-        assert_eq!(minmax(vec![1].into_iter()), None);
+        assert_eq!(minmax(vec![1].into_iter()), Some((1, 1)));
     }
 
+    #[test]
     fn test_minmax_two() {
         assert_eq!(minmax(vec![1, 2].into_iter()), Some((1, 2)));
         assert_eq!(minmax(vec![2, 1].into_iter()), Some((1, 2)));

src/common/mutable_enum.rs

@@ -0,0 +1,37 @@
+use std::ops::{Deref, DerefMut};
+
+pub(crate) struct MutableEnum<T>(Option<T>);
+
+impl<T> MutableEnum<T> {
+    pub fn wrap(val: T) -> Self {
+        MutableEnum(Some(val))
+    }
+
+    pub fn into(self) -> T {
+        self.0.unwrap()
+    }
+}
+
+impl<T> MutableEnum<T> {
+    pub fn map_mutate<E, F>(&mut self, transformation: F) -> Result<(), E>
+    where
+        F: FnOnce(T) -> Result<T, E>,
+    {
+        self.0 = self.0.take().map(transformation).transpose()?;
+        Ok(())
+    }
+}
+
+impl<T> Deref for MutableEnum<T> {
+    type Target = T;
+
+    fn deref(&self) -> &Self::Target {
+        self.0.as_ref().unwrap()
+    }
+}
+
+impl<T> DerefMut for MutableEnum<T> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        self.0.as_mut().unwrap()
+    }
+}

src/core/index.rs

@@ -12,8 +12,8 @@ use crate::directory::INDEX_WRITER_LOCK;
 use crate::directory::{Directory, RAMDirectory};
 use crate::error::DataCorruption;
 use crate::error::TantivyError;
-use crate::indexer::index_writer::HEAP_SIZE_MIN;
 use crate::indexer::segment_updater::save_new_metas;
+use crate::indexer::IndexWriterConfig;
 use crate::reader::IndexReader;
 use crate::reader::IndexReaderBuilder;
 use crate::schema::Field;
@@ -273,30 +273,14 @@ impl Index {
     pub fn writer_with_num_threads(
         &self,
         num_threads: usize,
-        overall_heap_size_in_bytes: usize,
+        overall_heap_size_in_bytes: u64,
     ) -> crate::Result<IndexWriter> {
-        let directory_lock = self
-            .directory
-            .acquire_lock(&INDEX_WRITER_LOCK)
-            .map_err(|err| {
-                TantivyError::LockFailure(
-                    err,
-                    Some(
-                        "Failed to acquire index lock. If you are using\
-                         a regular directory, this means there is already an \
-                         `IndexWriter` working on this `Directory`, in this process \
-                         or in a different process."
-                            .to_string(),
-                    ),
-                )
-            })?;
-        let heap_size_in_bytes_per_thread = overall_heap_size_in_bytes / num_threads;
-        IndexWriter::new(
-            self,
-            num_threads,
-            heap_size_in_bytes_per_thread,
-            directory_lock,
-        )
+        let config = IndexWriterConfig {
+            max_indexing_threads: num_threads,
+            memory_budget: overall_heap_size_in_bytes,
+            ..Default::default()
+        };
+        self.writer_from_config(config)
     }
 
     /// Creates a multithreaded writer
@@ -309,13 +293,36 @@ impl Index {
     /// If the lockfile already exists, returns `Error::FileAlreadyExists`.
     /// # Panics
     /// If the heap size per thread is too small, panics.
-    pub fn writer(&self, overall_heap_size_in_bytes: usize) -> crate::Result<IndexWriter> {
-        let mut num_threads = num_cpus::get();
-        let heap_size_in_bytes_per_thread = overall_heap_size_in_bytes / num_threads;
-        if heap_size_in_bytes_per_thread < HEAP_SIZE_MIN {
-            num_threads = (overall_heap_size_in_bytes / HEAP_SIZE_MIN).max(1);
-        }
-        self.writer_with_num_threads(num_threads, overall_heap_size_in_bytes)
+    pub fn writer(&self, overall_heap_size_in_bytes: u64) -> crate::Result<IndexWriter> {
+        let config = IndexWriterConfig {
+            max_indexing_threads: num_cpus::get(),
+            memory_budget: overall_heap_size_in_bytes,
+            ..Default::default()
+        };
+        self.writer_from_config(config)
+    }
+
+    /// Creates a new writer with a given configuration.
+    ///
+    /// See [`IndexWriterConfig`](./struct.IndexWriterConfig.html) for more information.
+    pub fn writer_from_config(&self, mut config: IndexWriterConfig) -> crate::Result<IndexWriter> {
+        config.validate()?;
+        let directory_lock = self
+            .directory
+            .acquire_lock(&INDEX_WRITER_LOCK)
+            .map_err(|err| {
+                TantivyError::LockFailure(
+                    err,
+                    Some(
+                        "Failed to acquire index lock. If you are using \
+                         a regular directory, this means there is already an \
+                         `IndexWriter` working on this `Directory`, in this process \
+                         or in a different process."
+                            .to_string(),
+                    ),
+                )
+            })?;
+        IndexWriter::new(self, config, directory_lock)
     }
 
     /// Accessor to the index schema
@@ -469,7 +476,7 @@ mod tests {
             .try_into()
             .unwrap();
         assert_eq!(reader.searcher().num_docs(), 0);
-        test_index_on_commit_reload_policy_aux(field, &index, &reader);
+        test_index_on_commit_reload_policy_aux(field, index.clone(), &index, &reader);
     }
 
     #[cfg(feature = "mmap")]
@@ -493,7 +500,7 @@ mod tests {
                 .try_into()
                 .unwrap();
             assert_eq!(reader.searcher().num_docs(), 0);
-            test_index_on_commit_reload_policy_aux(field, &index, &reader);
+            test_index_on_commit_reload_policy_aux(field, index.clone(), &index, &reader);
         }
 
         #[test]
@@ -535,12 +542,16 @@ mod tests {
                 .try_into()
                 .unwrap();
             assert_eq!(reader.searcher().num_docs(), 0);
-            test_index_on_commit_reload_policy_aux(field, &write_index, &reader);
+            test_index_on_commit_reload_policy_aux(field, read_index, &write_index, &reader);
         }
     }
 
-    fn test_index_on_commit_reload_policy_aux(field: Field, index: &Index, reader: &IndexReader) {
-        let mut reader_index = reader.index();
+    fn test_index_on_commit_reload_policy_aux(
+        field: Field,
+        mut reader_index: Index,
+        index: &Index,
+        reader: &IndexReader,
+    ) {
         let (sender, receiver) = crossbeam::channel::unbounded();
         let _watch_handle = reader_index.directory_mut().watch(Box::new(move || {
             let _ = sender.send(());

src/core/segment.rs

@@ -3,41 +3,140 @@ use crate::core::Index;
 use crate::core::SegmentId;
 use crate::core::SegmentMeta;
 use crate::directory::error::{OpenReadError, OpenWriteError};
-use crate::directory::Directory;
+use crate::directory::{Directory, ManagedDirectory, RAMDirectory};
 use crate::directory::{ReadOnlySource, WritePtr};
 use crate::indexer::segment_serializer::SegmentSerializer;
 use crate::schema::Schema;
 use crate::Opstamp;
 use std::fmt;
+use std::ops::{Deref, DerefMut};
 use std::path::PathBuf;
+use crate::indexer::{ResourceManager};
+
+#[derive(Clone)]
+pub(crate) enum SegmentDirectory {
+    Persisted(ManagedDirectory),
+    Volatile {
+        directory: RAMDirectory,
+    },
+}
+
+impl SegmentDirectory {
+    fn new_volatile(memory_manager: ResourceManager) -> SegmentDirectory {
+        SegmentDirectory::Volatile {
+            directory: RAMDirectory::create_with_memory_manager(memory_manager),
+        }
+    }
+}
+
+impl fmt::Debug for SegmentDirectory {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            SegmentDirectory::Volatile { .. } => write!(f, "volatile")?,
+            SegmentDirectory::Persisted(dir) => write!(f, "Persisted({:?})", dir)?,
+        }
+        Ok(())
+    }
+}
+
+impl From<ManagedDirectory> for SegmentDirectory {
+    fn from(directory: ManagedDirectory) -> Self {
+        SegmentDirectory::Persisted(directory)
+    }
+}
+
+impl Deref for SegmentDirectory {
+    type Target = dyn Directory;
+
+    fn deref(&self) -> &Self::Target {
+        match self {
+            SegmentDirectory::Volatile {
+                directory, ..
+            } => directory,
+            SegmentDirectory::Persisted(dir) => dir,
+        }
+    }
+}
+
+impl DerefMut for SegmentDirectory {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        match self {
+            SegmentDirectory::Volatile { directory, .. } => directory,
+            SegmentDirectory::Persisted(dir) => dir,
+        }
+    }
+}
 
 /// A segment is a piece of the index.
 #[derive(Clone)]
 pub struct Segment {
-    index: Index,
+    schema: Schema,
     meta: SegmentMeta,
+    directory: SegmentDirectory,
 }
 
 impl fmt::Debug for Segment {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "Segment({:?})", self.id().uuid_string())
+        write!(
+            f,
+            "Segment(id={:?}, directory={:?})",
+            self.id().uuid_string(),
+            self.directory
+        )
     }
 }
 
 impl Segment {
-    /// Creates a new segment given an `Index` and a `SegmentId`
-    pub(crate) fn for_index(index: Index, meta: SegmentMeta) -> Segment {
-        Segment { index, meta }
+    pub(crate) fn new_persisted(
+        meta: SegmentMeta,
+        directory: ManagedDirectory,
+        schema: Schema,
+    ) -> Segment {
+        Segment {
+            meta,
+            schema,
+            directory: SegmentDirectory::from(directory),
+        }
     }
 
-    /// Returns the index the segment belongs to.
-    pub fn index(&self) -> &Index {
-        &self.index
+    /// Creates a new segment that embeds its own `RAMDirectory`.
+    ///
+    /// That segment is entirely dissociated from the index directory.
+    /// It will be persisted by a background thread in charge of IO.
+    pub fn new_volatile(meta: SegmentMeta, schema: Schema, resource_manager: ResourceManager) -> Segment {
+        Segment {
+            schema,
+            meta,
+            directory: SegmentDirectory::new_volatile(resource_manager),
+        }
+    }
+
+    /// Creates a new segment given an `Index` and a `SegmentId`
+    pub(crate) fn for_index(index: Index, meta: SegmentMeta) -> Segment {
+        let segment_directory = index.directory().clone();
+        Segment::new_persisted(meta, segment_directory, index.schema())
+    }
+
+    /// Persists a given `Segment` to a directory.
+    pub fn persist(&mut self, mut dest_directory: ManagedDirectory) -> crate::Result<()> {
+        if let SegmentDirectory::Persisted(_) = self.directory {
+            // this segment is already persisted.
+            return Ok(());
+        }
+        if let SegmentDirectory::Volatile { directory, ..} = &self.directory {
+            directory.persist(&mut dest_directory)?;
+        }
+        self.directory = SegmentDirectory::Persisted(dest_directory);
+        Ok(())
+    }
+
+    pub fn into_volatile(&self, memory_manager: ResourceManager) -> Segment {
+        Segment::new_volatile(self.meta.clone(), self.schema.clone(), memory_manager)
     }
 
     /// Returns our index's schema.
     pub fn schema(&self) -> Schema {
-        self.index.schema()
+        self.schema.clone()
     }
 
     /// Returns the segment meta-information
@@ -51,16 +150,18 @@ impl Segment {
     /// as we finalize a fresh new segment.
     pub(crate) fn with_max_doc(self, max_doc: u32) -> Segment {
         Segment {
-            index: self.index,
+            schema: self.schema,
             meta: self.meta.with_max_doc(max_doc),
+            directory: self.directory,
         }
     }
 
     #[doc(hidden)]
     pub fn with_delete_meta(self, num_deleted_docs: u32, opstamp: Opstamp) -> Segment {
         Segment {
-            index: self.index,
+            schema: self.schema,
             meta: self.meta.with_delete_meta(num_deleted_docs, opstamp),
+            directory: self.directory,
         }
     }
 
@@ -80,15 +181,15 @@ impl Segment {
     /// Open one of the component file for a *regular* read.
     pub fn open_read(&self, component: SegmentComponent) -> Result<ReadOnlySource, OpenReadError> {
         let path = self.relative_path(component);
-        let source = self.index.directory().open_read(&path)?;
+        let source = self.directory.open_read(&path)?;
         Ok(source)
     }
 
     /// Open one of the component file for *regular* write.
     pub fn open_write(&mut self, component: SegmentComponent) -> Result<WritePtr, OpenWriteError> {
         let path = self.relative_path(component);
-        let write = self.index.directory_mut().open_write(&path)?;
-        Ok(write)
+        let wrt = self.directory.open_write(&path)?;
+        Ok(wrt)
     }
 }
 

src/directory/mod.rs

@@ -13,7 +13,10 @@ mod footer;
 mod managed_directory;
 mod ram_directory;
 mod read_only_source;
+mod spilling_writer;
 mod watch_event_router;
+mod persistor;
+
 
 /// Errors specific to the directory module.
 pub mod error;
@@ -23,6 +26,7 @@ pub use self::directory::{Directory, DirectoryClone};
 pub use self::directory_lock::{Lock, INDEX_WRITER_LOCK, META_LOCK};
 pub use self::ram_directory::RAMDirectory;
 pub use self::read_only_source::ReadOnlySource;
+pub(crate) use self::spilling_writer::{SpillingResult, SpillingWriter};
 pub use self::watch_event_router::{WatchCallback, WatchCallbackList, WatchHandle};
 use std::io::{self, BufWriter, Write};
 use std::path::PathBuf;
@@ -79,10 +83,16 @@ impl<W: TerminatingWrite> TerminatingWrite for BufWriter<W> {
     }
 }
 
+impl TerminatingWrite for Vec<u8> {
+    fn terminate_ref(&mut self, _a: AntiCallToken) -> io::Result<()> {
+        Ok(())
+    }
+}
+
 #[cfg(test)]
 impl<'a> TerminatingWrite for &'a mut Vec<u8> {
     fn terminate_ref(&mut self, _a: AntiCallToken) -> io::Result<()> {
-        self.flush()
+        Ok(())
     }
 }
 

src/directory/persistor.rs

@@ -0,0 +1,38 @@
+use crate::indexer::{SegmentManager, ResourceManager, MergeOperationInventory};
+use std::thread::JoinHandle;
+use crate::{IndexWriterConfig, SegmentId};
+use std::collections::HashSet;
+
+pub(crate) struct Persistor {
+    memory_manager: ResourceManager,
+    thread_handle: JoinHandle<()>,
+}
+
+impl Persistor {
+    pub(crate) fn create_and_start(segment_manager: SegmentManager,
+                            memory_manager: ResourceManager,
+                            config: IndexWriterConfig) -> crate::Result<Persistor> {
+        let memory_manager_clone = memory_manager.clone();
+        let thread_handle = std::thread::Builder::new()
+            .name("persistor-thread".to_string())
+            .spawn(move || {
+                while let Ok(_) = memory_manager_clone.wait_until_in_range(config.persist_low..) {
+                    segment_manager.largest_segment_not_in_merge();
+                }
+            }).map_err(|_err| crate::TantivyError::ErrorInThread("Failed to start persistor thread.".to_string()))?;
+        Ok(Persistor {
+            memory_manager,
+            thread_handle
+        })
+    }
+
+    /// Stop the persisting thread.
+    ///
+    /// The memory manager will be terminated, which will unlock the thread from any waiting
+    /// position.
+    /// This method blocks for a short amount of tim until the persistor thread has terminated.
+    pub fn stop(self) {
+        self.memory_manager.terminate();
+        let _ = self.thread_handle.join();
+    }
+}

src/directory/ram_directory.rs

@@ -11,6 +11,7 @@ use std::io::{self, BufWriter, Cursor, Seek, SeekFrom, Write};
 use std::path::{Path, PathBuf};
 use std::result;
 use std::sync::{Arc, RwLock};
+use crate::indexer::ResourceManager;
 
 /// Writer associated with the `RAMDirectory`
 ///
@@ -82,11 +83,12 @@ impl TerminatingWrite for VecWriter {
 struct InnerDirectory {
     fs: HashMap<PathBuf, ReadOnlySource>,
     watch_router: WatchCallbackList,
+    memory_manager: ResourceManager,
 }
 
 impl InnerDirectory {
     fn write(&mut self, path: PathBuf, data: &[u8]) -> bool {
-        let data = ReadOnlySource::new(Vec::from(data));
+        let data = ReadOnlySource::new_with_allocation(Vec::from(data), &self.memory_manager);
         self.fs.insert(path, data).is_some()
     }
 
@@ -112,8 +114,8 @@ impl InnerDirectory {
         self.watch_router.subscribe(watch_handle)
     }
 
-    fn total_mem_usage(&self) -> usize {
-        self.fs.values().map(|f| f.len()).sum()
+    fn total_mem_usage(&self) -> u64 {
+        self.fs.values().map(|source| source.len() as u64).sum()
     }
 }
 
@@ -134,14 +136,30 @@ pub struct RAMDirectory {
 }
 
 impl RAMDirectory {
-    /// Constructor
+
+    /// Creates a new RAMDirectory.
+    ///
+    /// Check `.create_with_memory_manager(..)` if you want to associate an external memory
+    /// manager to your RAMDirectory.
     pub fn create() -> RAMDirectory {
-        Self::default()
+        RAMDirectory::default()
+    }
+
+    /// Constructor
+    pub fn create_with_memory_manager(memory_manager: ResourceManager) -> RAMDirectory {
+        let inner_directory = InnerDirectory {
+            fs: Default::default(),
+            watch_router: Default::default(),
+            memory_manager
+        };
+        RAMDirectory {
+            fs: Arc::new(RwLock::new(inner_directory))
+        }
     }
 
     /// Returns the sum of the size of the different files
     /// in the RAMDirectory.
-    pub fn total_mem_usage(&self) -> usize {
+    pub fn total_mem_usage(&self) -> u64 {
         self.fs.read().unwrap().total_mem_usage()
     }
 
@@ -227,6 +245,9 @@ mod tests {
     use crate::Directory;
     use std::io::Write;
     use std::path::Path;
+    use crate::indexer::ResourceManager;
+    use crate::directory::TerminatingWrite;
+    use std::mem;
 
     #[test]
     fn test_persist() {
@@ -244,4 +265,59 @@ mod tests {
         assert_eq!(directory_copy.atomic_read(path_atomic).unwrap(), msg_atomic);
         assert_eq!(directory_copy.atomic_read(path_seq).unwrap(), msg_seq);
     }
+
+    #[test]
+    fn test_memory_manager_several_path() {
+        let memory_manager = ResourceManager::default();
+        let mut ram_directory = RAMDirectory::create_with_memory_manager(memory_manager.clone());
+        assert!(ram_directory.atomic_write(Path::new("/titi"), b"abcd").is_ok());
+        assert_eq!(memory_manager.total_amount(), 4u64);
+        assert!(ram_directory.atomic_write(Path::new("/toto"), b"abcde").is_ok());
+        assert_eq!(memory_manager.total_amount(), 9u64);
+    }
+
+    #[test]
+    fn test_memory_manager_override() {
+        let memory_manager = ResourceManager::default();
+        let mut ram_directory = RAMDirectory::create_with_memory_manager(memory_manager.clone());
+        assert!(ram_directory.atomic_write(Path::new("/titi"), b"abcde").is_ok());
+        assert_eq!(memory_manager.total_amount(), 5u64);
+        assert!(ram_directory.atomic_write(Path::new("/titi"), b"abcdef").is_ok());
+        assert_eq!(memory_manager.total_amount(), 6u64);
+    }
+
+    #[test]
+    fn test_memory_manager_seq_wrt() {
+        let memory_manager = ResourceManager::default();
+        let mut ram_directory = RAMDirectory::create_with_memory_manager(memory_manager.clone());
+        let mut wrt = ram_directory.open_write(Path::new("/titi")).unwrap();
+        assert!(wrt.write_all(b"abcde").is_ok());
+        assert!(wrt.terminate().is_ok());
+        assert_eq!(memory_manager.total_amount(), 5u64);
+        assert!(ram_directory.atomic_write(Path::new("/titi"), b"abcdef").is_ok());
+        assert_eq!(memory_manager.total_amount(), 6u64);
+    }
+
+    #[test]
+    fn test_release_on_drop() {
+        let memory_manager = ResourceManager::default();
+        let mut ram_directory = RAMDirectory::create_with_memory_manager(memory_manager.clone());
+        let mut wrt = ram_directory.open_write(Path::new("/titi")).unwrap();
+        assert!(wrt.write_all(b"abcde").is_ok());
+        assert!(wrt.terminate().is_ok());
+        assert_eq!(memory_manager.total_amount(), 5u64);
+        let mut wrt2 = ram_directory.open_write(Path::new("/toto")).unwrap();
+        assert!(wrt2.write_all(b"abcdefghijkl").is_ok());
+        assert!(wrt2.terminate().is_ok());
+        assert_eq!(memory_manager.total_amount(), 17u64);
+        let source = ram_directory.open_read(Path::new("/titi")).unwrap();
+        let source_clone = source.clone();
+        assert_eq!(memory_manager.total_amount(), 17u64);
+        mem::drop(ram_directory);
+        assert_eq!(memory_manager.total_amount(), 5u64);
+        mem::drop(source);
+        assert_eq!(memory_manager.total_amount(), 5u64);
+        mem::drop(source_clone);
+        assert_eq!(memory_manager.total_amount(), 0u64);
+    }
 }

src/directory/read_only_source.rs

@@ -2,6 +2,7 @@ use crate::common::HasLen;
 use stable_deref_trait::{CloneStableDeref, StableDeref};
 use std::ops::Deref;
 use std::sync::Arc;
+use crate::indexer::{Allocation, ResourceManager};
 
 pub type BoxedData = Box<dyn Deref<Target = [u8]> + Send + Sync + 'static>;
 
@@ -15,6 +16,7 @@ pub struct ReadOnlySource {
     data: Arc<BoxedData>,
     start: usize,
     stop: usize,
+    allocation: Option<Arc<Allocation>>
 }
 
 unsafe impl StableDeref for ReadOnlySource {}
@@ -35,11 +37,13 @@ impl From<Arc<BoxedData>> for ReadOnlySource {
             data,
             start: 0,
             stop: len,
+            allocation: None
         }
     }
 }
 
 impl ReadOnlySource {
+
     pub(crate) fn new<D>(data: D) -> ReadOnlySource
     where
         D: Deref<Target = [u8]> + Send + Sync + 'static,
@@ -49,9 +53,24 @@ impl ReadOnlySource {
             data: Arc::new(Box::new(data)),
             start: 0,
             stop: len,
+            allocation: None
         }
     }
 
+    pub(crate) fn new_with_allocation<D>(data: D, memory_manager: &ResourceManager) -> ReadOnlySource
+        where
+            D: Deref<Target = [u8]> + Send + Sync + 'static,
+    {
+        let len = data.len();
+        ReadOnlySource {
+            data: Arc::new(Box::new(data)),
+            start: 0,
+            stop: len,
+            allocation: Some(Arc::new(memory_manager.allocate(len as u64)))
+        }
+    }
+
+
     /// Creates an empty ReadOnlySource
     pub fn empty() -> ReadOnlySource {
         ReadOnlySource::new(&[][..])
@@ -98,6 +117,7 @@ impl ReadOnlySource {
             data: self.data.clone(),
             start: self.start + start,
             stop: self.start + stop,
+            allocation: self.allocation.clone()
         }
     }
 

src/directory/spilling_writer.rs

@@ -0,0 +1,186 @@
+use crate::common::MutableEnum;
+use crate::directory::{TerminatingWrite, WritePtr};
+use std::io::{self, Write};
+
+/// Represents the state of the `SpillingWriter`.
+enum SpillingState {
+    Buffer {
+        buffer: Vec<u8>,
+        capacity: usize,
+        write_factory: Box<dyn FnOnce() -> io::Result<WritePtr>>,
+    },
+    Spilled(WritePtr),
+}
+
+impl SpillingState {
+    fn new(
+        limit: usize,
+        write_factory: Box<dyn FnOnce() -> io::Result<WritePtr>>,
+    ) -> SpillingState {
+        SpillingState::Buffer {
+            buffer: Vec::with_capacity(limit),
+            capacity: limit,
+            write_factory,
+        }
+    }
+
+    // Change the state in such a way that it is ready to accept
+    // `extra_capacity` bytes.
+    //
+    fn reserve(self, extra_capacity: usize) -> io::Result<SpillingState> {
+        match self {
+            SpillingState::Buffer {
+                buffer,
+                capacity,
+                write_factory,
+            } => {
+                if capacity >= extra_capacity {
+                    Ok(SpillingState::Buffer {
+                        buffer,
+                        capacity: capacity - extra_capacity,
+                        write_factory,
+                    })
+                } else {
+                    let mut wrt = write_factory()?;
+                    wrt.write_all(&buffer[..])?;
+                    Ok(SpillingState::Spilled(wrt))
+                }
+            }
+            SpillingState::Spilled(wrt) => Ok(SpillingState::Spilled(wrt)),
+        }
+    }
+}
+
+/// The `SpillingWriter` is a writer that start by writing in a
+/// buffer.
+///
+/// Once a memory limit is reached, the spilling writer will
+/// call a given `WritePtr` factory and start spilling into it.
+///
+/// Spilling here includes:
+/// - writing all of the data that were written in the in-memory buffer so far
+/// - writing subsequent data as well.
+///
+/// Once entering "spilling" mode, the `SpillingWriter` stays in this mode.
+pub struct SpillingWriter {
+    state: MutableEnum<SpillingState>,
+}
+
+impl SpillingWriter {
+    //// Creates a new `Spilling Writer`.
+    pub fn new(
+        limit: usize,
+        write_factory: Box<dyn FnOnce() -> io::Result<WritePtr>>,
+    ) -> SpillingWriter {
+        let state = SpillingState::new(limit, write_factory);
+        SpillingWriter {
+            state: MutableEnum::wrap(state),
+        }
+    }
+
+    /// Finalizes the `SpillingWriter`.
+    ///
+    /// The `SpillingResult` object is an enum specific
+    /// to whether the `SpillingWriter` reached the spilling limit
+    /// (In that case, the buffer is returned).
+    ///
+    /// If the writer reached the spilling mode, the underlying `WritePtr`
+    /// is terminated and SpillingResult::Spilled is returned.
+    pub fn finalize(self) -> io::Result<SpillingResult> {
+        match self.state.into() {
+            SpillingState::Spilled(wrt) => {
+                wrt.terminate()?;
+                Ok(SpillingResult::Spilled)
+            }
+            SpillingState::Buffer { buffer, .. } => Ok(SpillingResult::Buffer(buffer)),
+        }
+    }
+}
+
+/// enum used as the result of `.finalize()`.
+pub enum SpillingResult {
+    Spilled,
+    Buffer(Vec<u8>),
+}
+
+impl io::Write for SpillingWriter {
+    fn write(&mut self, payload: &[u8]) -> io::Result<usize> {
+        self.write_all(payload)?;
+        Ok(payload.len())
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        if let SpillingState::Spilled(wrt) = &mut *self.state {
+            wrt.flush()?;
+        }
+        Ok(())
+    }
+
+    fn write_all(&mut self, payload: &[u8]) -> io::Result<()> {
+        self.state.map_mutate(|mut state| {
+            state = state.reserve(payload.len())?;
+            match &mut state {
+                SpillingState::Buffer { buffer, .. } => {
+                    buffer.extend_from_slice(payload);
+                }
+                SpillingState::Spilled(wrt) => {
+                    wrt.write_all(payload)?;
+                }
+            }
+            Ok(state)
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::SpillingWriter;
+    use crate::directory::spilling_writer::SpillingResult;
+    use crate::directory::RAMDirectory;
+    use crate::Directory;
+    use std::io::{self, Write};
+    use std::path::Path;
+
+    #[test]
+    fn test_no_spilling() {
+        let ram_directory = RAMDirectory::create();
+        let mut ram_directory_clone = ram_directory.clone();
+        let path = Path::new("test");
+        let write_factory = Box::new(move || {
+            ram_directory_clone
+                .open_write(path)
+                .map_err(|err| io::Error::new(io::ErrorKind::Other, err))
+        });
+        let mut spilling_wrt = SpillingWriter::new(10, write_factory);
+        assert!(spilling_wrt.write_all(b"abcd").is_ok());
+        if let SpillingResult::Buffer(buf) = spilling_wrt.finalize().unwrap() {
+            assert_eq!(buf, b"abcd")
+        } else {
+            panic!("spill writer should not have spilled");
+        }
+        assert!(!ram_directory.exists(path));
+    }
+
+    #[test]
+    fn test_spilling() {
+        let ram_directory = RAMDirectory::create();
+        let mut ram_directory_clone = ram_directory.clone();
+        let path = Path::new("test");
+        let write_factory = Box::new(move || {
+            ram_directory_clone
+                .open_write(path)
+                .map_err(|err| io::Error::new(io::ErrorKind::Other, err))
+        });
+        let mut spilling_wrt = SpillingWriter::new(10, write_factory);
+        assert!(spilling_wrt.write_all(b"abcd").is_ok());
+        assert!(spilling_wrt.write_all(b"efghijklmnop").is_ok());
+        if let SpillingResult::Spilled = spilling_wrt.finalize().unwrap() {
+        } else {
+            panic!("spill writer should have spilled");
+        }
+        assert_eq!(
+            ram_directory.atomic_read(path).unwrap(),
+            b"abcdefghijklmnop"
+        );
+    }
+}

src/fastfield/delete.rs

@@ -89,7 +89,7 @@ mod tests {
 
     fn test_delete_bitset_helper(bitset: &BitSet, max_doc: u32) {
         let test_path = PathBuf::from("test");
-        let mut directory = RAMDirectory::create();
+        let mut directory = RAMDirectory::default();
         {
             let mut writer = directory.open_write(&*test_path).unwrap();
             write_delete_bitset(bitset, max_doc, &mut writer).unwrap();

src/indexer/index_writer.rs

@@ -8,22 +8,26 @@ use crate::core::SegmentComponent;
 use crate::core::SegmentId;
 use crate::core::SegmentMeta;
 use crate::core::SegmentReader;
-use crate::directory::TerminatingWrite;
 use crate::directory::{DirectoryLock, GarbageCollectionResult};
+use crate::directory::{TerminatingWrite, WatchCallbackList};
 use crate::docset::DocSet;
 use crate::error::TantivyError;
 use crate::fastfield::write_delete_bitset;
 use crate::indexer::delete_queue::{DeleteCursor, DeleteQueue};
 use crate::indexer::doc_opstamp_mapping::DocToOpstampMapping;
 use crate::indexer::operation::DeleteOperation;
+use crate::indexer::segment_manager::SegmentRegisters;
+use crate::indexer::segment_register::SegmentRegister;
 use crate::indexer::stamper::Stamper;
-use crate::indexer::MergePolicy;
-use crate::indexer::SegmentEntry;
+use crate::indexer::{SegmentEntry, ResourceManager};
 use crate::indexer::SegmentWriter;
+use crate::indexer::{IndexWriterConfig, MergePolicy};
+use crate::reader::NRTReader;
 use crate::schema::Document;
 use crate::schema::IndexRecordOption;
 use crate::schema::Term;
-use crate::Opstamp;
+use crate::tokenizer::TokenizerManager;
+use crate::{IndexReader, Opstamp};
 use crossbeam::channel;
 use futures::executor::block_on;
 use futures::future::Future;
@@ -31,18 +35,10 @@ use smallvec::smallvec;
 use smallvec::SmallVec;
 use std::mem;
 use std::ops::Range;
-use std::sync::Arc;
+use std::sync::{Arc, RwLock};
 use std::thread;
 use std::thread::JoinHandle;
 
-// Size of the margin for the heap. A segment is closed when the remaining memory
-// in the heap goes below MARGIN_IN_BYTES.
-pub const MARGIN_IN_BYTES: usize = 1_000_000;
-
-// We impose the memory per thread to be at least 3 MB.
-pub const HEAP_SIZE_MIN: usize = ((MARGIN_IN_BYTES as u32) * 3u32) as usize;
-pub const HEAP_SIZE_MAX: usize = u32::max_value() as usize - MARGIN_IN_BYTES;
-
 // Add document will block if the number of docs waiting in the queue to be indexed
 // reaches `PIPELINE_MAX_SIZE_IN_DOCS`
 const PIPELINE_MAX_SIZE_IN_DOCS: usize = 10_000;
@@ -69,8 +65,9 @@ pub struct IndexWriter {
     _directory_lock: Option<DirectoryLock>,
 
     index: Index,
+    config: IndexWriterConfig,
 
-    heap_size_in_bytes_per_thread: usize,
+    segment_registers: Arc<RwLock<SegmentRegisters>>,
 
     workers_join_handle: Vec<JoinHandle<crate::Result<()>>>,
 
@@ -80,13 +77,14 @@ pub struct IndexWriter {
     segment_updater: SegmentUpdater,
 
     worker_id: usize,
-
-    num_threads: usize,
-
     delete_queue: DeleteQueue,
 
     stamper: Stamper,
     committed_opstamp: Opstamp,
+
+    on_commit: WatchCallbackList,
+
+    memory_manager: ResourceManager,
 }
 
 fn compute_deleted_bitset(
@@ -133,7 +131,6 @@ fn compute_deleted_bitset(
 /// For instance, there was no delete operation between the state of the `segment_entry` and
 /// the `target_opstamp`, `segment_entry` is not updated.
 pub(crate) fn advance_deletes(
-    mut segment: Segment,
     segment_entry: &mut SegmentEntry,
     target_opstamp: Opstamp,
 ) -> crate::Result<()> {
@@ -142,25 +139,33 @@ pub(crate) fn advance_deletes(
         return Ok(());
     }
 
-    if segment_entry.delete_bitset().is_none() && segment_entry.delete_cursor().get().is_none() {
+    let delete_bitset_opt = segment_entry.take_delete_bitset();
+
+    // We avoid directly advancing the `SegmentEntry` delete cursor, because
+    // we do not want to end up in an invalid state if the delete bitset
+    // serialization fails.
+    let mut delete_cursor = segment_entry.delete_cursor();
+
+    if delete_bitset_opt.is_none() && segment_entry.delete_cursor().get().is_none() {
         // There has been no `DeleteOperation` between the segment status and `target_opstamp`.
         return Ok(());
     }
 
+    // We open our current serialized segment to compute the new deleted bitset.
+    let segment = segment_entry.segment().clone();
     let segment_reader = SegmentReader::open(&segment)?;
 
     let max_doc = segment_reader.max_doc();
-    let mut delete_bitset: BitSet = match segment_entry.delete_bitset() {
-        Some(previous_delete_bitset) => (*previous_delete_bitset).clone(),
-        None => BitSet::with_max_value(max_doc),
-    };
+
+    let mut delete_bitset: BitSet =
+        delete_bitset_opt.unwrap_or_else(|| BitSet::with_max_value(max_doc));
 
     let num_deleted_docs_before = segment.meta().num_deleted_docs();
 
     compute_deleted_bitset(
         &mut delete_bitset,
         &segment_reader,
-        segment_entry.delete_cursor(),
+        &mut delete_cursor,
         &DocToOpstampMapping::None,
         target_opstamp,
     )?;
@@ -179,32 +184,40 @@ pub(crate) fn advance_deletes(
     let num_deleted_docs: u32 = delete_bitset.len() as u32;
     if num_deleted_docs > num_deleted_docs_before {
         // There are new deletes. We need to write a new delete file.
-        segment = segment.with_delete_meta(num_deleted_docs as u32, target_opstamp);
-        let mut delete_file = segment.open_write(SegmentComponent::DELETE)?;
+        let mut delete_file = segment
+            .with_delete_meta(num_deleted_docs as u32, target_opstamp)
+            .open_write(SegmentComponent::DELETE)?;
         write_delete_bitset(&delete_bitset, max_doc, &mut delete_file)?;
         delete_file.terminate()?;
+        segment_entry.reset_delete_meta(num_deleted_docs as u32, target_opstamp);
     }
 
-    segment_entry.set_meta(segment.meta().clone());
+    // Regardless of whether we did end up having to write a new file or not
+    // we advance the `delete_cursor`. This is an optimisation. We want to ensure we do not
+    // check that a given deleted term does not match any of our docs more than once.
+    segment_entry.set_delete_cursor(delete_cursor);
     Ok(())
 }
 
 fn index_documents(
-    memory_budget: usize,
+    config: IndexWriterConfig,
     segment: Segment,
     grouped_document_iterator: &mut dyn Iterator<Item = OperationGroup>,
     segment_updater: &mut SegmentUpdater,
+    tokenizers: &TokenizerManager,
     mut delete_cursor: DeleteCursor,
+    memory_manager: ResourceManager
 ) -> crate::Result<bool> {
     let schema = segment.schema();
 
-    let mut segment_writer = SegmentWriter::for_segment(memory_budget, segment.clone(), &schema)?;
+    let mut segment_writer = SegmentWriter::for_segment(
+        &config, segment, &schema, tokenizers, memory_manager)?;
     for document_group in grouped_document_iterator {
         for doc in document_group {
             segment_writer.add_document(doc, &schema)?;
         }
         let mem_usage = segment_writer.mem_usage();
-        if mem_usage >= memory_budget - MARGIN_IN_BYTES {
+        if mem_usage >= config.heap_size_before_flushing() {
             info!(
                 "Buffer limit reached, flushing segment with maxdoc={}.",
                 segment_writer.max_doc()
@@ -223,24 +236,14 @@ fn index_documents(
     // the worker thread.
     assert!(max_doc > 0);
 
-    let doc_opstamps: Vec<Opstamp> = segment_writer.finalize()?;
-
-    let segment_with_max_doc = segment.with_max_doc(max_doc);
+    let (segment, doc_opstamps): (Segment, Vec<Opstamp>) = segment_writer.finalize()?;
 
     let last_docstamp: Opstamp = *(doc_opstamps.last().unwrap());
 
-    let delete_bitset_opt = apply_deletes(
-        &segment_with_max_doc,
-        &mut delete_cursor,
-        &doc_opstamps,
-        last_docstamp,
-    )?;
+    let delete_bitset_opt =
+        apply_deletes(&segment, &mut delete_cursor, &doc_opstamps, last_docstamp)?;
 
-    let segment_entry = SegmentEntry::new(
-        segment_with_max_doc.meta().clone(),
-        delete_cursor,
-        delete_bitset_opt,
-    );
+    let segment_entry = SegmentEntry::new(segment, delete_cursor, delete_bitset_opt);
     block_on(segment_updater.schedule_add_segment(segment_entry))?;
     Ok(true)
 }
@@ -292,21 +295,10 @@ impl IndexWriter {
     /// If the heap size per thread is too small, panics.
     pub(crate) fn new(
         index: &Index,
-        num_threads: usize,
-        heap_size_in_bytes_per_thread: usize,
+        mut config: IndexWriterConfig,
         directory_lock: DirectoryLock,
     ) -> crate::Result<IndexWriter> {
-        if heap_size_in_bytes_per_thread < HEAP_SIZE_MIN {
-            let err_msg = format!(
-                "The heap size per thread needs to be at least {}.",
-                HEAP_SIZE_MIN
-            );
-            return Err(TantivyError::InvalidArgument(err_msg));
-        }
-        if heap_size_in_bytes_per_thread >= HEAP_SIZE_MAX {
-            let err_msg = format!("The heap size per thread cannot exceed {}", HEAP_SIZE_MAX);
-            return Err(TantivyError::InvalidArgument(err_msg));
-        }
+        config.validate()?;
         let (document_sender, document_receiver): (OperationSender, OperationReceiver) =
             channel::bounded(PIPELINE_MAX_SIZE_IN_DOCS);
 
@@ -314,16 +306,26 @@ impl IndexWriter {
 
         let current_opstamp = index.load_metas()?.opstamp;
 
+        let meta = index.load_metas()?;
+
         let stamper = Stamper::new(current_opstamp);
 
+        let commited_segments = SegmentRegister::new(
+            index.directory(),
+            &index.schema(),
+            meta.segments,
+            &delete_queue.cursor(),
+        );
+        let segment_registers = Arc::new(RwLock::new(SegmentRegisters::new(commited_segments)));
+
         let segment_updater =
-            SegmentUpdater::create(index.clone(), stamper.clone(), &delete_queue.cursor())?;
+            SegmentUpdater::create(segment_registers.clone(), index.clone(), stamper.clone())?;
 
         let mut index_writer = IndexWriter {
             _directory_lock: Some(directory_lock),
 
-            heap_size_in_bytes_per_thread,
             index: index.clone(),
+            config,
 
             operation_receiver: document_receiver,
             operation_sender: document_sender,
@@ -331,7 +333,6 @@ impl IndexWriter {
             segment_updater,
 
             workers_join_handle: vec![],
-            num_threads,
 
             delete_queue,
 
@@ -339,6 +340,10 @@ impl IndexWriter {
             stamper,
 
             worker_id: 0,
+            segment_registers,
+            on_commit: Default::default(),
+
+            memory_manager: Default::default()
         };
         index_writer.start_workers()?;
         Ok(index_writer)
@@ -366,23 +371,11 @@ impl IndexWriter {
                 })?;
         }
 
-        let result = self
+        self
             .segment_updater
-            .wait_merging_thread()
-            .map_err(|_| TantivyError::ErrorInThread("Failed to join merging thread.".into()));
+            .wait_merging_thread();
 
-        if let Err(ref e) = result {
-            error!("Some merging thread failed {:?}", e);
-        }
-
-        result
-    }
-
-    #[doc(hidden)]
-    pub fn add_segment(&self, segment_meta: SegmentMeta) -> crate::Result<()> {
-        let delete_cursor = self.delete_queue.cursor();
-        let segment_entry = SegmentEntry::new(segment_meta, delete_cursor, None);
-        block_on(self.segment_updater.schedule_add_segment(segment_entry))
+        Ok(())
     }
 
     /// Creates a new segment.
@@ -405,8 +398,9 @@ impl IndexWriter {
 
         let mut delete_cursor = self.delete_queue.cursor();
 
-        let mem_budget = self.heap_size_in_bytes_per_thread;
         let index = self.index.clone();
+        let config = self.config.clone();
+        let memory_manager = self.memory_manager.clone();
         let join_handle: JoinHandle<crate::Result<()>> = thread::Builder::new()
             .name(format!("thrd-tantivy-index{}", self.worker_id))
             .spawn(move || {
@@ -435,11 +429,13 @@ impl IndexWriter {
                     }
                     let segment = index.new_segment();
                     index_documents(
-                        mem_budget,
+                        config.clone(),
                         segment,
                         &mut document_iterator,
                         &mut segment_updater,
+                        index.tokenizers(),
                         delete_cursor.clone(),
+                        memory_manager.clone()
                     )?;
                 }
             })?;
@@ -459,7 +455,7 @@ impl IndexWriter {
     }
 
     fn start_workers(&mut self) -> crate::Result<()> {
-        for _ in 0..self.num_threads {
+        for _ in 0..self.config.max_indexing_threads {
             self.add_indexing_worker()?;
         }
         Ok(())
@@ -564,12 +560,8 @@ impl IndexWriter {
             .take()
             .expect("The IndexWriter does not have any lock. This is a bug, please report.");
 
-        let new_index_writer: IndexWriter = IndexWriter::new(
-            &self.index,
-            self.num_threads,
-            self.heap_size_in_bytes_per_thread,
-            directory_lock,
-        )?;
+        let new_index_writer: IndexWriter =
+            IndexWriter::new(&self.index, self.config.clone(), directory_lock)?;
 
         // the current `self` is dropped right away because of this call.
         //
@@ -608,7 +600,7 @@ impl IndexWriter {
     /// It is also possible to add a payload to the `commit`
     /// using this API.
     /// See [`PreparedCommit::set_payload()`](PreparedCommit.html)
-    pub fn prepare_commit(&mut self) -> crate::Result<PreparedCommit> {
+    pub fn prepare_commit(&mut self, soft_commit: bool) -> crate::Result<PreparedCommit> {
         // Here, because we join all of the worker threads,
         // all of the segment update for this commit have been
         // sent.
@@ -636,7 +628,7 @@ impl IndexWriter {
         }
 
         let commit_opstamp = self.stamper.stamp();
-        let prepared_commit = PreparedCommit::new(self, commit_opstamp);
+        let prepared_commit = PreparedCommit::new(self, commit_opstamp, soft_commit);
         info!("Prepared commit {}", commit_opstamp);
         Ok(prepared_commit)
     }
@@ -656,7 +648,25 @@ impl IndexWriter {
     /// that made it in the commit.
     ///
     pub fn commit(&mut self) -> crate::Result<Opstamp> {
-        self.prepare_commit()?.commit()
+        self.prepare_commit(false)?.commit()
+    }
+
+    pub fn soft_commit(&mut self) -> crate::Result<Opstamp> {
+        self.prepare_commit(true)?.commit()
+    }
+
+    pub(crate) fn trigger_commit(&self) -> impl Future<Output = ()> {
+        self.on_commit.broadcast()
+    }
+
+    pub fn reader(&self, num_searchers: usize) -> crate::Result<IndexReader> {
+        let nrt_reader = NRTReader::create(
+            num_searchers,
+            self.index.clone(),
+            self.segment_registers.clone(),
+            &self.on_commit,
+        )?;
+        Ok(IndexReader::NRT(nrt_reader))
     }
 
     pub(crate) fn segment_updater(&self) -> &SegmentUpdater {
@@ -1054,7 +1064,8 @@ mod tests {
                 index_writer.add_document(doc!(text_field => "a"));
             }
             {
-                let mut prepared_commit = index_writer.prepare_commit().expect("commit failed");
+                let mut prepared_commit =
+                    index_writer.prepare_commit(false).expect("commit failed");
                 prepared_commit.set_payload("first commit");
                 prepared_commit.commit().expect("commit failed");
             }
@@ -1087,7 +1098,8 @@ mod tests {
                 index_writer.add_document(doc!(text_field => "a"));
             }
             {
-                let mut prepared_commit = index_writer.prepare_commit().expect("commit failed");
+                let mut prepared_commit =
+                    index_writer.prepare_commit(false).expect("commit failed");
                 prepared_commit.set_payload("first commit");
                 prepared_commit.abort().expect("commit failed");
             }
@@ -1265,4 +1277,41 @@ mod tests {
         let commit = index_writer.commit();
         assert!(commit.is_ok());
     }
+
+    #[test]
+    fn test_index_writer_reader() {
+        let mut schema_builder = schema::Schema::builder();
+        let idfield = schema_builder.add_text_field("id", STRING);
+        schema_builder.add_text_field("optfield", STRING);
+        let index = Index::create_in_ram(schema_builder.build());
+        let mut index_writer = index.writer_with_num_threads(1, 3_000_000).unwrap();
+        index_writer.add_document(doc!(idfield=>"myid"));
+        assert!(index_writer.commit().is_ok());
+        let reader = index_writer.reader(2).unwrap();
+        let searcher = reader.searcher();
+        assert_eq!(searcher.num_docs(), 1u64);
+        index_writer.add_document(doc!(idfield=>"myid"));
+        assert!(index_writer.commit().is_ok());
+        assert_eq!(reader.searcher().num_docs(), 2u64);
+        assert_eq!(searcher.num_docs(), 1u64);
+    }
+
+    #[test]
+    fn test_index_writer_reader_soft_commit() {
+        let mut schema_builder = schema::Schema::builder();
+        let idfield = schema_builder.add_text_field("id", STRING);
+        schema_builder.add_text_field("optfield", STRING);
+        let index = Index::create_in_ram(schema_builder.build());
+        let mut index_writer = index.writer_with_num_threads(1, 3_000_000).unwrap();
+        index_writer.add_document(doc!(idfield=>"myid"));
+        assert!(index_writer.soft_commit().is_ok());
+        let nrt_reader = index_writer.reader(2).unwrap();
+        let normal_reader = index.reader_builder().try_into().unwrap();
+        assert_eq!(nrt_reader.searcher().num_docs(), 1u64);
+        assert_eq!(normal_reader.searcher().num_docs(), 0u64);
+        assert!(index_writer.commit().is_ok());
+        assert!(normal_reader.reload().is_ok());
+        assert_eq!(nrt_reader.searcher().num_docs(), 1u64);
+        assert_eq!(normal_reader.searcher().num_docs(), 1u64);
+    }
 }

src/indexer/index_writer_config.rs

@@ -0,0 +1,133 @@
+use serde::{Deserialize, Serialize};
+
+// Size of the margin for the heap. A segment is closed when the remaining memory
+// in the heap goes below MARGIN_IN_BYTES.
+const MARGIN_IN_BYTES: u64 = 1_000_000;
+
+// We impose the memory per thread to be at least 3 MB.
+const HEAP_SIZE_MIN: u64 = MARGIN_IN_BYTES * 3u64;
+const HEAP_SIZE_MAX: u64 = u32::max_value() as u64 - MARGIN_IN_BYTES;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct IndexWriterConfig {
+    pub max_indexing_threads: usize,
+    pub max_merging_threads: usize,
+    pub memory_budget: u64,
+    pub store_flush_num_bytes: u64,
+    pub persist_low: u64,
+    pub persist_high: u64,
+}
+
+impl Default for IndexWriterConfig {
+    fn default() -> Self {
+        IndexWriterConfig {
+            max_indexing_threads: 1,
+            max_merging_threads: 3,
+            memory_budget: 50_000_000u64,
+            store_flush_num_bytes: 10_000_000u64,
+            persist_low: 10_000_000u64,
+            persist_high: 50_000_000u64
+        }
+    }
+}
+
+impl IndexWriterConfig {
+    #[cfg(test)]
+    pub fn for_test() -> IndexWriterConfig {
+        IndexWriterConfig {
+            max_indexing_threads: 1,
+            max_merging_threads: 5,
+            memory_budget: 4_000_000u64,
+            store_flush_num_bytes: 500_000u64,
+            persist_low: 2_000_000u64,
+            persist_high: 3_000_000u64,
+        }
+    }
+
+    // Ensures the `IndexWriterConfig` is correct.
+    //
+    // This method checks that the values in the `IndexWriterConfig`
+    // are valid. If it is not, it may mutate some of the values (like `max_num_threads`) to
+    // fit the contracts or return an error with an explicit error message.
+    //
+    // If called twice, the config is guaranteed to not be updated the second time.
+    pub fn validate(&mut self) -> crate::Result<()> {
+        if self.memory_budget < HEAP_SIZE_MIN {
+            let err_msg = format!(
+                "The heap size per thread needs to be at least {}.",
+                HEAP_SIZE_MIN
+            );
+            return Err(crate::TantivyError::InvalidArgument(err_msg));
+        }
+        let heap_size_in_bytes_per_thread = self.heap_size_in_byte_per_thread();
+        if heap_size_in_bytes_per_thread >= HEAP_SIZE_MAX {
+            let err_msg = format!("The heap size per thread cannot exceed {}", HEAP_SIZE_MAX);
+            return Err(crate::TantivyError::InvalidArgument(err_msg));
+        }
+        if heap_size_in_bytes_per_thread < HEAP_SIZE_MIN {
+            self.max_indexing_threads = (self.memory_budget / HEAP_SIZE_MIN) as usize;
+        }
+        Ok(())
+    }
+
+    pub fn heap_size_in_byte_per_thread(&self) -> u64 {
+        self.memory_budget / self.max_indexing_threads as u64
+    }
+
+    pub fn heap_size_before_flushing(&self) -> u64 {
+        self.heap_size_in_byte_per_thread() - MARGIN_IN_BYTES
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::IndexWriterConfig;
+
+    #[test]
+    fn test_index_writer_config_simple() {
+        let mut index = IndexWriterConfig {
+            max_indexing_threads: 3,
+            memory_budget: super::HEAP_SIZE_MIN * 3,
+            ..Default::default()
+        };
+        assert!(index.validate().is_ok());
+        assert_eq!(index.max_indexing_threads, 3);
+        assert_eq!(index.heap_size_in_byte_per_thread(), super::HEAP_SIZE_MIN);
+    }
+
+    #[test]
+    fn test_index_writer_config_reduce_num_threads() {
+        let mut index = IndexWriterConfig {
+            max_indexing_threads: 3,
+            memory_budget: super::HEAP_SIZE_MIN,
+            ..Default::default()
+        };
+        assert!(index.validate().is_ok());
+        assert_eq!(index.max_indexing_threads, 1);
+        assert_eq!(index.heap_size_in_byte_per_thread(), super::HEAP_SIZE_MIN);
+    }
+
+    #[test]
+    fn test_index_writer_config_not_enough_memory() {
+        let mut index = IndexWriterConfig {
+            max_indexing_threads: 1,
+            memory_budget: super::HEAP_SIZE_MIN - 1,
+            ..Default::default()
+        };
+        assert!(
+            matches!(index.validate(), Err(crate::TantivyError::InvalidArgument(msg) ) if msg.contains("The heap size per thread needs to be at least"))
+        );
+    }
+
+    #[test]
+    fn test_index_writer_config_too_much_memory() {
+        let mut index = IndexWriterConfig {
+            max_indexing_threads: 1,
+            memory_budget: (u32::max_value() as u64) + 1,
+            ..Default::default()
+        };
+        assert!(
+            matches!(index.validate(), Err(crate::TantivyError::InvalidArgument(msg) ) if msg.contains("The heap size per thread cannot exceed"))
+        );
+    }
+}

src/indexer/merge_operation.rs

@@ -1,17 +1,22 @@
+use crate::indexer::resource_manager::{Allocation, ResourceManager};
 use crate::Opstamp;
 use crate::SegmentId;
 use census::{Inventory, TrackedObject};
 use std::collections::HashSet;
+use std::fmt;
 use std::ops::Deref;
 
-#[derive(Default)]
-pub(crate) struct MergeOperationInventory(Inventory<InnerMergeOperation>);
+#[derive(Default, Clone)]
+pub(crate) struct MergeOperationInventory {
+    inventory: Inventory<InnerMergeOperation>,
+    num_merge_watcher: ResourceManager,
+}
 
 impl Deref for MergeOperationInventory {
     type Target = Inventory<InnerMergeOperation>;
 
     fn deref(&self) -> &Self::Target {
-        &self.0
+        &self.inventory
     }
 }
 
@@ -25,6 +30,10 @@ impl MergeOperationInventory {
         }
         segment_in_merge
     }
+
+    pub fn wait_until_empty(&self) {
+        let _ = self.num_merge_watcher.wait_until_in_range(0..1);
+    }
 }
 
 /// A `MergeOperation` has two roles.
@@ -47,6 +56,17 @@ pub struct MergeOperation {
 pub(crate) struct InnerMergeOperation {
     target_opstamp: Opstamp,
     segment_ids: Vec<SegmentId>,
+    _allocation: Allocation,
+}
+
+impl fmt::Debug for InnerMergeOperation {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(
+            f,
+            "MergeOp(target_opstamp={:?}, segment_ids={:?})",
+            self.target_opstamp, self.segment_ids
+        )
+    }
 }
 
 impl MergeOperation {
@@ -55,9 +75,11 @@ impl MergeOperation {
         target_opstamp: Opstamp,
         segment_ids: Vec<SegmentId>,
     ) -> MergeOperation {
+        let allocation = inventory.num_merge_watcher.allocate(1);
         let inner_merge_operation = InnerMergeOperation {
             target_opstamp,
             segment_ids,
+            _allocation: allocation,
         };
         MergeOperation {
             inner: inventory.track(inner_merge_operation),

src/indexer/merger.rs

@@ -2,6 +2,8 @@ use crate::common::MAX_DOC_LIMIT;
 use crate::core::Segment;
 use crate::core::SegmentReader;
 use crate::core::SerializableSegment;
+use crate::directory::TerminatingWrite;
+use crate::directory::WritePtr;
 use crate::docset::DocSet;
 use crate::fastfield::BytesFastFieldReader;
 use crate::fastfield::DeleteBitSet;
@@ -660,7 +662,8 @@ impl IndexMerger {
         Ok(term_ordinal_mappings)
     }
 
-    fn write_storable_fields(&self, store_writer: &mut StoreWriter) -> crate::Result<()> {
+    pub fn write_storable_fields(&self, store_wrt: WritePtr) -> crate::Result<()> {
+        let mut store_writer = StoreWriter::new(store_wrt);
         for reader in &self.readers {
             let store_reader = reader.get_store_reader();
             if reader.num_deleted_docs() > 0 {
@@ -672,6 +675,8 @@ impl IndexMerger {
                 store_writer.stack(&store_reader)?;
             }
         }
+        let store_wrt = store_writer.close()?;
+        store_wrt.terminate()?;
         Ok(())
     }
 }
@@ -681,7 +686,6 @@ impl SerializableSegment for IndexMerger {
         let term_ord_mappings = self.write_postings(serializer.get_postings_serializer())?;
         self.write_fieldnorms(serializer.get_fieldnorms_serializer())?;
         self.write_fast_fields(serializer.get_fast_field_serializer(), term_ord_mappings)?;
-        self.write_storable_fields(serializer.get_store_writer())?;
         serializer.close()?;
         Ok(self.max_doc)
     }

src/indexer/mod.rs

@@ -2,12 +2,14 @@ pub mod delete_queue;
 
 mod doc_opstamp_mapping;
 pub mod index_writer;
+mod index_writer_config;
 mod log_merge_policy;
 mod merge_operation;
 pub mod merge_policy;
 pub mod merger;
 pub mod operation;
 mod prepared_commit;
+mod resource_manager;
 mod segment_entry;
 mod segment_manager;
 mod segment_register;
@@ -16,13 +18,17 @@ pub mod segment_updater;
 mod segment_writer;
 mod stamper;
 
+pub(crate) use self::resource_manager::{Allocation, ResourceManager};
+pub(crate) use self::merge_operation::MergeOperationInventory;
 pub use self::index_writer::IndexWriter;
+pub use self::index_writer_config::IndexWriterConfig;
 pub use self::log_merge_policy::LogMergePolicy;
 pub use self::merge_operation::MergeOperation;
 pub use self::merge_policy::{MergeCandidate, MergePolicy, NoMergePolicy};
 pub use self::prepared_commit::PreparedCommit;
 pub use self::segment_entry::SegmentEntry;
 pub use self::segment_manager::SegmentManager;
+pub(crate) use self::segment_manager::SegmentRegisters;
 pub use self::segment_serializer::SegmentSerializer;
 pub use self::segment_writer::SegmentWriter;
 

src/indexer/prepared_commit.rs

@@ -7,14 +7,20 @@ pub struct PreparedCommit<'a> {
     index_writer: &'a mut IndexWriter,
     payload: Option<String>,
     opstamp: Opstamp,
+    soft_commit: bool,
 }
 
 impl<'a> PreparedCommit<'a> {
-    pub(crate) fn new(index_writer: &'a mut IndexWriter, opstamp: Opstamp) -> PreparedCommit<'_> {
+    pub(crate) fn new(
+        index_writer: &'a mut IndexWriter,
+        opstamp: Opstamp,
+        soft_commit: bool,
+    ) -> PreparedCommit<'_> {
         PreparedCommit {
             index_writer,
             payload: None,
             opstamp,
+            soft_commit,
         }
     }
 
@@ -32,11 +38,12 @@ impl<'a> PreparedCommit<'a> {
 
     pub fn commit(self) -> crate::Result<Opstamp> {
         info!("committing {}", self.opstamp);
-        let _ = block_on(
-            self.index_writer
-                .segment_updater()
-                .schedule_commit(self.opstamp, self.payload),
-        );
+        block_on(self.index_writer.segment_updater().schedule_commit(
+            self.opstamp,
+            self.payload,
+            self.soft_commit,
+        ))?;
+        block_on(self.index_writer.trigger_commit());
         Ok(self.opstamp)
     }
 }

src/indexer/resource_manager.rs

@@ -0,0 +1,213 @@
+use std::ops::RangeBounds;
+use std::sync::{Arc, Condvar, Mutex, MutexGuard, RwLock};
+
+struct LockedData {
+    count: u64,
+    enabled: bool
+}
+
+impl Default for LockedData {
+    fn default() -> Self {
+        LockedData {
+            count: 0u64,
+            enabled: true
+        }
+    }
+}
+
+#[derive(Default)]
+struct Inner {
+    resource_level: Mutex<LockedData>,
+    convdvar: Condvar,
+}
+
+
+/// The resource manager makes it possible to track the amount of level of a given resource.
+/// There is no magic here : it is to the description of the user to declare how much
+/// of the resource is being held.
+///
+/// Allocation of a resource is bound to the lifetime of a `Allocation` instance.
+///
+/// ```rust
+/// let resource_manager = ResourceManager::default();
+///
+/// ```
+///
+/// In tantivy, this is used to check the number of merging thread and the number of memory
+/// used by the volatile segments.
+///
+#[derive(Clone, Default)]
+pub struct ResourceManager {
+    inner: Arc<Inner>,
+}
+
+impl ResourceManager {
+    /// Return the total amount of reousrce allocated
+    pub fn total_amount(&self) -> u64 {
+        self.lock().count
+    }
+
+    fn lock(&self) -> MutexGuard<LockedData> {
+        self.inner
+            .resource_level
+            .lock()
+            .expect("Failed to obtain lock for ReservedMemory. This should never happen.")
+    }
+
+    fn record_delta(&self, delta: i64) {
+        if delta == 0i64 {
+            return;
+        }
+        let mut lock = self.lock();
+        let new_val = lock.count as i64 + delta;
+        lock.count = new_val as u64;
+        self.inner.convdvar.notify_all();
+    }
+
+    /// Records a new allocation.
+    ///
+    /// The returned allocate object is used to automatically release the allocated resource
+    /// on drop.
+    pub fn allocate(&self, amount: u64) -> Allocation {
+        self.record_delta(amount as i64);
+        Allocation {
+            resource_manager: self.clone(),
+            amount: RwLock::new(amount),
+        }
+    }
+
+    /// Stops the resource manager.
+    ///
+    /// If any thread is waiting via `.wait_until_in_range(...)`, the method will stop
+    /// being blocking and will return an error.
+    pub fn terminate(&self) {
+        self.lock().enabled = false;
+        self.inner.convdvar.notify_all();
+    }
+
+    /// Blocks the current thread until the resource level reaches the given range,
+    /// in a cpu-efficient way.
+    ///
+    /// This method does not necessarily wakes up the current thread at every transition
+    /// into the targetted range, but any durable entry in the range will be detected.
+    pub fn wait_until_in_range<R: RangeBounds<u64>>(&self, range: R) -> Result<u64, u64> {
+        let mut levels = self.lock();
+        if !levels.enabled {
+            return Err(levels.count)
+        }
+        while !range.contains(&levels.count) {
+           levels = self.inner.convdvar.wait(levels).unwrap();
+            if !levels.enabled {
+                return Err(levels.count)
+            }
+        }
+        Ok(levels.count)
+    }
+}
+
+pub struct Allocation {
+    resource_manager: ResourceManager,
+    amount: RwLock<u64>,
+}
+
+impl Allocation {
+    pub fn amount(&self) -> u64 {
+        *self.amount.read().unwrap()
+    }
+
+    pub fn modify(&self, new_amount: u64) {
+        let mut wlock = self.amount.write().unwrap();
+        let delta = new_amount as i64 - *wlock as i64;
+        *wlock = new_amount;
+        self.resource_manager.record_delta(delta);
+    }
+}
+
+impl Drop for Allocation {
+    fn drop(&mut self) {
+        let amount = self.amount();
+        self.resource_manager.record_delta(-(amount as i64))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::ResourceManager;
+    use futures::channel::oneshot;
+    use futures::executor::block_on;
+    use std::{mem, thread};
+
+    #[test]
+    fn test_simple_allocation() {
+        let memory = ResourceManager::default();
+        assert_eq!(memory.total_amount(), 0u64);
+        let _allocation = memory.allocate(10u64);
+        assert_eq!(memory.total_amount(), 10u64);
+    }
+
+    #[test]
+    fn test_multiple_allocation() {
+        let memory = ResourceManager::default();
+        assert_eq!(memory.total_amount(), 0u64);
+        let _allocation = memory.allocate(10u64);
+        let _allocation_2 = memory.allocate(11u64);
+        assert_eq!(memory.total_amount(), 21u64);
+    }
+
+    #[test]
+    fn test_release_on_drop() {
+        let memory = ResourceManager::default();
+        assert_eq!(memory.total_amount(), 0u64);
+        let allocation = memory.allocate(10u64);
+        let allocation_2 = memory.allocate(11u64);
+        assert_eq!(memory.total_amount(), 21u64);
+        mem::drop(allocation);
+        assert_eq!(memory.total_amount(), 11u64);
+        mem::drop(allocation_2);
+        assert_eq!(memory.total_amount(), 0u64);
+    }
+
+    #[test]
+    fn test_wait_until() {
+        let memory = ResourceManager::default();
+        let (send, recv) = oneshot::channel::<()>();
+        let memory_clone = memory.clone();
+        thread::spawn(move || {
+            let _allocation1 = memory_clone.allocate(2u64);
+            let _allocation2 = memory_clone.allocate(3u64);
+            let _allocation3 = memory_clone.allocate(4u64);
+            std::mem::drop(_allocation3);
+            assert!(block_on(recv).is_ok());
+        });
+        assert_eq!(memory.wait_until_in_range(5u64..8u64), Ok(5u64));
+        assert!(send.send(()).is_ok());
+    }
+
+    #[test]
+    fn test_modify_amount() {
+        let memory = ResourceManager::default();
+        let alloc = memory.allocate(2u64);
+        assert_eq!(memory.total_amount(), 2u64);
+        assert_eq!(alloc.amount(), 2u64);
+        let alloc2 = memory.allocate(3u64);
+        assert_eq!(memory.total_amount(), 2u64 + 3u64);
+        assert_eq!(alloc2.amount(), 3u64);
+        alloc.modify(14u64);
+        assert_eq!(alloc.amount(), 14u64);
+        assert_eq!(memory.total_amount(), 14u64 + 3u64)
+    }
+    
+    #[test]
+    fn test_stop_resource_manager() {
+        let resource_manager = ResourceManager::default();
+        let resource_manager_clone = resource_manager.clone();
+        let (sender, recv) = oneshot::channel();
+        let join_handle = thread::spawn(move || {
+            assert!(sender.send(()).is_ok());
+            resource_manager_clone.wait_until_in_range(10..20)
+        });
+        let _ = block_on(recv);
+        resource_manager.terminate();
+        assert_eq!(join_handle.join().unwrap(), Err(0u64));
+    }
+}

src/indexer/segment_entry.rs

@@ -1,7 +1,9 @@
 use crate::common::BitSet;
 use crate::core::SegmentId;
 use crate::core::SegmentMeta;
+use crate::directory::ManagedDirectory;
 use crate::indexer::delete_queue::DeleteCursor;
+use crate::{Opstamp, Segment};
 use std::fmt;
 
 /// A segment entry describes the state of
@@ -19,7 +21,7 @@ use std::fmt;
 /// in the .del file or in the `delete_bitset`.
 #[derive(Clone)]
 pub struct SegmentEntry {
-    meta: SegmentMeta,
+    segment: Segment,
     delete_bitset: Option<BitSet>,
     delete_cursor: DeleteCursor,
 }
@@ -27,47 +29,67 @@ pub struct SegmentEntry {
 impl SegmentEntry {
     /// Create a new `SegmentEntry`
     pub fn new(
-        segment_meta: SegmentMeta,
+        segment: Segment,
         delete_cursor: DeleteCursor,
         delete_bitset: Option<BitSet>,
     ) -> SegmentEntry {
         SegmentEntry {
-            meta: segment_meta,
+            segment,
             delete_bitset,
             delete_cursor,
         }
     }
 
-    /// Return a reference to the segment entry deleted bitset.
-    ///
-    /// `DocId` in this bitset are flagged as deleted.
-    pub fn delete_bitset(&self) -> Option<&BitSet> {
-        self.delete_bitset.as_ref()
+    pub fn persist(&mut self, dest_directory: ManagedDirectory) -> crate::Result<()> {
+        // TODO take in account delete bitset?
+        self.segment.persist(dest_directory)?;
+        Ok(())
     }
 
-    /// Set the `SegmentMeta` for this segment.
-    pub fn set_meta(&mut self, segment_meta: SegmentMeta) {
-        self.meta = segment_meta;
+    pub fn set_delete_cursor(&mut self, delete_cursor: DeleteCursor) {
+        self.delete_cursor = delete_cursor;
+    }
+
+    /// `Takes` (as in Option::take) the delete bitset of a segment entry.
+    /// `DocId` in this bitset are flagged as deleted.
+    pub fn take_delete_bitset(&mut self) -> Option<BitSet> {
+        self.delete_bitset.take()
+    }
+
+    /// Reset the delete information in this segment.
+    ///
+    /// The `SegmentEntry` segment's `SegmentMeta` gets updated, and
+    /// any delete bitset is drop and set to None.
+    pub fn reset_delete_meta(&mut self, num_deleted_docs: u32, target_opstamp: Opstamp) {
+        self.segment = self
+            .segment
+            .clone()
+            .with_delete_meta(num_deleted_docs, target_opstamp);
+        self.delete_bitset = None;
     }
 
     /// Return a reference to the segment_entry's delete cursor
-    pub fn delete_cursor(&mut self) -> &mut DeleteCursor {
-        &mut self.delete_cursor
+    pub fn delete_cursor(&mut self) -> DeleteCursor {
+        self.delete_cursor.clone()
     }
 
     /// Returns the segment id.
     pub fn segment_id(&self) -> SegmentId {
-        self.meta.id()
+        self.meta().id()
     }
 
+    /// Returns the `segment` associated to the `SegmentEntry`.
+    pub fn segment(&self) -> &Segment {
+        &self.segment
+    }
     /// Accessor to the `SegmentMeta`
     pub fn meta(&self) -> &SegmentMeta {
-        &self.meta
+        self.segment.meta()
     }
 }
 
 impl fmt::Debug for SegmentEntry {
     fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(formatter, "SegmentEntry({:?})", self.meta)
+        write!(formatter, "SegmentEntry({:?})", self.meta())
     }
 }

src/indexer/segment_manager.rs

@@ -2,15 +2,15 @@ use super::segment_register::SegmentRegister;
 use crate::core::SegmentId;
 use crate::core::SegmentMeta;
 use crate::error::TantivyError;
-use crate::indexer::delete_queue::DeleteCursor;
-use crate::indexer::SegmentEntry;
+use crate::indexer::{SegmentEntry, MergeOperationInventory, MergeCandidate, MergeOperation};
+use crate::{Segment, Opstamp};
 use std::collections::hash_set::HashSet;
 use std::fmt::{self, Debug, Formatter};
-use std::sync::RwLock;
+use std::sync::{Arc, RwLock};
 use std::sync::{RwLockReadGuard, RwLockWriteGuard};
 
 #[derive(Default)]
-struct SegmentRegisters {
+pub(crate) struct SegmentRegisters {
     uncommitted: SegmentRegister,
     committed: SegmentRegister,
 }
@@ -22,6 +22,17 @@ pub(crate) enum SegmentsStatus {
 }
 
 impl SegmentRegisters {
+    pub fn new(committed: SegmentRegister) -> SegmentRegisters {
+        SegmentRegisters {
+            uncommitted: Default::default(),
+            committed,
+        }
+    }
+
+    pub fn committed_segment(&self) -> Vec<Segment> {
+        self.committed.segments()
+    }
+
     /// Check if all the segments are committed or uncommited.
     ///
     /// If some segment is missing or segments are in a different state (this should not happen
@@ -44,7 +55,8 @@ impl SegmentRegisters {
 /// changes (merges especially)
 #[derive(Default)]
 pub struct SegmentManager {
-    registers: RwLock<SegmentRegisters>,
+    registers: Arc<RwLock<SegmentRegisters>>,
+    merge_operations: MergeOperationInventory,
 }
 
 impl Debug for SegmentManager {
@@ -58,34 +70,28 @@ impl Debug for SegmentManager {
     }
 }
 
-pub fn get_mergeable_segments(
-    in_merge_segment_ids: &HashSet<SegmentId>,
-    segment_manager: &SegmentManager,
-) -> (Vec<SegmentMeta>, Vec<SegmentMeta>) {
-    let registers_lock = segment_manager.read();
-    (
-        registers_lock
-            .committed
-            .get_mergeable_segments(in_merge_segment_ids),
-        registers_lock
-            .uncommitted
-            .get_mergeable_segments(in_merge_segment_ids),
-    )
-}
 
 impl SegmentManager {
-    pub fn from_segments(
-        segment_metas: Vec<SegmentMeta>,
-        delete_cursor: &DeleteCursor,
-    ) -> SegmentManager {
+    pub(crate) fn new(registers: Arc<RwLock<SegmentRegisters>>) -> SegmentManager {
         SegmentManager {
-            registers: RwLock::new(SegmentRegisters {
-                uncommitted: SegmentRegister::default(),
-                committed: SegmentRegister::new(segment_metas, delete_cursor),
-            }),
+            registers,
+            merge_operations: Default::default()
         }
     }
 
+    pub fn new_merge_operation(&self, opstamp: Opstamp, merge_candidate: MergeCandidate) -> MergeOperation {
+        MergeOperation::new(
+            &self.merge_operations,
+            opstamp,
+            merge_candidate.0
+
+        )
+    }
+
+    pub fn wait_merging_thread(&self) {
+        self.merge_operations.wait_until_empty()
+    }
+
     /// Returns all of the segment entries (committed or uncommitted)
     pub fn segment_entries(&self) -> Vec<SegmentEntry> {
         let registers_lock = self.read();
@@ -94,6 +100,34 @@ impl SegmentManager {
         segment_entries
     }
 
+    /// Returns the segments that are currently not in merge.
+    ///
+    /// They are split over two `Vec`. The committed segments on one hand
+    /// and the uncommitted ones on the other hand.
+    ///
+    /// This method is useful when searching for merge candidate or segments
+    /// to persists.
+    pub fn segments_not_in_merge(&self) -> (Vec<SegmentMeta>, Vec<SegmentMeta>) {
+        let in_merge_segment_ids: HashSet<SegmentId> = self.merge_operations.segment_in_merge();
+        let registers_lock = self.read();
+        (
+            registers_lock
+                .committed
+                .get_mergeable_segments(&in_merge_segment_ids),
+            registers_lock
+                .uncommitted
+                .get_mergeable_segments(&in_merge_segment_ids),
+        )
+    }
+
+    pub fn largest_segment_not_in_merge(&self) -> Option<SegmentMeta> {
+        let (committed, uncommitted) = self.segments_not_in_merge();
+        let mut segments = vec![];
+        segments.extend(committed);
+        segments.extend(uncommitted);
+        None
+    }
+
     // Lock poisoning should never happen :
     // The lock is acquired and released within this class,
     // and the operations cannot panic.

src/indexer/segment_register.rs

@@ -1,7 +1,10 @@
 use crate::core::SegmentId;
 use crate::core::SegmentMeta;
+use crate::directory::ManagedDirectory;
 use crate::indexer::delete_queue::DeleteCursor;
 use crate::indexer::segment_entry::SegmentEntry;
+use crate::schema::Schema;
+use crate::Segment;
 use std::collections::HashMap;
 use std::collections::HashSet;
 use std::fmt::{self, Debug, Formatter};
@@ -46,6 +49,13 @@ impl SegmentRegister {
             .collect()
     }
 
+    pub fn segments(&self) -> Vec<Segment> {
+        self.segment_states
+            .values()
+            .map(|segment_entry| segment_entry.segment().clone())
+            .collect()
+    }
+
     pub fn segment_entries(&self) -> Vec<SegmentEntry> {
         self.segment_states.values().cloned().collect()
     }
@@ -79,11 +89,17 @@ impl SegmentRegister {
         self.segment_states.get(segment_id).cloned()
     }
 
-    pub fn new(segment_metas: Vec<SegmentMeta>, delete_cursor: &DeleteCursor) -> SegmentRegister {
+    pub fn new(
+        directory: &ManagedDirectory,
+        schema: &Schema,
+        segment_metas: Vec<SegmentMeta>,
+        delete_cursor: &DeleteCursor,
+    ) -> SegmentRegister {
         let mut segment_states = HashMap::new();
         for segment_meta in segment_metas {
             let segment_id = segment_meta.id();
-            let segment_entry = SegmentEntry::new(segment_meta, delete_cursor.clone(), None);
+            let segment = Segment::new_persisted(segment_meta, directory.clone(), schema.clone());
+            let segment_entry = SegmentEntry::new(segment, delete_cursor.clone(), None);
             segment_states.insert(segment_id, segment_entry);
         }
         SegmentRegister { segment_states }
@@ -95,6 +111,7 @@ mod tests {
     use super::*;
     use crate::core::{SegmentId, SegmentMetaInventory};
     use crate::indexer::delete_queue::*;
+    use crate::indexer::ResourceManager;
 
     fn segment_ids(segment_register: &SegmentRegister) -> Vec<SegmentId> {
         segment_register
@@ -108,28 +125,34 @@ mod tests {
     fn test_segment_register() {
         let inventory = SegmentMetaInventory::default();
         let delete_queue = DeleteQueue::new();
+        let schema = Schema::builder().build();
 
         let mut segment_register = SegmentRegister::default();
         let segment_id_a = SegmentId::generate_random();
         let segment_id_b = SegmentId::generate_random();
         let segment_id_merged = SegmentId::generate_random();
 
+        let memory_manager = ResourceManager::default();
+
         {
-            let segment_meta = inventory.new_segment_meta(segment_id_a, 0u32);
-            let segment_entry = SegmentEntry::new(segment_meta, delete_queue.cursor(), None);
+            let meta = inventory.new_segment_meta(segment_id_a, 0u32);
+            let segment = Segment::new_volatile(meta, schema.clone(), memory_manager.clone());
+            let segment_entry = SegmentEntry::new(segment, delete_queue.cursor(), None);
             segment_register.add_segment_entry(segment_entry);
         }
         assert_eq!(segment_ids(&segment_register), vec![segment_id_a]);
         {
-            let segment_meta = inventory.new_segment_meta(segment_id_b, 0u32);
-            let segment_entry = SegmentEntry::new(segment_meta, delete_queue.cursor(), None);
+            let meta = inventory.new_segment_meta(segment_id_b, 0u32);
+            let segment = Segment::new_volatile(meta, schema.clone(), memory_manager.clone());
+            let segment_entry = SegmentEntry::new(segment, delete_queue.cursor(), None);
             segment_register.add_segment_entry(segment_entry);
         }
         segment_register.remove_segment(&segment_id_a);
         segment_register.remove_segment(&segment_id_b);
         {
             let segment_meta_merged = inventory.new_segment_meta(segment_id_merged, 0u32);
-            let segment_entry = SegmentEntry::new(segment_meta_merged, delete_queue.cursor(), None);
+            let segment_merged = Segment::new_volatile(segment_meta_merged, schema.clone(), memory_manager.clone());
+            let segment_entry = SegmentEntry::new(segment_merged, delete_queue.cursor(), None);
             segment_register.add_segment_entry(segment_entry);
         }
         assert_eq!(segment_ids(&segment_register), vec![segment_id_merged]);

src/indexer/segment_serializer.rs

@@ -3,12 +3,10 @@ use crate::core::SegmentComponent;
 use crate::fastfield::FastFieldSerializer;
 use crate::fieldnorm::FieldNormsSerializer;
 use crate::postings::InvertedIndexSerializer;
-use crate::store::StoreWriter;
 
 /// Segment serializer is in charge of laying out on disk
 /// the data accumulated and sorted by the `SegmentWriter`.
 pub struct SegmentSerializer {
-    store_writer: StoreWriter,
     fast_field_serializer: FastFieldSerializer,
     fieldnorms_serializer: FieldNormsSerializer,
     postings_serializer: InvertedIndexSerializer,
@@ -17,8 +15,6 @@ pub struct SegmentSerializer {
 impl SegmentSerializer {
     /// Creates a new `SegmentSerializer`.
     pub fn for_segment(segment: &mut Segment) -> crate::Result<SegmentSerializer> {
-        let store_write = segment.open_write(SegmentComponent::STORE)?;
-
         let fast_field_write = segment.open_write(SegmentComponent::FASTFIELDS)?;
         let fast_field_serializer = FastFieldSerializer::from_write(fast_field_write)?;
 
@@ -27,7 +23,6 @@ impl SegmentSerializer {
 
         let postings_serializer = InvertedIndexSerializer::open(segment)?;
         Ok(SegmentSerializer {
-            store_writer: StoreWriter::new(store_write),
             fast_field_serializer,
             fieldnorms_serializer,
             postings_serializer,
@@ -49,16 +44,10 @@ impl SegmentSerializer {
         &mut self.fieldnorms_serializer
     }
 
-    /// Accessor to the `StoreWriter`.
-    pub fn get_store_writer(&mut self) -> &mut StoreWriter {
-        &mut self.store_writer
-    }
-
     /// Finalize the segment serialization.
     pub fn close(self) -> crate::Result<()> {
         self.fast_field_serializer.close()?;
         self.postings_serializer.close()?;
-        self.store_writer.close()?;
         self.fieldnorms_serializer.close()?;
         Ok(())
     }

src/indexer/segment_updater.rs

@@ -1,4 +1,4 @@
-use super::segment_manager::{get_mergeable_segments, SegmentManager};
+use super::segment_manager::SegmentManager;
 use crate::core::Index;
 use crate::core::IndexMeta;
 use crate::core::Segment;
@@ -7,18 +7,16 @@ use crate::core::SegmentMeta;
 use crate::core::SerializableSegment;
 use crate::core::META_FILEPATH;
 use crate::directory::{Directory, DirectoryClone, GarbageCollectionResult};
-use crate::indexer::delete_queue::DeleteCursor;
 use crate::indexer::index_writer::advance_deletes;
-use crate::indexer::merge_operation::MergeOperationInventory;
 use crate::indexer::merger::IndexMerger;
-use crate::indexer::segment_manager::SegmentsStatus;
+use crate::indexer::segment_manager::{SegmentRegisters, SegmentsStatus};
 use crate::indexer::stamper::Stamper;
 use crate::indexer::SegmentEntry;
 use crate::indexer::SegmentSerializer;
 use crate::indexer::{DefaultMergePolicy, MergePolicy};
 use crate::indexer::{MergeCandidate, MergeOperation};
 use crate::schema::Schema;
-use crate::Opstamp;
+use crate::{Opstamp, SegmentComponent};
 use futures::channel::oneshot;
 use futures::executor::{ThreadPool, ThreadPoolBuilder};
 use futures::future::Future;
@@ -117,15 +115,14 @@ fn merge(
 
     // First we apply all of the delet to the merged segment, up to the target opstamp.
     for segment_entry in &mut segment_entries {
-        let segment = index.segment(segment_entry.meta().clone());
-        advance_deletes(segment, segment_entry, target_opstamp)?;
+        advance_deletes(segment_entry, target_opstamp)?;
     }
 
-    let delete_cursor = segment_entries[0].delete_cursor().clone();
+    let delete_cursor = segment_entries[0].delete_cursor();
 
     let segments: Vec<Segment> = segment_entries
         .iter()
-        .map(|segment_entry| index.segment(segment_entry.meta().clone()))
+        .map(|segment_entry| segment_entry.segment().clone())
         .collect();
 
     // An IndexMerger is like a "view" of our merged segments.
@@ -134,11 +131,16 @@ fn merge(
     // ... we just serialize this index merger in our new segment to merge the two segments.
     let segment_serializer = SegmentSerializer::for_segment(&mut merged_segment)?;
 
-    let num_docs = merger.write(segment_serializer)?;
+    let store_wrt = merged_segment.open_write(SegmentComponent::STORE)?;
+    merger.write_storable_fields(store_wrt)?;
 
-    let segment_meta = index.new_segment_meta(merged_segment.id(), num_docs);
+    let max_doc = merger.write(segment_serializer)?;
 
-    Ok(SegmentEntry::new(segment_meta, delete_cursor, None))
+    Ok(SegmentEntry::new(
+        merged_segment.with_max_doc(max_doc),
+        delete_cursor,
+        None,
+    ))
 }
 
 pub(crate) struct InnerSegmentUpdater {
@@ -157,17 +159,15 @@ pub(crate) struct InnerSegmentUpdater {
     merge_policy: RwLock<Arc<Box<dyn MergePolicy>>>,
     killed: AtomicBool,
     stamper: Stamper,
-    merge_operations: MergeOperationInventory,
 }
 
 impl SegmentUpdater {
     pub fn create(
+        segment_registers: Arc<RwLock<SegmentRegisters>>,
         index: Index,
         stamper: Stamper,
-        delete_cursor: &DeleteCursor,
     ) -> crate::Result<SegmentUpdater> {
-        let segments = index.searchable_segment_metas()?;
-        let segment_manager = SegmentManager::from_segments(segments, delete_cursor);
+        let segment_manager = SegmentManager::new(segment_registers);
         let pool = ThreadPoolBuilder::new()
             .name_prefix("segment_updater")
             .pool_size(1)
@@ -196,7 +196,6 @@ impl SegmentUpdater {
             merge_policy: RwLock::new(Arc::new(Box::new(DefaultMergePolicy::default()))),
             killed: AtomicBool::new(false),
             stamper,
-            merge_operations: Default::default(),
         })))
     }
 
@@ -262,8 +261,7 @@ impl SegmentUpdater {
     fn purge_deletes(&self, target_opstamp: Opstamp) -> crate::Result<Vec<SegmentEntry>> {
         let mut segment_entries = self.segment_manager.segment_entries();
         for segment_entry in &mut segment_entries {
-            let segment = self.index.segment(segment_entry.meta().clone());
-            advance_deletes(segment, segment_entry, target_opstamp)?;
+            advance_deletes(segment_entry, target_opstamp)?;
         }
         Ok(segment_entries)
     }
@@ -331,12 +329,21 @@ impl SegmentUpdater {
         &self,
         opstamp: Opstamp,
         payload: Option<String>,
+        soft_commit: bool,
     ) -> impl Future<Output = crate::Result<()>> {
         let segment_updater: SegmentUpdater = self.clone();
+        let directory = self.index.directory().clone();
         self.schedule_future(async move {
-            let segment_entries = segment_updater.purge_deletes(opstamp)?;
+            let mut segment_entries = segment_updater.purge_deletes(opstamp)?;
+            if !soft_commit {
+                for segment_entry in &mut segment_entries {
+                    segment_entry.persist(directory.clone())?;
+                }
+            }
             segment_updater.segment_manager.commit(segment_entries);
-            segment_updater.save_metas(opstamp, payload)?;
+            if !soft_commit {
+                segment_updater.save_metas(opstamp, payload)?;
+            }
             let _ = garbage_collect_files(segment_updater.clone()).await;
             segment_updater.consider_merge_options().await;
             Ok(())
@@ -353,7 +360,7 @@ impl SegmentUpdater {
 
     pub(crate) fn make_merge_operation(&self, segment_ids: &[SegmentId]) -> MergeOperation {
         let commit_opstamp = self.load_metas().opstamp;
-        MergeOperation::new(&self.merge_operations, commit_opstamp, segment_ids.to_vec())
+        self.segment_manager.new_merge_operation(commit_opstamp, MergeCandidate(segment_ids.to_vec()))
     }
 
     // Starts a merge operation. This function will block until the merge operation is effectively
@@ -427,9 +434,8 @@ impl SegmentUpdater {
     }
 
     async fn consider_merge_options(&self) {
-        let merge_segment_ids: HashSet<SegmentId> = self.merge_operations.segment_in_merge();
         let (committed_segments, uncommitted_segments) =
-            get_mergeable_segments(&merge_segment_ids, &self.segment_manager);
+            self.segment_manager.segments_not_in_merge();
 
         // Committed segments cannot be merged with uncommitted_segments.
         // We therefore consider merges using these two sets of segments independently.
@@ -440,7 +446,7 @@ impl SegmentUpdater {
             .compute_merge_candidates(&uncommitted_segments)
             .into_iter()
             .map(|merge_candidate| {
-                MergeOperation::new(&self.merge_operations, current_opstamp, merge_candidate.0)
+                self.segment_manager.new_merge_operation(current_opstamp, merge_candidate)
             })
             .collect();
 
@@ -449,7 +455,7 @@ impl SegmentUpdater {
             .compute_merge_candidates(&committed_segments)
             .into_iter()
             .map(|merge_candidate: MergeCandidate| {
-                MergeOperation::new(&self.merge_operations, commit_opstamp, merge_candidate.0)
+                self.segment_manager.new_merge_operation(commit_opstamp, merge_candidate)
             })
             .collect::<Vec<_>>();
         merge_candidates.extend(committed_merge_candidates.into_iter());
@@ -474,17 +480,13 @@ impl SegmentUpdater {
         let end_merge_future = self.schedule_future(async move {
             info!("End merge {:?}", after_merge_segment_entry.meta());
             {
-                let mut delete_cursor = after_merge_segment_entry.delete_cursor().clone();
+                let mut delete_cursor = after_merge_segment_entry.delete_cursor();
                 if let Some(delete_operation) = delete_cursor.get() {
                     let committed_opstamp = segment_updater.load_metas().opstamp;
                     if delete_operation.opstamp < committed_opstamp {
-                        let index = &segment_updater.index;
-                        let segment = index.segment(after_merge_segment_entry.meta().clone());
-                        if let Err(e) = advance_deletes(
-                            segment,
-                            &mut after_merge_segment_entry,
-                            committed_opstamp,
-                        ) {
+                        if let Err(e) =
+                            advance_deletes(&mut after_merge_segment_entry, committed_opstamp)
+                        {
                             error!(
                                 "Merge of {:?} was cancelled (advancing deletes failed): {:?}",
                                 merge_operation.segment_ids(),
@@ -533,9 +535,8 @@ impl SegmentUpdater {
     ///
     /// Obsolete files will eventually be cleaned up
     /// by the directory garbage collector.
-    pub fn wait_merging_thread(&self) -> crate::Result<()> {
-        self.merge_operations.wait_until_empty();
-        Ok(())
+    pub fn wait_merging_thread(&self) {
+        self.segment_manager.wait_merging_thread()
     }
 }
 
@@ -544,7 +545,8 @@ mod tests {
 
     use crate::indexer::merge_policy::tests::MergeWheneverPossible;
     use crate::schema::*;
-    use crate::Index;
+    use crate::{Index, SegmentId};
+    use futures::executor::block_on;
 
     #[test]
     fn test_delete_during_merge() {
@@ -695,4 +697,27 @@ mod tests {
             .segment_entries();
         assert!(seg_vec.is_empty());
     }
+
+    #[test]
+    fn test_merge_over_soft_commit() {
+        let mut schema_builder = Schema::builder();
+        let text_field = schema_builder.add_text_field("text", TEXT);
+        let schema = schema_builder.build();
+        let index = Index::create_in_ram(schema);
+        // writing the segment
+        let mut index_writer = index.writer_with_num_threads(1, 3_000_000).unwrap();
+        index_writer.add_document(doc!(text_field=>"a"));
+        assert!(index_writer.soft_commit().is_ok());
+        index_writer.add_document(doc!(text_field=>"a"));
+        assert!(index_writer.soft_commit().is_ok());
+
+        let reader = index_writer.reader(1).unwrap();
+        let segment_ids: Vec<SegmentId> = reader
+            .searcher()
+            .segment_readers()
+            .iter()
+            .map(|reader| reader.segment_id())
+            .collect();
+        assert!(block_on(index_writer.merge(&segment_ids)).is_ok());
+    }
 }

src/indexer/segment_writer.rs

@@ -1,9 +1,11 @@
 use super::operation::AddOperation;
 use crate::core::Segment;
 use crate::core::SerializableSegment;
+use crate::directory::{SpillingResult, SpillingWriter, TerminatingWrite};
 use crate::fastfield::FastFieldsWriter;
 use crate::fieldnorm::FieldNormsWriter;
 use crate::indexer::segment_serializer::SegmentSerializer;
+use crate::indexer::IndexWriterConfig;
 use crate::postings::compute_table_size;
 use crate::postings::MultiFieldPostingsWriter;
 use crate::schema::FieldType;
@@ -11,21 +13,24 @@ use crate::schema::Schema;
 use crate::schema::Term;
 use crate::schema::Value;
 use crate::schema::{Field, FieldEntry};
+use crate::store::StoreWriter;
 use crate::tokenizer::{BoxTokenStream, PreTokenizedStream};
-use crate::tokenizer::{FacetTokenizer, TextAnalyzer};
+use crate::tokenizer::{FacetTokenizer, TextAnalyzer, TokenizerManager};
 use crate::tokenizer::{TokenStreamChain, Tokenizer};
-use crate::DocId;
 use crate::Opstamp;
+use crate::{DocId, SegmentComponent};
 use std::io;
+use std::io::Write;
 use std::str;
+use crate::indexer::resource_manager::ResourceManager;
 
 /// Computes the initial size of the hash table.
 ///
 /// Returns a number of bit `b`, such that the recommended initial table size is 2^b.
-fn initial_table_size(per_thread_memory_budget: usize) -> crate::Result<usize> {
-    let table_memory_upper_bound = per_thread_memory_budget / 3;
+fn initial_table_size(per_thread_memory_budget: u64) -> crate::Result<usize> {
+    let table_memory_upper_bound = per_thread_memory_budget / 3u64;
     if let Some(limit) = (10..)
-        .take_while(|num_bits: &usize| compute_table_size(*num_bits) < table_memory_upper_bound)
+        .take_while(|num_bits| compute_table_size(*num_bits) < table_memory_upper_bound)
         .last()
     {
         Ok(limit.min(19)) // we cap it at 2^19 = 512K.
@@ -43,11 +48,13 @@ fn initial_table_size(per_thread_memory_budget: usize) -> crate::Result<usize> {
 pub struct SegmentWriter {
     max_doc: DocId,
     multifield_postings: MultiFieldPostingsWriter,
-    segment_serializer: SegmentSerializer,
+    segment: Segment,
     fast_field_writers: FastFieldsWriter,
     fieldnorms_writer: FieldNormsWriter,
     doc_opstamps: Vec<Opstamp>,
     tokenizers: Vec<Option<TextAnalyzer>>,
+    store_writer: StoreWriter<SpillingWriter>,
+    memory_manager: ResourceManager,
 }
 
 impl SegmentWriter {
@@ -61,12 +68,13 @@ impl SegmentWriter {
     /// - segment: The segment being written
     /// - schema
     pub fn for_segment(
-        memory_budget: usize,
-        mut segment: Segment,
+        config: &IndexWriterConfig,
+        segment: Segment,
         schema: &Schema,
+        tokenizer_manager: &TokenizerManager,
+        memory_manager: ResourceManager
     ) -> crate::Result<SegmentWriter> {
-        let table_num_bits = initial_table_size(memory_budget)?;
-        let segment_serializer = SegmentSerializer::for_segment(&mut segment)?;
+        let table_num_bits = initial_table_size(config.heap_size_in_byte_per_thread())?;
         let multifield_postings = MultiFieldPostingsWriter::new(schema, table_num_bits);
         let tokenizers = schema
             .fields()
@@ -76,20 +84,32 @@ impl SegmentWriter {
                         .get_indexing_options()
                         .and_then(|text_index_option| {
                             let tokenizer_name = &text_index_option.tokenizer();
-                            segment.index().tokenizers().get(tokenizer_name)
+                            tokenizer_manager.get(tokenizer_name)
                         }),
                     _ => None,
                 },
             )
             .collect();
+        let mut segment_clone = segment.clone();
+        let spilling_wrt = SpillingWriter::new(
+            50_000_000,
+            Box::new(move || {
+                segment_clone
+                    .open_write(SegmentComponent::STORE)
+                    .map_err(|err| io::Error::new(io::ErrorKind::Other, err))
+            }),
+        );
+        let store_writer = StoreWriter::new(spilling_wrt);
         Ok(SegmentWriter {
             max_doc: 0,
             multifield_postings,
             fieldnorms_writer: FieldNormsWriter::for_schema(schema),
-            segment_serializer,
+            segment,
             fast_field_writers: FastFieldsWriter::from_schema(schema),
             doc_opstamps: Vec::with_capacity(1_000),
             tokenizers,
+            store_writer,
+            memory_manager
         })
     }
 
@@ -97,18 +117,33 @@ impl SegmentWriter {
     ///
     /// Finalize consumes the `SegmentWriter`, so that it cannot
     /// be used afterwards.
-    pub fn finalize(mut self) -> crate::Result<Vec<u64>> {
+    pub fn finalize(mut self) -> crate::Result<(Segment, Vec<u64>)> {
         self.fieldnorms_writer.fill_up_to_max_doc(self.max_doc);
+        let spilling_wrt = self.store_writer.close()?;
+        let mut segment: Segment;
+        match spilling_wrt.finalize()? {
+            SpillingResult::Spilled => {
+                segment = self.segment.clone();
+            }
+            SpillingResult::Buffer(buf) => {
+                segment = self.segment.into_volatile(self.memory_manager.clone());
+                let mut store_wrt = segment.open_write(SegmentComponent::STORE)?;
+                store_wrt.write_all(&buf[..])?;
+                store_wrt.terminate()?;
+            }
+        }
+        let segment_serializer = SegmentSerializer::for_segment(&mut segment)?;
+        segment = segment.with_max_doc(self.max_doc);
         write(
             &self.multifield_postings,
             &self.fast_field_writers,
             &self.fieldnorms_writer,
-            self.segment_serializer,
+            segment_serializer,
         )?;
-        Ok(self.doc_opstamps)
+        Ok((segment, self.doc_opstamps))
     }
 
-    pub fn mem_usage(&self) -> usize {
+    pub fn mem_usage(&self) -> u64 {
         self.multifield_postings.mem_usage()
     }
 
@@ -246,8 +281,7 @@ impl SegmentWriter {
         }
         doc.filter_fields(|field| schema.get_field_entry(field).is_stored());
         doc.prepare_for_store();
-        let doc_writer = self.segment_serializer.get_store_writer();
-        doc_writer.store(&doc)?;
+        self.store_writer.store(&doc)?;
         self.max_doc += 1;
         Ok(())
     }

src/lib.rs

@@ -163,7 +163,7 @@ pub use crate::core::{Index, IndexMeta, Searcher, Segment, SegmentId, SegmentMet
 pub use crate::core::{InvertedIndexReader, SegmentReader};
 pub use crate::directory::Directory;
 pub use crate::indexer::operation::UserOperation;
-pub use crate::indexer::IndexWriter;
+pub use crate::indexer::{IndexWriter, IndexWriterConfig};
 pub use crate::postings::Postings;
 pub use crate::reader::LeasedItem;
 pub use crate::schema::{Document, Term};

src/postings/mod.rs

@@ -54,7 +54,7 @@ pub mod tests {
     use crate::docset::{DocSet, SkipResult};
     use crate::fieldnorm::FieldNormReader;
     use crate::indexer::operation::AddOperation;
-    use crate::indexer::SegmentWriter;
+    use crate::indexer::{IndexWriterConfig, SegmentWriter, ResourceManager};
     use crate::merge_policy::NoMergePolicy;
     use crate::query::Scorer;
     use crate::schema::{Document, Schema, Term, INDEXED, STRING, TEXT};
@@ -216,11 +216,17 @@ pub mod tests {
         let text_field = schema_builder.add_text_field("text", TEXT);
         let schema = schema_builder.build();
         let index = Index::create_in_ram(schema.clone());
-        let segment = index.new_segment();
 
-        {
-            let mut segment_writer =
-                SegmentWriter::for_segment(3_000_000, segment.clone(), &schema).unwrap();
+        let resource_manager = ResourceManager::default();
+        let segment = {
+            let mut segment_writer = SegmentWriter::for_segment(
+                &IndexWriterConfig::for_test(),
+                index.new_segment(),
+                &schema,
+                index.tokenizers(),
+                resource_manager
+            )
+            .unwrap();
             {
                 let mut doc = Document::default();
                 // checking that position works if the field has two values
@@ -252,8 +258,9 @@ pub mod tests {
                 };
                 segment_writer.add_document(op, &schema).unwrap();
             }
-            segment_writer.finalize().unwrap();
-        }
+            let (segment, _) = segment_writer.finalize().unwrap();
+            segment
+        };
         {
             let segment_reader = SegmentReader::open(&segment).unwrap();
             {

src/postings/postings_writer.rs

@@ -95,7 +95,7 @@ impl MultiFieldPostingsWriter {
         }
     }
 
-    pub fn mem_usage(&self) -> usize {
+    pub fn mem_usage(&self) -> u64 {
         self.term_index.mem_usage() + self.heap.mem_usage()
     }
 

src/postings/stacker/memory_arena.rs

@@ -106,8 +106,8 @@ impl MemoryArena {
     ///
     /// Internally, it counts a number of `1MB` pages
     /// and therefore delivers an upperbound.
-    pub fn mem_usage(&self) -> usize {
-        self.pages.len() * PAGE_SIZE
+    pub fn mem_usage(&self) -> u64 {
+        (self.pages.len() as u64) * (PAGE_SIZE as u64)
     }
 
     pub fn write_at<Item: Copy + 'static>(&mut self, addr: Addr, val: Item) {

src/postings/stacker/term_hashmap.rs

@@ -12,8 +12,8 @@ use std::slice;
 
 /// Returns the actual memory size in bytes
 /// required to create a table of size $2^num_bits$.
-pub fn compute_table_size(num_bits: usize) -> usize {
-    (1 << num_bits) * mem::size_of::<KeyValue>()
+pub fn compute_table_size(num_bits: usize) -> u64 {
+    (1u64 << num_bits as u64) * mem::size_of::<KeyValue>() as u64
 }
 
 /// `KeyValue` is the item stored in the hash table.
@@ -116,8 +116,8 @@ impl TermHashMap {
         QuadraticProbing::compute(hash as usize, self.mask)
     }
 
-    pub fn mem_usage(&self) -> usize {
-        self.table.len() * mem::size_of::<KeyValue>()
+    pub fn mem_usage(&self) -> u64 {
+        self.table.len() as u64 * mem::size_of::<KeyValue>() as u64
     }
 
     fn is_saturated(&self) -> bool {

src/reader/index_writer_reader.rs

@@ -0,0 +1,84 @@
+use crate::directory::{WatchCallbackList, WatchHandle};
+use crate::indexer::SegmentRegisters;
+use crate::reader::pool::Pool;
+use crate::{Index, LeasedItem, Searcher, Segment, SegmentReader};
+use std::iter::repeat_with;
+use std::sync::{Arc, RwLock, Weak};
+
+struct InnerNRTReader {
+    num_searchers: usize,
+    index: Index,
+    searcher_pool: Pool<Searcher>,
+    segment_registers: Arc<RwLock<SegmentRegisters>>,
+}
+
+impl InnerNRTReader {
+    fn load_segment_readers(&self) -> crate::Result<Vec<SegmentReader>> {
+        let segments: Vec<Segment> = self
+            .segment_registers
+            .read()
+            .expect("lock should never be polluted. Please report.")
+            .committed_segment();
+        segments
+            .iter()
+            .map(SegmentReader::open)
+            .collect::<crate::Result<Vec<SegmentReader>>>()
+    }
+
+    pub fn reload(&self) -> crate::Result<()> {
+        let segment_readers: Vec<SegmentReader> = self.load_segment_readers()?;
+        let schema = self.index.schema();
+        let searchers = repeat_with(|| {
+            Searcher::new(schema.clone(), self.index.clone(), segment_readers.clone())
+        })
+        .take(self.num_searchers)
+        .collect();
+        self.searcher_pool.publish_new_generation(searchers);
+        Ok(())
+    }
+
+    pub fn searcher(&self) -> LeasedItem<Searcher> {
+        self.searcher_pool.acquire()
+    }
+}
+
+#[derive(Clone)]
+pub struct NRTReader {
+    inner: Arc<InnerNRTReader>,
+    watch_handle: WatchHandle,
+}
+
+impl NRTReader {
+    pub fn reload(&self) -> crate::Result<()> {
+        self.inner.reload()
+    }
+
+    pub fn searcher(&self) -> LeasedItem<Searcher> {
+        self.inner.searcher()
+    }
+
+    pub(crate) fn create(
+        num_searchers: usize,
+        index: Index,
+        segment_registers: Arc<RwLock<SegmentRegisters>>,
+        watch_callback_list: &WatchCallbackList,
+    ) -> crate::Result<Self> {
+        let inner_reader: Arc<InnerNRTReader> = Arc::new(InnerNRTReader {
+            num_searchers,
+            index,
+            searcher_pool: Pool::new(),
+            segment_registers,
+        });
+        let inner_reader_weak: Weak<InnerNRTReader> = Arc::downgrade(&inner_reader);
+        let watch_handle = watch_callback_list.subscribe(Box::new(move || {
+            if let Some(nrt_reader_arc) = inner_reader_weak.upgrade() {
+                let _ = nrt_reader_arc.reload();
+            }
+        }));
+        inner_reader.reload()?;
+        Ok(NRTReader {
+            inner: inner_reader,
+            watch_handle,
+        })
+    }
+}

src/reader/meta_file_reader.rs

@@ -0,0 +1,177 @@
+use super::pool::Pool;
+use crate::core::Segment;
+use crate::directory::Directory;
+use crate::directory::WatchHandle;
+use crate::directory::META_LOCK;
+use crate::Searcher;
+use crate::SegmentReader;
+use crate::{Index, LeasedItem};
+use crate::{IndexReader, Result};
+use std::iter::repeat_with;
+use std::sync::Arc;
+
+/// 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::Manual,
+            index,
+        }
+    }
+
+    /// 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
+    }
+
+    /// 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.
+    pub fn try_into(self) -> crate::Result<IndexReader> {
+        let inner_reader = MetaFileIndexReaderInner {
+            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::from(MetaFileIndexReader {
+            inner: inner_reader_arc,
+            watch_handle_opt,
+        }))
+    }
+}
+
+struct MetaFileIndexReaderInner {
+    num_searchers: usize,
+    searcher_pool: Pool<Searcher>,
+    index: Index,
+}
+
+impl MetaFileIndexReaderInner {
+    fn load_segment_readers(&self) -> crate::Result<Vec<SegmentReader>> {
+        // We keep the lock until we have effectively finished opening the
+        // the `SegmentReader` because it prevents a diffferent process
+        // to garbage collect these file while we open them.
+        //
+        // Once opened, on linux & mac, the mmap will remain valid after
+        // the file has been deleted
+        // On windows, the file deletion will fail.
+        let _meta_lock = self.index.directory().acquire_lock(&META_LOCK)?;
+        let searchable_segments = self.searchable_segments()?;
+        searchable_segments
+            .iter()
+            .map(SegmentReader::open)
+            .collect::<Result<_>>()
+    }
+
+    fn reload(&self) -> crate::Result<()> {
+        let segment_readers: Vec<SegmentReader> = self.load_segment_readers()?;
+        let schema = self.index.schema();
+        let searchers = repeat_with(|| {
+            Searcher::new(schema.clone(), self.index.clone(), segment_readers.clone())
+        })
+        .take(self.num_searchers)
+        .collect();
+        self.searcher_pool.publish_new_generation(searchers);
+        Ok(())
+    }
+
+    /// Returns the list of segments that are searchable
+    fn searchable_segments(&self) -> crate::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 MetaFileIndexReader {
+    inner: Arc<MetaFileIndexReaderInner>,
+    watch_handle_opt: Option<WatchHandle>,
+}
+
+impl MetaFileIndexReader {
+    pub fn reload(&self) -> crate::Result<()> {
+        self.inner.reload()
+    }
+    pub fn searcher(&self) -> LeasedItem<Searcher> {
+        self.inner.searcher()
+    }
+}

src/reader/mod.rs

@@ -1,151 +1,14 @@
+mod index_writer_reader;
+mod meta_file_reader;
 mod pool;
 
+use self::meta_file_reader::MetaFileIndexReader;
+pub use self::meta_file_reader::{IndexReaderBuilder, ReloadPolicy};
 pub use self::pool::LeasedItem;
-use self::pool::Pool;
-use crate::core::Segment;
-use crate::directory::Directory;
-use crate::directory::WatchHandle;
-use crate::directory::META_LOCK;
-use crate::Index;
+
+pub(crate) use crate::reader::index_writer_reader::NRTReader;
+
 use crate::Searcher;
-use crate::SegmentReader;
-use std::sync::Arc;
-
-/// 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) -> crate::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) -> crate::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::<crate::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) -> crate::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.
 ///
@@ -155,17 +18,12 @@ impl InnerIndexReader {
 /// `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>,
+pub enum IndexReader {
+    FromMetaFile(MetaFileIndexReader),
+    NRT(NRTReader),
 }
 
 impl IndexReader {
-    #[cfg(test)]
-    pub(crate) fn index(&self) -> Index {
-        self.inner.index.clone()
-    }
-
     /// Update searchers so that they reflect the state of the last
     /// `.commit()`.
     ///
@@ -176,7 +34,10 @@ impl IndexReader {
     /// 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) -> crate::Result<()> {
-        self.inner.reload()
+        match self {
+            IndexReader::FromMetaFile(meta_file_reader) => meta_file_reader.reload(),
+            IndexReader::NRT(nrt_reader) => nrt_reader.reload(),
+        }
     }
 
     /// Returns a searcher
@@ -190,6 +51,21 @@ impl IndexReader {
     /// 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()
+        match self {
+            IndexReader::FromMetaFile(meta_file_reader) => meta_file_reader.searcher(),
+            IndexReader::NRT(nrt_reader) => nrt_reader.searcher(),
+        }
+    }
+}
+
+impl From<MetaFileIndexReader> for IndexReader {
+    fn from(meta_file_reader: MetaFileIndexReader) -> Self {
+        IndexReader::FromMetaFile(meta_file_reader)
+    }
+}
+
+impl From<NRTReader> for IndexReader {
+    fn from(nrt_reader: NRTReader) -> Self {
+        IndexReader::NRT(nrt_reader)
     }
 }

src/store/writer.rs

@@ -3,8 +3,6 @@ use super::skiplist::SkipListBuilder;
 use super::StoreReader;
 use crate::common::CountingWriter;
 use crate::common::{BinarySerializable, VInt};
-use crate::directory::TerminatingWrite;
-use crate::directory::WritePtr;
 use crate::schema::Document;
 use crate::DocId;
 use std::io::{self, Write};
@@ -19,20 +17,20 @@ const BLOCK_SIZE: usize = 16_384;
 ///
 /// The skip list index on the other hand, is built in memory.
 ///
-pub struct StoreWriter {
+pub struct StoreWriter<W: io::Write> {
     doc: DocId,
     offset_index_writer: SkipListBuilder<u64>,
-    writer: CountingWriter<WritePtr>,
+    writer: CountingWriter<W>,
     intermediary_buffer: Vec<u8>,
     current_block: Vec<u8>,
 }
 
-impl StoreWriter {
+impl<W: io::Write> StoreWriter<W> {
     /// Create a store writer.
     ///
     /// The store writer will writes blocks on disc as
     /// document are added.
-    pub fn new(writer: WritePtr) -> StoreWriter {
+    pub fn new(writer: W) -> StoreWriter<W> {
         StoreWriter {
             doc: 0,
             offset_index_writer: SkipListBuilder::new(4),
@@ -102,7 +100,9 @@ impl StoreWriter {
     ///
     /// Compress the last unfinished block if any,
     /// and serializes the skip list index on disc.
-    pub fn close(mut self) -> io::Result<()> {
+    ///
+    /// The returned writer is not flushed.
+    pub fn close(mut self) -> io::Result<W> {
         if !self.current_block.is_empty() {
             self.write_and_compress_block()?;
         }
@@ -110,6 +110,7 @@ impl StoreWriter {
         self.offset_index_writer.write(&mut self.writer)?;
         header_offset.serialize(&mut self.writer)?;
         self.doc.serialize(&mut self.writer)?;
-        self.writer.terminate()
+        let (wrt, _) = self.writer.finish()?;
+        Ok(wrt)
     }
 }