heimoshuiyu/rust-synapse-compress-state
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

All printing replaced by logging (#67)

Browse Source
This commit is contained in:
Azrenbeth
2021-09-27 12:20:12 +01:00
committed by GitHub
parent bc30942e2d
commit b8e323c321
9 changed files with 66 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
src/database.rs
View File

@@ -13,6 +13,7 @@
// limitations under the License.
use indicatif::{ProgressBar, ProgressStyle};
use log::{debug, trace};
use openssl::ssl::{SslConnector, SslMethod, SslVerifyMode};
use postgres::{fallible_iterator::FallibleIterator, types::ToSql, Client};
use postgres_openssl::MakeTlsConnector;
@@ -40,7 +41,6 @@ use super::StateGroupEntry;
/// * `max_state_group` - If specified, then only fetch the entries for state
/// groups lower than or equal to this number.
/// * 'groups_to_compress' - The number of groups to get from the database before stopping
pub fn get_data_from_db(
db_url: &str,
room_id: &str,
@@ -59,7 +59,6 @@ pub fn get_data_from_db(
// Search for the group id of the groups_to_compress'th group after min_state_group
// If this is saved, then the compressor can continue by having min_state_group being
// set to this maximum. If no such group can be found then return None.
let max_group_found = find_max_group(
&mut client,
room_id,
@@ -222,7 +221,7 @@ fn load_map_from_db(
max_group_found,
));
println!("Got initial state from database. Checking for any missing state groups...");
debug!("Got initial state from database. Checking for any missing state groups...");
// Due to reasons some of the state groups appear in the edges table, but
// not in the state_groups_state table.
@@ -251,14 +250,14 @@ fn load_map_from_db(
.collect();
if missing_sgs.is_empty() {
// println!("No missing state groups");
trace!("No missing state groups");
break;
}
missing_sgs.sort_unstable();
missing_sgs.dedup();
// println!("Missing {} state groups", missing_sgs.len());
trace!("Missing {} state groups", missing_sgs.len());
// find state groups not picked up already and add them to the map
let map = get_missing_from_db(client, &missing_sgs, min_state_group, max_group_found);
@@ -535,7 +534,7 @@ pub fn send_changes_to_db(
let mut client = Client::connect(db_url, connector).unwrap();
println!("Writing changes...");
debug!("Writing changes...");
// setup the progress bar
let pb: ProgressBar;

41
src/lib.rs
View File

@@ -20,6 +20,7 @@
// of arguments - this hopefully doesn't make the code unclear
// #[allow(clippy::too_many_arguments)] is therefore used around some functions
use log::{info, warn};
use pyo3::{exceptions, prelude::*};
use clap::{crate_authors, crate_description, crate_name, crate_version, value_t, App, Arg};
@@ -290,7 +291,7 @@ impl Config {
pub fn run(mut config: Config) {
// First we need to get the current state groups
println!("Fetching state from DB for room '{}'...", config.room_id);
info!("Fetching state from DB for room '{}'...", config.room_id);
let (state_group_map, max_group_found) = database::get_data_from_db(
&config.db_url,
@@ -301,19 +302,19 @@ pub fn run(mut config: Config) {
)
.unwrap_or_else(|| panic!("No state groups found within this range"));
println!("Fetched state groups up to {}", max_group_found);
info!("Fetched state groups up to {}", max_group_found);
println!("Number of state groups: {}", state_group_map.len());
info!("Number of state groups: {}", state_group_map.len());
let original_summed_size = state_group_map
.iter()
.fold(0, |acc, (_, v)| acc + v.state_map.len());
println!("Number of rows in current table: {}", original_summed_size);
info!("Number of rows in current table: {}", original_summed_size);
// Now we actually call the compression algorithm.
println!("Compressing state...");
info!("Compressing state...");
let compressor = Compressor::compress(&state_group_map, &config.level_sizes.0);
@@ -327,22 +328,22 @@ pub fn run(mut config: Config) {
let ratio = (compressed_summed_size as f64) / (original_summed_size as f64);
println!(
info!(
"Number of rows after compression: {} ({:.2}%)",
compressed_summed_size,
ratio * 100.
);
println!("Compression Statistics:");
println!(
info!("Compression Statistics:");
info!(
" Number of forced resets due to lacking prev: {}",
compressor.stats.resets_no_suitable_prev
);
println!(
info!(
" Number of compressed rows caused by the above: {}",
compressor.stats.resets_no_suitable_prev_size
);
println!(
info!(
" Number of state groups changed: {}",
compressor.stats.state_groups_changed
);
@@ -352,14 +353,14 @@ pub fn run(mut config: Config) {
}
if ratio > 1.0 {
println!("This compression would not remove any rows. Exiting.");
warn!("This compression would not remove any rows. Exiting.");
return;
}
if let Some(min) = config.min_saved_rows {
let saving = (original_summed_size - compressed_summed_size) as i32;
if saving < min {
println!(
warn!(
"Only {} rows would be saved by this compression. Skipping output.",
saving
);
@@ -485,7 +486,7 @@ fn output_sql(
return;
}
println!("Writing changes...");
info!("Writing changes...");
let pb: ProgressBar;
if cfg!(feature = "no-progress-bars") {
@@ -559,7 +560,7 @@ pub fn continue_run(
let ratio = (new_num_rows as f64) / (original_num_rows as f64);
if ratio > 1.0 {
println!("This compression would not remove any rows. Aborting.");
warn!("This compression would not remove any rows. Aborting.");
return Some(ChunkStats {
new_level_info: compressor.get_level_info(),
last_compressed_group: max_group_found,
@@ -600,7 +601,7 @@ fn check_that_maps_match(
old_map: &BTreeMap<i64, StateGroupEntry>,
new_map: &BTreeMap<i64, StateGroupEntry>,
) {
println!("Checking that state maps match...");
info!("Checking that state maps match...");
let pb: ProgressBar;
if cfg!(feature = "no-progress-bars") {
@@ -625,10 +626,10 @@ fn check_that_maps_match(
pb.inc(1);
if expected != actual {
println!("State Group: {}", sg);
println!("Expected: {:#?}", expected);
println!("actual: {:#?}", actual);
Err(format!("States for group {} do not match", sg))
Err(format!(
"States for group {} do not match. Expected {:#?}, found {:#?}",
sg, expected, actual
))
} else {
Ok(())
}
@@ -637,7 +638,7 @@ fn check_that_maps_match(
pb.finish();
println!("New state map matches old one");
info!("New state map matches old one");
}
/// Gets the full state for a given group from the map (of deltas)

19
src/main.rs
View File

@@ -20,8 +20,27 @@
#[global_allocator]
static GLOBAL: jemallocator::Jemalloc = jemallocator::Jemalloc;
use log::LevelFilter;
use std::env;
use std::io::Write;
use synapse_compress_state as comp_state;
fn main() {
// setup the logger
// The default can be overwritten with RUST_LOG
// see the README for more information <---- TODO
if env::var("RUST_LOG").is_err() {
let mut log_builder = env_logger::builder();
// Only output the log message (and not the prefixed timestamp etc.)
log_builder.format(|buf, record| writeln!(buf, "{}", record.args()));
// By default print all of the debugging messages from this library
log_builder.filter_module("synapse_compress_state", LevelFilter::Debug);
log_builder.init();
} else {
// If RUST_LOG was set then use that
env_logger::Builder::from_env("RUST_LOG").init();
}
comp_state::run(comp_state::Config::parse_arguments());
}