Rewrite in Rust to obtain standalone static binary
In contradiction with Jean-Paul's guidelines on not using Rust due to lack of knowledge about it inside Nexedi, I am using it here because it is the fastest way for me to get a working standalone static binary, I know that language best. Considering we must be getting results ASAP, this is the best strategy for me. We may later rewrite it in another language if necessary. A shell script is included to build the static binary, you need to install rustup to get rust for musl, an alternative libc that allows to create real static binaries that embed libc itself too. Rustup can be found at: https://rustup.rs/ You can get a musl toolchain with: $ rustup target add x86_64-unknown-linux-musl The acl library is being downloaded and built as a static library by the script, and the rust build system will also build a vendored copy of openssl as a static library. Parallel hashing is done a bit differently in that Rust version, only files contained in the currently processed directories will be hashed in parallel. If there is a single big file in a directory hashing will be stuck on that file until it's done and it goes onto the next directory. To clarify, each file is only hashed on a single thread, the Python version also does this, it just keeps the number of files being hashed in parallel to a constant number as long as there is more files to process, this version will only hash with one thread per file in the currently processed directory. It was done that way for sake of simplicity but we can implement an offload threadpool to mimick what was done in Python later on.
| [package] | ||
| name = "metadata-collect-agent" | ||
| version = "0.1.0" | ||
| authors = ["Leo Le Bouter <leo.le.bouter@nexedi.com>"] | ||
| edition = "2018" | ||
| [dependencies] | ||
| posix-acl = "1.0.0" | ||
| xattr = "0.2.2" | ||
| md-5 = "0.9.1" | ||
| sha-1 = "0.9.1" | ||
| sha2 = "0.9.1" | ||
| hex = "0.4.2" | ||
| anyhow = "1.0.32" | ||
| clap = "2.33.3" | ||
| psutil = { git = "https://github.com/leo-lb/rust-psutil", branch = "lle-bout/impl-serde", version = "3.1.0", features = ["serde"] } | ||
| reqwest = { version = "0.10.7", features = ["blocking", "native-tls-vendored"] } | ||
| rmp-serde = "0.14.4" | ||
| nix = "0.18.0" | ||
| serde = { version = "1.0.115", features = ["derive"] } | ||
| base64 = "0.12.3" | ||
| rayon = "1.3.1" | ||
| [profile.release] | ||
| opt-level = 'z' | ||
| lto = true | ||
| codegen-units = 1 | ||
| \ No newline at end of file |
| use anyhow::Result; | ||
| use blocking::multipart; | ||
| use clap::{App, Arg}; | ||
| use multipart::Part; | ||
| use rayon::prelude::*; | ||
| use reqwest::{ | ||
| blocking::{self, Client}, | ||
| header, redirect, | ||
| }; | ||
| use serde::Serialize; | ||
| use std::collections::HashMap; | ||
| use std::{ | ||
| ffi::OsString, | ||
| fs::DirEntry, | ||
| io::{Cursor, Read}, | ||
| path::PathBuf, | ||
| sync::{Arc, Mutex}, | ||
| thread, | ||
| time::Duration, | ||
| }; | ||
| #[derive(Debug, Serialize)] | ||
| struct FileStat { | ||
| st_dev: u64, | ||
| st_ino: u64, | ||
| st_nlink: u64, | ||
| st_mode: u32, | ||
| st_uid: u32, | ||
| st_gid: u32, | ||
| st_rdev: u64, | ||
| st_size: i64, | ||
| st_blksize: i64, | ||
| st_blocks: i64, | ||
| st_atime: i64, | ||
| st_atime_nsec: i64, | ||
| st_mtime: i64, | ||
| st_mtime_nsec: i64, | ||
| st_ctime: i64, | ||
| st_ctime_nsec: i64, | ||
| } | ||
| impl From<nix::sys::stat::FileStat> for FileStat { | ||
| fn from(x: nix::sys::stat::FileStat) -> Self { | ||
| Self { | ||
| st_dev: x.st_dev, | ||
| st_ino: x.st_ino, | ||
| st_nlink: x.st_nlink, | ||
| st_mode: x.st_mode, | ||
| st_uid: x.st_uid, | ||
| st_gid: x.st_gid, | ||
| st_rdev: x.st_rdev, | ||
| st_size: x.st_size, | ||
| st_blksize: x.st_blksize, | ||
| st_blocks: x.st_blocks, | ||
| st_atime: x.st_atime, | ||
| st_atime_nsec: x.st_atime_nsec, | ||
| st_mtime: x.st_mtime, | ||
| st_mtime_nsec: x.st_mtime_nsec, | ||
| st_ctime: x.st_ctime, | ||
| st_ctime_nsec: x.st_ctime_nsec, | ||
| } | ||
| } | ||
| } | ||
| #[derive(Default, Debug, Serialize)] | ||
| struct Tree { | ||
| childs: HashMap<OsString, Tree>, | ||
| stat: Option<FileStat>, | ||
| xattrs: HashMap<OsString, Option<Vec<u8>>>, | ||
| posix_acls: Option<String>, | ||
| ignored: bool, | ||
| symlink_target: Option<OsString>, | ||
| md5: Option<String>, | ||
| sha1: Option<String>, | ||
| sha256: Option<String>, | ||
| sha512: Option<String>, | ||
| } | ||
| #[derive(Serialize)] | ||
| struct Snapshot { | ||
| disk_partitions: Vec<psutil::disk::Partition>, | ||
| fs_tree: Tree, | ||
| } | ||
| fn construct_fs_tree( | ||
| cur_tree: Option<Tree>, | ||
| path: &PathBuf, | ||
| dev_whitelist: &Vec<u64>, | ||
| ignored_dirs: &Vec<PathBuf>, | ||
| ) -> Result<Tree> { | ||
| let mut cur_tree = match cur_tree { | ||
| Some(cur_tree) => cur_tree, | ||
| None => Tree { | ||
| stat: Some(nix::sys::stat::lstat(path)?.into()), | ||
| ..Tree::default() | ||
| }, | ||
| }; | ||
| if !dev_whitelist.iter().any(|x| match &cur_tree.stat { | ||
| Some(stat) if stat.st_dev == *x => true, | ||
| _ => false, | ||
| }) { | ||
| return Ok(cur_tree); | ||
| } | ||
| if ignored_dirs.iter().any(|x| path.starts_with(x)) { | ||
| cur_tree.ignored = true; | ||
| return Ok(cur_tree); | ||
| } | ||
| let entries: Vec<Result<DirEntry, _>> = match std::fs::read_dir(&path) { | ||
| Ok(x) => x, | ||
| _ => return Ok(cur_tree), | ||
| } | ||
| .collect(); | ||
| let cur_tree = { | ||
| let cur_tree = Arc::new(Mutex::new(cur_tree)); | ||
| entries.par_iter().for_each(|entry| match entry { | ||
| Ok(entry) => { | ||
| let mut tree = Tree { | ||
| stat: match nix::sys::stat::lstat(&entry.path()) { | ||
| Ok(s) => Some(s.into()), | ||
| _ => None, | ||
| }, | ||
| posix_acls: match posix_acl::PosixACL::read_acl(&entry.path()) { | ||
| Ok(acl) => Some(acl.as_text()), | ||
| _ => None, | ||
| }, | ||
| xattrs: match xattr::list(&entry.path()) { | ||
| Ok(xattrs) => xattrs | ||
| .filter_map(|name| match xattr::get(&entry.path(), &name) { | ||
| Ok(xattr) => Some((name, xattr)), | ||
| _ => None, | ||
| }) | ||
| .collect(), | ||
| _ => HashMap::new(), | ||
| }, | ||
| ..Tree::default() | ||
| }; | ||
| match entry.file_type() { | ||
| Ok(file_type) if file_type.is_dir() => { | ||
| tree = construct_fs_tree( | ||
| Some(tree), | ||
| &entry.path(), | ||
| dev_whitelist, | ||
| ignored_dirs, | ||
| ) | ||
| .unwrap(); | ||
| } | ||
| Ok(file_type) if file_type.is_file() => { | ||
| if let Ok(mut file) = std::fs::File::open(&entry.path()) { | ||
| use md5::{Digest, Md5}; | ||
| use sha1::Sha1; | ||
| use sha2::{Sha256, Sha512}; | ||
| let mut md5 = Md5::new(); | ||
| let mut sha1 = Sha1::new(); | ||
| let mut sha256 = Sha256::new(); | ||
| let mut sha512 = Sha512::new(); | ||
| let buf: &mut [u8] = &mut [0; 8 * 1024]; | ||
| loop { | ||
| match file.read(buf) { | ||
| Ok(0) => { | ||
| tree.md5 = Some(hex::encode(md5.finalize())); | ||
| tree.sha1 = Some(hex::encode(sha1.finalize())); | ||
| tree.sha256 = Some(hex::encode(sha256.finalize())); | ||
| tree.sha512 = Some(hex::encode(sha512.finalize())); | ||
| break; | ||
| } | ||
| Ok(n) => { | ||
| md5.update(&buf[0..n - 1]); | ||
| sha1.update(&buf[0..n - 1]); | ||
| sha256.update(&buf[0..n - 1]); | ||
| sha512.update(&buf[0..n - 1]); | ||
| } | ||
| Err(e) if e.kind() == std::io::ErrorKind::Interrupted => { | ||
| continue; | ||
| } | ||
| Err(e) => { | ||
| eprintln!("{:#?}", e); | ||
| break; | ||
| } | ||
| }; | ||
| } | ||
| } | ||
| } | ||
| Ok(file_type) if file_type.is_symlink() => { | ||
| tree.symlink_target = std::fs::read_link(&entry.path()) | ||
| .map(|x| x.as_os_str().into()) | ||
| .ok(); | ||
| } | ||
| _ => {} | ||
| }; | ||
| { | ||
| let mut locked = cur_tree.lock().unwrap(); | ||
| locked.childs.insert(entry.file_name(), tree); | ||
| } | ||
| } | ||
| _ => {} | ||
| }); | ||
| Arc::try_unwrap(cur_tree).unwrap().into_inner().unwrap() | ||
| }; | ||
| Ok(cur_tree) | ||
| } | ||
| fn upload_to_erp5<R: Read + Send + 'static>( | ||
| reader: R, | ||
| size: u64, | ||
| base_url: &str, | ||
| username: &str, | ||
| password: &str, | ||
| reference: &str, | ||
| ) -> Result<()> { | ||
| let mut headers = header::HeaderMap::new(); | ||
| headers.insert( | ||
| header::AUTHORIZATION, | ||
| header::HeaderValue::from_str(&format!( | ||
| "Basic {}", | ||
| &base64::encode(&format!("{}:{}", username, password)) | ||
| ))?, | ||
|
||
| ); | ||
| let client = Client::builder() | ||
| .redirect(redirect::Policy::none()) | ||
| .default_headers(headers) | ||
| .build()?; | ||
| let resp = client | ||
| .get(&format!("{}/portal_contributions/newContent", base_url)) | ||
| .query(&[ | ||
| ("portal_type", "File"), | ||
| ("filename", &format!("{}.metadata", &reference)), | ||
| ("container_path", "document_module"), | ||
|
||
| ("data", ""), | ||
| ]) | ||
| .send()?; | ||
| // workaround for ERP5 disappearing documents race condition bug | ||
| thread::sleep(Duration::from_secs(3)); | ||
| let doc_url: reqwest::Url = resp | ||
| .headers() | ||
| .get("X-Location") | ||
| .ok_or_else(|| anyhow::format_err!("X-Location must exist"))? | ||
| .to_str()? | ||
| .parse()?; | ||
| let resp = client | ||
| .post(doc_url.to_owned()) | ||
| .multipart( | ||
| multipart::Form::new() | ||
| .text("form_id", "File_view") | ||
| .text("object_path", doc_url.path().to_owned()) | ||
| .text("field_my_reference", reference.to_owned()) | ||
| .part( | ||
| "field_my_file", | ||
| Part::reader_with_length(reader, size) | ||
| .file_name(format!("{}.metadata", &reference)), | ||
| ) | ||
| .text("Base_edit:method", ""), | ||
| ) | ||
| .send()?; | ||
| // workaround for ERP5 disappearing documents race condition bug | ||
| thread::sleep(Duration::from_secs(3)); | ||
| let mut params = HashMap::new(); | ||
| params.insert("dialog_id", "Base_viewWorkflowActionDialog"); | ||
|
||
| params.insert("dialog_method", "Workflow_statusModify"); | ||
| params.insert("form_id", "File_view"); | ||
| params.insert("object_path", doc_url.path()); | ||
| params.insert("field_your_comment", ""); | ||
| params.insert("field_your_workflow_action", "share_action"); | ||
| params.insert("Base_callDialogMethod:method", ""); | ||
| let resp = client.post(doc_url.to_owned()).form(¶ms).send()?; | ||
| // workaround for ERP5 disappearing documents race condition bug | ||
| thread::sleep(Duration::from_secs(3)); | ||
| Ok(()) | ||
| } | ||
| fn main() -> Result<()> { | ||
| let m = App::new("metadata-collect-agent") | ||
| .args(&[ | ||
| Arg::with_name("start_directory") | ||
| .required(true) | ||
| .takes_value(true) | ||
| .multiple(false), | ||
| Arg::with_name("erp5-user") | ||
| .takes_value(true) | ||
| .multiple(false) | ||
| .required(true), | ||
| Arg::with_name("erp5-pass") | ||
| .takes_value(true) | ||
| .multiple(false) | ||
| .required(true), | ||
| Arg::with_name("erp5-file-reference") | ||
| .takes_value(true) | ||
| .multiple(false) | ||
| .required(true), | ||
| Arg::with_name("erp5-base-url") | ||
| .takes_value(true) | ||
| .multiple(false) | ||
| .required(true), | ||
| Arg::with_name("ignored-dirs") | ||
| .takes_value(true) | ||
| .multiple(true) | ||
| .required(false), | ||
| ]) | ||
| .get_matches(); | ||
| let ignored_dirs = m | ||
| .values_of("ignored-dirs") | ||
| .unwrap_or(clap::Values::default()) | ||
| .map(|v| PathBuf::from(v)) | ||
| .collect(); | ||
| let disk_partitions = psutil::disk::partitions_physical()?; | ||
| let dev_whitelist = disk_partitions | ||
| .iter() | ||
| .map(|p| nix::sys::stat::lstat(p.mountpoint()).unwrap().st_dev) | ||
| .collect(); | ||
| let snapshot = Snapshot { | ||
| disk_partitions, | ||
| fs_tree: construct_fs_tree( | ||
| None, | ||
| &PathBuf::from(m.value_of("start_directory").unwrap()), | ||
| &dev_whitelist, | ||
| &ignored_dirs, | ||
| )?, | ||
| }; | ||
| let packed = rmp_serde::encode::to_vec_named(&snapshot)?; | ||
| let packed_size = packed.len() as u64; | ||
| upload_to_erp5( | ||
| Cursor::new(packed), | ||
| packed_size, | ||
| m.value_of("erp5-base-url").unwrap(), | ||
| m.value_of("erp5-user").unwrap(), | ||
| m.value_of("erp5-pass").unwrap(), | ||
| m.value_of("erp5-file-reference").unwrap(), | ||
| )?; | ||
| Ok(()) | ||
| } | ||
-
OwnerI'm halfway through reading https://doc.rust-lang.org/stable/book/ and I have to say rust is really excellent.
-
Owner
BTW, in slapos!799 (merged) there's the beginning of rust component for slapos (which takes ~2 hours to compile from source and I'm not really sure it's reproductible, the setup seems to download things) and rust support in theia.
-
Owner
Hashing several files in parallel looks like a bad idea. We already did it in SlapOS for the resilience and it was a disaster. It's rare that hashing is slower than IO (maybe only a hashing a big file on high-performance NVMe with the slowest algorithm). Most of the time, it's so inefficient that it's slower: at best, it could be faster but the hardware consumes a lot.
But there may other to parallelizing work. First, by doing IO and hashing in separate threads, i.e. pipelining. Since you compute several hashes, you could use 1 thread per hash.
In any case, I find such attempt to optimize very premature, in particular if it is planned to rewrite.
-
@jm It's really really slow with a single thread, the use case is whole file system hashing, and it's performed during boot with nothing else running on the system, it's mandatory that it is fast, otherwise we might end up with terrible dozens of minutes boot times. I did some tests before making it parallel and it made it much faster and I/O utilization is 100% now on my fast NVMe drive so it can't be any faster, but before it wasnt 100%, so the first benefit is from doing I/O from multiple threads but then on my system hashing on a single thread is as slow or a bit faster than I/O on a single thread for something like 100MB/s, so you would have to parallelize anyway because it wouldnt keep up hashing data from I/O being done on multiple threads on one thread.
The hashing functions come from: https://github.com/RustCrypto/hashes - maybe they could be optimized but they already use hand-written assembly in select places for performance so it turns out making it parallel is easier for me than digging into the assembly with cryptographer knowledge that I don't have.
Note that the performance issues were identical with Python that uses OpenSSL where hash functions are already well optimized there.
In any case, I find such attempt to optimize very premature, in particular if it is planned to rewrite.
It's pretty trivial in Rust, add Arc/Mutex on data structure then use Rayon's parallel iterator and it's done, so it wasnt much effort and time spent there.