Playground/1brc
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

mostly made output a bit nicer

Browse Source
This commit is contained in:
Fabian Schmidt 2024-07-23 13:23:26 +02:00
parent 393f802741
commit b4e3992c65
3 changed files with 19 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/rust/Cargo.toml
View File

@ -14,3 +14,7 @@ memmap = "0.7.0"
rayon = "1.10.0" rayon = "1.10.0"
rustc-hash = "2.0.0" rustc-hash = "2.0.0"
[profile.release]
lto = "fat"
strip = "symbols"
panic = "abort"

22
src/main/rust/src/bin/multi_threaded.rs
View File

@ -6,12 +6,14 @@ use std::{
use std::collections::HashMap; use std::collections::HashMap;
use std::sync::mpsc; use std::sync::mpsc;
use std::time::Instant; use std::time::Instant;
use onebrc::format_nums;
use onebrc::parse_temp;
const DEFAULT_HASHMAP_LENGTH: usize = 10000; const DEFAULT_HASHMAP_LENGTH: usize = 10000;
fn main() { fn main() {
print!("\x1b[2J"); //print!("\x1b[2J");
print!("\x1b[s");
thread::scope(|s| { thread::scope(|s| {
let mut stations: HashMap<String, onebrc::StationMeasurements> = HashMap::with_capacity(DEFAULT_HASHMAP_LENGTH); let mut stations: HashMap<String, onebrc::StationMeasurements> = HashMap::with_capacity(DEFAULT_HASHMAP_LENGTH);
let (tx, rx) = mpsc::channel(); let (tx, rx) = mpsc::channel();
@ -19,6 +21,8 @@ fn main() {
let cores: usize = thread::available_parallelism().unwrap().into(); let cores: usize = thread::available_parallelism().unwrap().into();
let chunk_length = 1_000_000_000 / cores; let chunk_length = 1_000_000_000 / cores;
for i in 0..cores { for i in 0..cores {
let print_line = i + 1;
print!("\x1b[u\x1b[{print_line}B\x1b[0CThread #{i:0>2}: 0%");
let file = File::open("../../../measurements.txt").expect("File measurements.txt not found"); let file = File::open("../../../measurements.txt").expect("File measurements.txt not found");
let reader = BufReader::new(file); let reader = BufReader::new(file);
let line_chunk = reader.lines().skip(chunk_length * i).take(chunk_length); let line_chunk = reader.lines().skip(chunk_length * i).take(chunk_length);
@ -28,20 +32,21 @@ fn main() {
HashMap::with_capacity(DEFAULT_HASHMAP_LENGTH); HashMap::with_capacity(DEFAULT_HASHMAP_LENGTH);
let now_read_line = Instant::now(); let now_read_line = Instant::now();
let print_line = i + 1;
let mut line_num = 0; let mut line_num = 0;
line_chunk.for_each(|line| { line_chunk.for_each(|line| {
if line_num == 0 { if line_num == 0 {
print!("\x1b[{print_line};30HStart read line {}ms", now_read_line.elapsed().as_millis()); print!("\x1b[u\x1b[{print_line}B\x1b[30CStart read line {}ms", now_read_line.elapsed().as_millis());
} }
if line_num % 10000 == 0 { if line_num % 10000 == 0 {
let formatted = format_nums(line_num); //let formatted = format_nums(line_num);
print!("\x1b[{print_line};0HThread #{i:0>2}: {formatted}"); //print!("\x1b[u\x1b[{print_line}B\x1b[0CThread #{i:0>2}: {formatted}");
let percent = (line_num as f64 / chunk_length as f64) * 100.0;
print!("\x1b[u\x1b[{print_line}B\x1b[0CThread #{i:0>2}: {percent:.2}%");
} }
line_num += 1; line_num += 1;
let line = line.expect("could not read line"); let line = line.expect("could not read line");
let (station, temp) = line.split_once(';').expect("Error while splitting"); let (station, temp) = line.split_once(';').expect("Error while splitting");
let temp = onebrc::parse_temp(temp.as_bytes()); let temp = parse_temp(temp.as_bytes());
let measurements_option = t_stations.get_mut(station); let measurements_option = t_stations.get_mut(station);
if let Some(measurements) = measurements_option { if let Some(measurements) = measurements_option {
measurements.update(temp); measurements.update(temp);
@ -55,10 +60,11 @@ fn main() {
t_stations.insert(station.to_owned(), measurements); t_stations.insert(station.to_owned(), measurements);
} }
}); });
print!("\x1b[{print_line};60HTime reading lines in thread {i}={} ms", now_read_line.elapsed().as_millis()); print!("\x1b[u\x1b[{print_line}B\x1b[60CTime reading lines in thread {i}={} ms", now_read_line.elapsed().as_millis());
let _ = tx.send(t_stations); let _ = tx.send(t_stations);
}); });
} }
print!("\x1b[{cores}B");
drop(tx); drop(tx);
while let Ok(t_stations) = rx.recv() { while let Ok(t_stations) = rx.recv() {
for (station, measurements) in t_stations.iter() { for (station, measurements) in t_stations.iter() {

3
src/main/rust/src/bin/referenceImpl.rs
View File

@ -68,8 +68,7 @@ fn merge<'a>(a: &mut HashMap<&'a BStr, State>, b: &HashMap<&'a BStr, State>) {
} }
fn main() { fn main() {
//let cores: usize = std::thread::available_parallelism().unwrap().into(); let cores: usize = std::thread::available_parallelism().unwrap().into();
let cores: usize = 1;
let path = match std::env::args().skip(1).next() { let path = match std::env::args().skip(1).next() {
Some(path) => path, Some(path) => path,
None => "measurements.txt".to_owned(), None => "measurements.txt".to_owned(),