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

Use FxHashMap for multi_threaded_smol.rs

Browse Source
This commit is contained in:
Fabian Schmidt 2024-08-19 10:57:18 +02:00
parent 7add8793a5
commit 2a89d061a0
1 changed files with 14 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
src/main/rust/src/implementations/multi_threaded_smol.rs
View File

@ -1,10 +1,9 @@
use smol::fs::File;
use smol::io::{AsyncBufReadExt, AsyncSeekExt, BufReader, SeekFrom};
use rustc_hash::{FxHashMap as HashMap, FxBuildHasher};
use crate::models::station_measurements::StationMeasurements;
use crate::utils::parse;
use crate::utils::parse::hashstr;
use std::collections::HashMap;
use std::sync::mpsc;
use std::thread;
use std::time::Instant;
@ -15,8 +14,9 @@ pub fn run() {
const FILE_PATH: &str = "../../../measurements.txt";
let now = Instant::now();
thread::scope(|s| {
let mut stations: HashMap<usize, (String, StationMeasurements)> =
HashMap::with_capacity(DEFAULT_HASHMAP_LENGTH);
let hasher = FxBuildHasher::default();
let mut stations: HashMap<String, StationMeasurements> =
HashMap::with_capacity_and_hasher(DEFAULT_HASHMAP_LENGTH, hasher);
let (tx, rx) = mpsc::channel();
let cores = thread::available_parallelism().unwrap().into();
let bounds = smol::block_on(async {
@ -57,8 +57,8 @@ pub fn run() {
.expect("File measurements.txt not found");
let mut reader = BufReader::new(&mut file);
reader.seek(SeekFrom::Start(currposition)).await.unwrap();
let mut t_stations: HashMap<usize, (String, StationMeasurements)> =
HashMap::with_capacity(DEFAULT_HASHMAP_LENGTH);
let mut t_stations: HashMap<String, StationMeasurements> =
HashMap::with_capacity_and_hasher(DEFAULT_HASHMAP_LENGTH, hasher);
let mut line = Vec::with_capacity(108);
loop {
let line_len = reader
@ -69,11 +69,10 @@ pub fn run() {
break;
}
let (station, temp) = line.rsplit_once(|&byte| byte == b';').unwrap();
let hash = hashstr(station);
let station = unsafe { String::from_utf8_unchecked(Vec::from(station)) };
let temp = parse::temp(temp.split_last().unwrap().1);
let measurements_option = t_stations.get_mut(&hash);
if let Some((_, measurements)) = measurements_option {
let measurements_option = t_stations.get_mut(&station);
if let Some(measurements) = measurements_option {
measurements.update(temp);
} else {
let measurements = StationMeasurements {
@ -82,7 +81,7 @@ pub fn run() {
count: 1,
sum: temp,
};
t_stations.insert(hash, (station, measurements));
t_stations.insert(station, measurements);
}
currposition += line_len as u64;
if currposition >= end {
@ -96,18 +95,18 @@ pub fn run() {
}
drop(tx);
while let Ok(t_stations) = rx.recv() {
for (&hash, (station, measurements)) in t_stations.iter() {
let joined_measurements_options = stations.get_mut(&hash);
if let Some((_, joined_measurements)) = joined_measurements_options {
for (station, measurements) in t_stations.iter() {
let joined_measurements_options = stations.get_mut(station);
if let Some(joined_measurements) = joined_measurements_options {
joined_measurements.merge(measurements);
} else {
stations.insert(hash, (station.to_owned(), *measurements));
stations.insert(station.to_owned(), *measurements);
}
}
}
let mut stations: Vec<String> = stations
.iter()
.map(|(_, (station, measurements))| {
.map(|(station, measurements)| {
let measurements = measurements.to_string();
#[cfg(feature = "json")]
{