Merge branch 'main' into next

src/model/garage.rs

@@ -124,7 +124,7 @@ impl Garage {
 				info!("Opening Sled database at: {}", db_path.display());
 				let db = db::sled_adapter::sled::Config::default()
 					.path(&db_path)
-					.cache_capacity(config.sled_cache_capacity)
+					.cache_capacity(config.sled_cache_capacity as u64)
 					.flush_every_ms(Some(config.sled_flush_every_ms))
 					.open()
 					.ok_or_message("Unable to open sled DB")?;
@@ -163,7 +163,10 @@ impl Garage {
 				info!("Opening LMDB database at: {}", db_path.display());
 				std::fs::create_dir_all(&db_path)
 					.ok_or_message("Unable to create LMDB data directory")?;
-				let map_size = garage_db::lmdb_adapter::recommended_map_size();
+				let map_size = match config.lmdb_map_size {
+					v if v == usize::default() => garage_db::lmdb_adapter::recommended_map_size(),
+					v => v - (v % 4096),
+				};
 
 				use db::lmdb_adapter::heed;
 				let mut env_builder = heed::EnvOpenOptions::new();
@@ -182,6 +185,7 @@ impl Garage {
 							"OutOfMemory error while trying to open LMDB database. This can happen \
 							if your operating system is not allowing you to use sufficient virtual \
 							memory address space. Please check that no limit is set (ulimit -v). \
+							You may also try to set a smaller `lmdb_map_size` configuration parameter. \
 							On 32-bit machines, you should probably switch to another database engine.".into()))
 					}
 					x => x.ok_or_message("Unable to open LMDB DB")?,

src/rpc/Cargo.toml

@@ -26,7 +26,7 @@ tracing = "0.1"
 rand = "0.8"
 itertools="0.10"
 sodiumoxide = { version = "0.2.5-0", package = "kuska-sodiumoxide" }
-systemstat = "0.2.3"
+nix = { version = "0.27", default-features = false, features = ["fs"] }
 
 async-trait = "0.1.7"
 serde = { version = "1.0", default-features = false, features = ["derive", "rc"] }

src/rpc/system.rs

