From 8b3f5a84787c2f1f971c4346324629b52ee5aa15 Mon Sep 17 00:00:00 2001 From: Emily Toop Date: Wed, 16 May 2018 17:29:27 +0100 Subject: [PATCH] First pass at Stores --- src/conn.rs | 9 +++--- src/lib.rs | 1 + src/stores.rs | 82 ++++++++++++++++++++++++++++++++++++++++++++------- 3 files changed, 77 insertions(+), 15 deletions(-) diff --git a/src/conn.rs b/src/conn.rs index 24a84499..4f76db2c 100644 --- a/src/conn.rs +++ b/src/conn.rs @@ -115,6 +115,7 @@ use query::{ /// changing parts (schema) that we want to share across threads. /// /// See https://github.com/mozilla/mentat/wiki/Thoughts:-modeling-db-conn-in-Rust. +#[derive(Clone)] pub struct Metadata { pub generation: u64, pub partition_map: PartitionMap, @@ -135,6 +136,7 @@ impl Metadata { } /// A mutable, safe reference to the current Mentat store. +#[derive(Clone)] pub struct Conn { /// `Mutex` since all reads and writes need to be exclusive. Internally, owned data for the /// volatile parts (generation and partition map), and `Arc` for the infrequently changing parts @@ -151,13 +153,12 @@ pub struct Conn { /// gives us copy-on-write semantics. /// We store that cached `Arc` here in a `Mutex`, so that the main copy can be carefully /// replaced on commit. - metadata: Mutex, + metadata: Arc>, // TODO: maintain set of change listeners or handles to transaction report queues. #298. // TODO: maintain cache of query plans that could be shared across threads and invalidated when // the schema changes. #315. - tx_observer_service: Mutex, pub(crate) tx_observer_service: Arc>, } @@ -572,8 +573,8 @@ impl Conn { // Intentionally not public. fn new(partition_map: PartitionMap, schema: Schema) -> Conn { Conn { - metadata: Mutex::new(Metadata::new(0, partition_map, Arc::new(schema), Default::default())), - tx_observer_service: Mutex::new(TxObservationService::new()), + metadata: Arc::new(Mutex::new(Metadata::new(0, partition_map, Arc::new(schema), Default::default()))), + tx_observer_service: Arc::new(Mutex::new(TxObservationService::new())), } } diff --git a/src/lib.rs b/src/lib.rs index d2f5eff9..ac7b3776 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -137,6 +137,7 @@ pub use conn::{ pub use stores::{ Store, + Stores, }; #[cfg(test)] diff --git a/src/stores.rs b/src/stores.rs index ce6f7d49..2a745b29 100644 --- a/src/stores.rs +++ b/src/stores.rs @@ -16,6 +16,7 @@ use std::collections::{ use std::path::{ Path, + PathBuf, }; use std::sync::{ @@ -62,10 +63,66 @@ use query::{ QueryOutput, }; +pub struct Stores { + stores: BTreeMap, +} + +impl Stores { + fn new() -> Stores { + Stores { + stores: Default::default(), + } + } +} + +impl Stores { + fn is_open(&self, path: PathBuf) -> bool { + self.stores.contains_key(&path) + } + + pub fn open<'p, P>(&mut self, path: P) -> Result<&mut Store> where P: Into<&'p Path> { + let path = path.into(); + let canonical = path.canonicalize()?; + Ok(self.stores.entry(canonical).or_insert(Store::open(path.to_str().unwrap())?)) + } + + pub fn get<'p, P>(&mut self, path: P) -> Result> where P: Into<&'p Path> { + let canonical = path.into().canonicalize()?; + Ok(self.stores.get(&canonical)) + } + + pub fn get_mut<'p, P>(&mut self, path: P) -> Result> where P: Into<&'p Path> { + let canonical = path.into().canonicalize()?; + Ok(self.stores.get_mut(&canonical)) + } + + pub fn connect<'p, P>(&mut self, path: P) -> Result where P: Into<&'p Path> { + let path = path.into(); + let canonical = path.canonicalize()?; + let store = self.stores.get_mut(&canonical).unwrap(); + let connection = ::new_connection(path.to_str().unwrap())?; + Ok(store.fork(connection)) + } + + fn open_connections_for_store(&self, path: &PathBuf) -> usize { + Arc::strong_count(self.stores.get(path).unwrap().conn()) + } + + pub fn close<'p, P>(&mut self, path: P) -> Result<()> where P: Into<&'p Path> { + let canonical = path.into().canonicalize()?; + if self.open_connections_for_store(&canonical) <= 1 { + self.stores.remove(&canonical); + Ok(()) + } else { + Ok(()) + } + } +} + /// A convenience wrapper around a single SQLite connection and a Conn. This is suitable /// for applications that don't require complex connection management. pub struct Store { - conn: Conn, + conn: Arc, sqlite: rusqlite::Connection, } @@ -75,11 +132,15 @@ impl Store { let mut connection = ::new_connection(path)?; let conn = Conn::connect(&mut connection)?; Ok(Store { - conn: conn, + conn: Arc::new(conn), sqlite: connection, }) } + pub fn open_in_memory() -> Result { + Store::open("") + } + /// Returns a totally blank store with no bootstrap schema. Use `open` instead. pub fn open_empty(path: &str) -> Result { if !path.is_empty() { @@ -91,7 +152,7 @@ impl Store { let mut connection = ::new_connection(path)?; let conn = Conn::empty(&mut connection)?; Ok(Store { - conn: conn, + conn: Arc::new(conn), sqlite: connection, }) } @@ -124,25 +185,25 @@ impl Store { } } - pub fn dismantle(self) -> (rusqlite::Connection, Conn) { + pub fn dismantle(self) -> (rusqlite::Connection, Arc) { (self.sqlite, self.conn) } - pub fn conn(&self) -> &Conn { + pub fn conn(&self) -> &Arc { &self.conn } pub fn begin_read<'m>(&'m mut self) -> Result> { - self.conn.begin_read(&mut self.sqlite) + Arc::make_mut(&mut self.conn).begin_read(&mut self.sqlite) } pub fn begin_transaction<'m>(&'m mut self) -> Result> { - self.conn.begin_transaction(&mut self.sqlite) + Arc::make_mut(&mut self.conn).begin_transaction(&mut self.sqlite) } pub fn cache(&mut self, attr: &Keyword, direction: CacheDirection) -> Result<()> { let schema = &self.conn.current_schema(); - self.conn.cache(&mut self.sqlite, + Arc::make_mut(&mut self.conn).cache(&mut self.sqlite, schema, attr, direction, @@ -150,11 +211,11 @@ impl Store { } pub fn register_observer(&mut self, key: String, observer: Arc) { - self.conn.register_observer(key, observer); + Arc::make_mut(&mut self.conn).register_observer(key, observer); } pub fn unregister_observer(&mut self, key: &String) { - self.conn.unregister_observer(key); + Arc::make_mut(&mut self.conn).unregister_observer(key); } } @@ -232,7 +293,6 @@ mod tests { use mentat_core::{ CachedAttributes, - HasSchema, TypedValue, ValueType, };