From 06d255b9c9942c9bb2abf2b0550bc0cef1d3e5c1 Mon Sep 17 00:00:00 2001 From: Paul Masurel Date: Tue, 29 Mar 2016 10:46:12 +0900 Subject: [PATCH] Naive multithreading. Performance is already there. --- Cargo.toml | 1 + src/core/directory.rs | 15 ++++--- src/core/index.rs | 9 +++- src/core/writer.rs | 95 ++++++++++++++++++++++++++++++++++++------- src/lib.rs | 1 + 5 files changed, 96 insertions(+), 25 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 12c0d2157..a68030cc4 100644 --- a/Cargo.toml +++ b/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" diff --git a/src/core/directory.rs b/src/core/directory.rs index d9d1322ae..2f4d02a4c 100644 --- a/src/core/directory.rs +++ b/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; // CannotCreateTempDirectory(io::Error), // } -pub trait Directory: fmt::Debug { +pub trait Directory: fmt::Debug + Send + Sync { fn open_read(&self, path: &Path) -> io::Result; fn open_write(&mut self, path: &Path) -> io::Result; 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>, + mmap_cache: RwLock>, _temp_directory: Option, } @@ -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 { 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 { 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) => { diff --git a/src/core/index.rs b/src/core/index.rs index 440b2b4e8..2e0d0ebe2 100644 --- a/src/core/index.rs +++ b/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::open(self, num_threads) + } + pub fn writer(&self,) -> io::Result { - IndexWriter::open(self,) + self.writer_with_num_threads(num_cpus::get()) } pub fn searcher(&self,) -> io::Result { @@ -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[..]) diff --git a/src/core/writer.rs b/src/core/writer.rs index c5d9d32ee..48a228919 100644 --- a/src/core/writer.rs +++ b/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, - directory: Index, + // segment_writers: Vec, + threads: Vec>, + index: Index, schema: Schema, + queue_input: SyncSender, } +type ArcDoc = Arc; + impl IndexWriter { - pub fn open(directory: &Index) -> io::Result { - 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 { + let schema = index.schema(); + let (queue_input, queue_output): (SyncSender, Receiver) = 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 { + pub fn commit(&mut self,) -> io::Result> { + /* 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(()) } } diff --git a/src/lib.rs b/src/lib.rs index 991c65ebb..5f5f1150b 100644 --- a/src/lib.rs +++ b/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;