@@ -896,46 +896,40 @@ impl NodeStatus {
 		data_dir: &DataDirEnum,
 		metrics: &SystemMetrics,
 	) {
-		use systemstat::{Platform, System};
-		let mounts = System::new().mounts().unwrap_or_default();
-
-		let mount_avail = |path: &Path| {
-			mounts
-				.iter()
-				.filter(|x| path.starts_with(&x.fs_mounted_on))
-				.max_by_key(|x| x.fs_mounted_on.len())
-				.map(|x| (x.avail.as_u64(), x.total.as_u64()))
+		use nix::sys::statvfs::statvfs;
+		let mount_avail = |path: &Path| match statvfs(path) {
+			Ok(x) => {
+				let avail = x.blocks_available() * x.fragment_size();
+				let total = x.blocks() * x.fragment_size();
+				Some((x.filesystem_id(), avail, total))
+			}
+			Err(_) => None,
 		};
 
-		self.meta_disk_avail = mount_avail(meta_dir);
+		self.meta_disk_avail = mount_avail(meta_dir).map(|(_, a, t)| (a, t));
 		self.data_disk_avail = match data_dir {
-			DataDirEnum::Single(dir) => mount_avail(dir),
-			DataDirEnum::Multiple(dirs) => {
-				// Take mounts corresponding to all specified data directories that
-				// can be used for writing data
-				let mounts = dirs
-					.iter()
-					.filter(|dir| dir.capacity.is_some())
-					.map(|dir| {
-						mounts
-							.iter()
-							.filter(|mnt| dir.path.starts_with(&mnt.fs_mounted_on))
-							.max_by_key(|mnt| mnt.fs_mounted_on.len())
-					})
-					.collect::<Vec<_>>();
-				if mounts.iter().any(|x| x.is_none()) {
-					None // could not get info for at least one mount
-				} else {
-					// dedup mounts in case several data directories are on the same filesystem
-					let mut mounts = mounts.iter().map(|x| x.unwrap()).collect::<Vec<_>>();
-					mounts.sort_by(|x, y| x.fs_mounted_on.cmp(&y.fs_mounted_on));
-					mounts.dedup_by(|x, y| x.fs_mounted_on == y.fs_mounted_on);
-					// calculate sum of available and total space
-					Some(mounts.iter().fold((0, 0), |(x, y), mnt| {
-						(x + mnt.avail.as_u64(), y + mnt.total.as_u64())
-					}))
+			DataDirEnum::Single(dir) => mount_avail(dir).map(|(_, a, t)| (a, t)),
+			DataDirEnum::Multiple(dirs) => (|| {
+				// TODO: more precise calculation that takes into account
+				// how data is going to be spread among partitions
+				let mut mounts = HashMap::new();
+				for dir in dirs.iter() {
+					if dir.capacity.is_none() {
+						continue;
+					}
+					match mount_avail(&dir.path) {
+						Some((fsid, avail, total)) => {
+							mounts.insert(fsid, (avail, total));
+						}
+						None => return None,
+					}
 				}
-			}
+				Some(
+					mounts
+						.into_iter()
+						.fold((0, 0), |(x, y), (_, (a, b))| (x + a, y + b)),
+				)
+			})(),
 		};
 
 		if let Some((avail, total)) = self.meta_disk_avail {

src/util/Cargo.toml

@@ -20,6 +20,7 @@ arc-swap = "1.0"
 async-trait = "0.1"
 blake2 = "0.10"
 bytes = "1.0"
+bytesize = "1.2"
 digest = "0.10"
 err-derive = "0.3"
 hexdump = "0.1"

src/util/config.rs

@@ -1,4 +1,5 @@
 //! Contains type and functions related to Garage configuration file
+use std::convert::TryFrom;
 use std::io::Read;
 use std::net::SocketAddr;
 use std::path::PathBuf;
@@ -23,7 +24,10 @@ pub struct Config {
 	pub data_fsync: bool,
 
 	/// Size of data blocks to save to disk
-	#[serde(default = "default_block_size")]
+	#[serde(
+		deserialize_with = "deserialize_capacity",
+		default = "default_block_size"
+	)]
 	pub block_size: usize,
 
 	/// Replication mode. Supported values:
@@ -73,12 +77,19 @@ pub struct Config {
 	pub db_engine: String,
 
 	/// Sled cache size, in bytes
-	#[serde(default = "default_sled_cache_capacity")]
-	pub sled_cache_capacity: u64,
+	#[serde(
+		deserialize_with = "deserialize_capacity",
+		default = "default_sled_cache_capacity"
+	)]
+	pub sled_cache_capacity: usize,
 	/// Sled flush interval in milliseconds
 	#[serde(default = "default_sled_flush_every_ms")]
 	pub sled_flush_every_ms: u64,
 
+	/// LMDB map size
+	#[serde(deserialize_with = "deserialize_capacity", default)]
+	pub lmdb_map_size: usize,
+
 	// -- APIs
 	/// Configuration for S3 api
 	pub s3_api: S3ApiConfig,
@@ -213,7 +224,7 @@ fn default_db_engine() -> String {
 	"lmdb".into()
 }
 
-fn default_sled_cache_capacity() -> u64 {
+fn default_sled_cache_capacity() -> usize {
 	128 * 1024 * 1024
 }
 fn default_sled_flush_every_ms() -> u64 {
@@ -293,8 +304,6 @@ fn deserialize_compression<'de, D>(deserializer: D) -> Result<Option<i32>, D::Er
 where
 	D: de::Deserializer<'de>,
 {
-	use std::convert::TryFrom;
-
 	struct OptionVisitor;
 
 	impl<'de> serde::de::Visitor<'de> for OptionVisitor {
@@ -339,6 +348,50 @@ where
 	deserializer.deserialize_any(OptionVisitor)
 }
 
+fn deserialize_capacity<'de, D>(deserializer: D) -> Result<usize, D::Error>
+where
+	D: de::Deserializer<'de>,
+{
+	struct CapacityVisitor;
+
+	impl<'de> serde::de::Visitor<'de> for CapacityVisitor {
+		type Value = usize;
+		fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
+			formatter.write_str("int or '<capacity>'")
+		}
+
+		fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
+		where
+			E: de::Error,
+		{
+			value
+				.parse::<bytesize::ByteSize>()
+				.map(|x| x.as_u64())
+				.map_err(|e| E::custom(format!("invalid capacity value: {}", e)))
+				.and_then(|v| {
+					usize::try_from(v)
+						.map_err(|_| E::custom("capacity value out of bound".to_owned()))
+				})
+		}
+
+		fn visit_i64<E>(self, v: i64) -> Result<Self::Value, E>
+		where
+			E: de::Error,
+		{
+			usize::try_from(v).map_err(|_| E::custom("capacity value out of bound".to_owned()))
+		}
+
+		fn visit_u64<E>(self, v: u64) -> Result<Self::Value, E>
+		where
+			E: de::Error,
+		{
+			usize::try_from(v).map_err(|_| E::custom("capacity value out of bound".to_owned()))
+		}
+	}
+
+	deserializer.deserialize_any(CapacityVisitor)
+}
+
 #[cfg(test)]
 mod tests {
 	use crate::error::Error;