Naive multithreading. Performance is already there.

Cargo.toml

@@ -20,6 +20,7 @@ tempdir = "0.3.4"
 bincode = "0.4.0"
 serde = "0.6.11"
 libc = "0.2.6"
+num_cpus = "0.2"
 lz4 = "1.13.131"
 time = "0.1.34"
 uuid = "0.1"

src/core/directory.rs

@@ -1,4 +1,6 @@
 use std::io::BufWriter;
+use std::marker::Send;
+use std::marker::Sync;
 use std::io;
 use std::io::Cursor;
 use std::io::Write;
@@ -84,7 +86,7 @@ pub type WritePtr = Box<SeekableWrite>;
 //     CannotCreateTempDirectory(io::Error),
 // }
 
-pub trait Directory: fmt::Debug {
+pub trait Directory: fmt::Debug + Send + Sync {
     fn open_read(&self, path: &Path) -> io::Result<ReadOnlySource>;
     fn open_write(&mut self, path: &Path) -> io::Result<WritePtr>;
     fn atomic_write(&mut self, path: &Path, data: &[u8]) -> io::Result<()>;
@@ -93,15 +95,12 @@ pub trait Directory: fmt::Debug {
 }
 
 
-
-
-
 ////////////////////////////////////////////////////////////////
 // MmapDirectory
 
 pub struct MmapDirectory {
     root_path: PathBuf,
-    mmap_cache: RefCell<HashMap<PathBuf, MmapReadOnly>>,
+    mmap_cache: RwLock<HashMap<PathBuf, MmapReadOnly>>,
     _temp_directory: Option<TempDir>,
 }
 
@@ -120,7 +119,7 @@ impl MmapDirectory {
         let tempdir_path = PathBuf::from(tempdir.path());
         let directory = MmapDirectory {
             root_path: PathBuf::from(tempdir_path),
-            mmap_cache: RefCell::new(HashMap::new()),
+            mmap_cache: RwLock::new(HashMap::new()),
             _temp_directory: Some(tempdir)
         };
         Ok(directory)
@@ -129,7 +128,7 @@ impl MmapDirectory {
     pub fn create(filepath: &Path) -> io::Result<MmapDirectory> {
         Ok(MmapDirectory {
             root_path: PathBuf::from(filepath),
-            mmap_cache: RefCell::new(HashMap::new()),
+            mmap_cache: RwLock::new(HashMap::new()),
             _temp_directory: None
         })
     }
@@ -144,7 +143,7 @@ impl MmapDirectory {
 impl Directory for MmapDirectory {
     fn open_read(&self, path: &Path) -> io::Result<ReadOnlySource> {
         let full_path = self.resolve_path(path);
-        let mut mmap_cache = self.mmap_cache.borrow_mut();
+        let mut mmap_cache = self.mmap_cache.write().unwrap();
         let mmap = match mmap_cache.entry(full_path.clone()) {
             HashMapEntry::Occupied(e) => e.get().clone(),
             HashMapEntry::Vacant(vacant_entry) => {

src/core/index.rs

@@ -12,6 +12,7 @@ use core::directory::{Directory, MmapDirectory, RAMDirectory, ReadOnlySource, Wr
 use core::writer::IndexWriter;
 use core::searcher::Searcher;
 use uuid::Uuid;
+use num_cpus;
 use core::codec::SegmentSerializer;
 
 
@@ -93,8 +94,12 @@ impl Index {
         Ok(index)
     }
 
+    pub fn writer_with_num_threads(&self, num_threads: usize) -> io::Result<IndexWriter> {
+        IndexWriter::open(self, num_threads)
+    }
+
     pub fn writer(&self,) -> io::Result<IndexWriter> {
-        IndexWriter::open(self,)
+        self.writer_with_num_threads(num_cpus::get())
     }
 
     pub fn searcher(&self,) -> io::Result<Searcher> {
@@ -187,7 +192,7 @@ impl Index {
         let mut w = Vec::new();
         {
             let metas_lock = self.metas.read().unwrap() ;
-            let data = write!(&mut w, "{}\n", json::as_pretty_json(&*metas_lock));
+            try!(write!(&mut w, "{}\n", json::as_pretty_json(&*metas_lock)));
         };
         try!(self.rw_directory())
             .atomic_write(&META_FILEPATH, &w[..])

src/core/writer.rs

@@ -10,32 +10,94 @@ use core::index::Segment;
 use core::index::SegmentInfo;
 use core::postings::PostingsWriter;
 use core::fastfield::U32FastFieldsWriter;
-
+use std::sync::mpsc;
+use std::sync::mpsc::channel;
+use std::thread;
+use std::sync::Mutex;
+use std::sync::mpsc::SyncSender;
+use std::sync::mpsc::Receiver;
+use std::thread::JoinHandle;
+use std::sync::Arc;
 
 pub struct IndexWriter {
-	segment_writer: Rc<SegmentWriter>,
-	directory: Index,
+	// segment_writers: Vec<SegmentWriter>,
+	threads: Vec<JoinHandle<()>>,
+	index: Index,
 	schema: Schema,
+	queue_input: SyncSender<ArcDoc>,
 }
 
+type ArcDoc = Arc<Document>;
+
 impl IndexWriter {
 
-    pub fn open(directory: &Index) -> io::Result<IndexWriter> {
-		let segment = directory.new_segment();
-		let schema = directory.schema();
-		let segment_writer = try!(SegmentWriter::for_segment(segment, &schema));
+	pub fn open(index: &Index, num_threads: usize) -> io::Result<IndexWriter> {
+		let schema = index.schema();
+		let (queue_input, queue_output): (SyncSender<ArcDoc>, Receiver<ArcDoc>) = mpsc::sync_channel(10_000);
+		let queue_output_sendable = Arc::new(Mutex::new(queue_output));
+		let threads = (0..num_threads).map(|thread_id|  {
+			let queue_output_clone = queue_output_sendable.clone();
+			let index_clone = index.clone();
+			let schema_clone = schema.clone();
+			thread::spawn(move || {
+				let mut docs_remaining = true;
+				while docs_remaining {
+					let segment = index_clone.new_segment();
+					let mut segment_writer = SegmentWriter::for_segment(segment, &schema_clone).unwrap();
+					println!("thread_id {}", thread_id);
+					for i in 0..500 {
+						let doc: ArcDoc;
+						{
+							let queue = queue_output_clone.lock().unwrap();
+							match queue.recv() {
+								Ok(doc_) => {
+									doc = doc_;
+								}
+								Err(_) => {
+									docs_remaining = false;
+									println!("err");
+									break;
+								}
+							}
+						}
+						// TODO stop unwrapping that one.
+						segment_writer.add_document(&*doc, &schema_clone).unwrap();
+					}
+					println!("finalize {}", thread_id);
+					segment_writer.finalize().unwrap();
+					// try!(self.directory.sync(segment.clone()));
+					// try!(self.directory.publish_segment(segment.clone()));
+					// segment_writer.commit().unwrap();
+				}
+			})
+		}).collect();
+		// let segment = directory.new_segment();
+		// let segment_writer = try!(SegmentWriter::for_segment(segment, &schema));
+		// segment_writers.push(segment_writer);
 		Ok(IndexWriter {
-			segment_writer: Rc::new(segment_writer),
-			directory: directory.clone(),
+			threads: threads,
+			// segment_writers: segment_writers, //Rc::new(segment_writer),
+			index: index.clone(),
 			schema: schema,
+			queue_input: queue_input,
 		})
+	}
+
+	pub fn wait(self,) {
+		for thread in self.threads {
+			thread.join();
+		}
+	}
+
+    pub fn add_document(&mut self, doc: Document) -> io::Result<()> {
+        // Rc::get_mut(&mut self.segment_writer).unwrap().add_document(&doc, &self.schema)
+		let arc_doc = ArcDoc::new(doc);
+		self.queue_input.send(arc_doc);
+		Ok(())
     }
 
-    pub fn add_document(&mut self, doc: &Document) -> io::Result<()> {
-        Rc::get_mut(&mut self.segment_writer).unwrap().add_document(&doc, &self.schema)
-    }
-
-    pub fn commit(&mut self,) -> io::Result<Segment> {
+    pub fn commit(&mut self,) -> io::Result<Vec<Segment>> {
+		/*
 		let segment_writer_rc = self.segment_writer.clone();
 		let segment = self.directory.new_segment();
 		self.segment_writer = Rc::new(try!(SegmentWriter::for_segment(segment, &self.schema)));
@@ -51,6 +113,8 @@ impl IndexWriter {
 				panic!("error while acquiring segment writer.");
 			}
 		}
+		*/
+		Ok(Vec::new())
 	}
 
 }
@@ -135,6 +199,7 @@ impl SerializableSegment for SegmentWriter {
 	fn write(&self, mut serializer: SegmentSerializer) -> io::Result<()> {
 		try!(self.postings_writer.serialize(serializer.get_postings_serializer()));
 		try!(self.fast_field_writers.serialize(serializer.get_fast_field_serializer()));
-		serializer.close()
+		try!(serializer.close());
+		Ok(())
 	}
 }

src/lib.rs

@@ -23,6 +23,7 @@ extern crate serde;
 extern crate libc;
 extern crate lz4;
 extern crate uuid;
+extern crate num_cpus;
 
 #[cfg(test)] extern crate test;
 #[cfg(test)] extern crate rand;