Dynamic LMDB mapsize allocation [1.1.0]

store/src/lib.rs

@@ -31,7 +31,7 @@ extern crate grin_core as core;
 //use grin_core as core;
 
 pub mod leaf_set;
-mod lmdb;
+pub mod lmdb;
 pub mod pmmr;
 pub mod prune_list;
 pub mod types;

store/src/lmdb.rs

@@ -24,6 +24,10 @@ use lmdb_zero::LmdbResultExt;
 
 use crate::core::ser;
 
+/// number of bytes to grow the database by when needed
+pub const ALLOC_CHUNK_SIZE: usize = 134_217_728; //128 MB
+const RESIZE_PERCENT: f32 = 0.9;
+
 /// Main error type for this lmdb
 #[derive(Clone, Eq, PartialEq, Debug, Fail)]
 pub enum Error {
@@ -71,26 +75,13 @@ pub fn new_named_env(path: String, name: String, max_readers: Option<u32>) -> lm
 
 	let mut env_builder = lmdb::EnvBuilder::new().unwrap();
 	env_builder.set_maxdbs(8).unwrap();
-	// half a TB should give us plenty room, will be an issue on 32 bits
-	// (which we don't support anyway)
-
-	#[cfg(not(target_os = "windows"))]
-	env_builder.set_mapsize(5_368_709_120).unwrap_or_else(|e| {
-		panic!("Unable to allocate LMDB space: {:?}", e);
-	});
-	//TODO: This is temporary to support (beta) windows support
-	//Windows allocates the entire file at once, so this needs to
-	//be changed to allocate as little as possible and increase as needed
-	#[cfg(target_os = "windows")]
-	env_builder.set_mapsize(524_288_000).unwrap_or_else(|e| {
-		panic!("Unable to allocate LMDB space: {:?}", e);
-	});
 
 	if let Some(max_readers) = max_readers {
 		env_builder
 			.set_maxreaders(max_readers)
 			.expect("Unable set max_readers");
 	}
+
 	unsafe {
 		env_builder
 			.open(&full_path, lmdb::open::NOTLS, 0o600)
@@ -120,6 +111,53 @@ impl Store {
 		Store { env, db }
 	}
 
+	/// Determines whether the environment needs a resize based on a simple percentage threshold
+	pub fn needs_resize(&self) -> Result<bool, Error> {
+		let env_info = self.env.info()?;
+		let stat = self.env.stat()?;
+
+		let size_used = stat.psize as usize * env_info.last_pgno;
+		trace!("DB map size: {}", env_info.mapsize);
+		trace!("Space used: {}", size_used);
+		trace!("Space remaining: {}", env_info.mapsize - size_used);
+		let resize_percent = RESIZE_PERCENT;
+		trace!(
+			"Percent used: {:.*}  Percent threshold: {:.*}",
+			4,
+			size_used as f64 / env_info.mapsize as f64,
+			4,
+			resize_percent
+		);
+
+		if size_used as f32 / env_info.mapsize as f32 > resize_percent
+			|| env_info.mapsize < ALLOC_CHUNK_SIZE
+		{
+			trace!("Resize threshold met (percent-based)");
+			Ok(true)
+		} else {
+			trace!("Resize threshold not met (percent-based)");
+			Ok(false)
+		}
+	}
+
+	/// Increments the database size by one ALLOC_CHUNK_SIZE
+	pub fn do_resize(&self) -> Result<(), Error> {
+		let env_info = self.env.info()?;
+		let new_mapsize = if env_info.mapsize < ALLOC_CHUNK_SIZE {
+			ALLOC_CHUNK_SIZE
+		} else {
+			env_info.mapsize + ALLOC_CHUNK_SIZE
+		};
+		unsafe {
+			self.env.set_mapsize(new_mapsize)?;
+		}
+		info!(
+			"Resized database from {} to {}",
+			env_info.mapsize, new_mapsize
+		);
+		Ok(())
+	}
+
 	/// Gets a value from the db, provided its key
 	pub fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Error> {
 		let txn = lmdb::ReadTransaction::new(self.env.clone())?;
@@ -178,6 +216,10 @@ impl Store {
 
 	/// Builds a new batch to be used with this store.
 	pub fn batch(&self) -> Result<Batch<'_>, Error> {
+		// check if the db needs resizing before returning the batch
+		if self.needs_resize()? {
+			self.do_resize()?;
+		}
 		let txn = lmdb::WriteTransaction::new(self.env.clone())?;
 		Ok(Batch {
 			store: self,

store/tests/lmdb.rs

@@ -0,0 +1,107 @@
+// Copyright 2018 The Grin Developers
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use grin_store as store;
+use grin_util as util;
+
+use grin_core::ser::{self, Readable, Reader, Writeable, Writer};
+
+use std::fs;
+use std::sync::Arc;
+
+const WRITE_CHUNK_SIZE: usize = 20;
+const TEST_ALLOC_SIZE: usize = store::lmdb::ALLOC_CHUNK_SIZE / 8 / WRITE_CHUNK_SIZE;
+
+#[derive(Clone)]
+struct PhatChunkStruct {
+	phatness: u64,
+}
+
+impl PhatChunkStruct {
+	/// create
+	pub fn new() -> PhatChunkStruct {
+		PhatChunkStruct { phatness: 0 }
+	}
+}
+
+impl Readable for PhatChunkStruct {
+	fn read(reader: &mut Reader) -> Result<PhatChunkStruct, ser::Error> {
+		let mut retval = PhatChunkStruct::new();
+		for _ in 0..TEST_ALLOC_SIZE {
+			retval.phatness = reader.read_u64()?;
+		}
+		Ok(retval)
+	}
+}
+
+impl Writeable for PhatChunkStruct {
+	fn write<W: Writer>(&self, writer: &mut W) -> Result<(), ser::Error> {
+		// write many times
+		for _ in 0..TEST_ALLOC_SIZE {
+			writer.write_u64(self.phatness)?;
+		}
+		Ok(())
+	}
+}
+
+fn clean_output_dir(test_dir: &str) {
+	let _ = fs::remove_dir_all(test_dir);
+}
+
+fn setup(test_dir: &str) {
+	util::init_test_logger();
+	clean_output_dir(test_dir);
+}
+
+#[test]
+fn lmdb_allocate() -> Result<(), store::Error> {
+	let test_dir = "test_output/lmdb_allocate";
+	setup(test_dir);
+	// Allocate more than the initial chunk, ensuring
+	// the DB resizes underneath
+	{
+		let env = Arc::new(store::new_env(test_dir.to_owned()));
+		let store = store::Store::open(env.clone(), "test1");
+
+		for i in 0..WRITE_CHUNK_SIZE * 2 {
+			println!("Allocating chunk: {}", i);
+			let chunk = PhatChunkStruct::new();
+			let mut key_val = format!("phat_chunk_set_1_{}", i).as_bytes().to_vec();
+			let batch = store.batch()?;
+			let key = store::to_key(b'P', &mut key_val);
+			batch.put_ser(&key, &chunk)?;
+			batch.commit()?;
+		}
+	}
+	// Open env again and keep adding
+	{
+		println!("***********************************");
+		println!("***************NEXT*****************");
+		println!("***********************************");
+		let env = Arc::new(store::new_env(test_dir.to_owned()));
+		let store = store::Store::open(env.clone(), "test1");
+
+		for i in 0..WRITE_CHUNK_SIZE * 2 {
+			println!("Allocating chunk: {}", i);
+			let chunk = PhatChunkStruct::new();
+			let mut key_val = format!("phat_chunk_set_2_{}", i).as_bytes().to_vec();
+			let batch = store.batch()?;
+			let key = store::to_key(b'P', &mut key_val);
+			batch.put_ser(&key, &chunk)?;
+			batch.commit()?;
+		}
+	}
+
+	Ok(())
+}