#[tokio::main(flavor = "multi_thread", worker_threads = 1024)]
1024 - потеряли весь профит от небольшого числа потоков, теперь ОС будет распределять 1024 потока на небольшое количество ядер CPU.async fn set(&mut self, name: String, value: String) {
self.data.insert(name, value);
}
у этого метода нет ни одной причины быть асинхронным, операции с HashMap - чистый CPU-bound.fn main() {
let start = Instant::now();
let handles: Vec<_> = (0..4)
.map(|table_index| {
std::thread::spawn(move || {
let mut table = Table::new();
for i in (0..3000000).filter(|i| (i % 4 + 1) == table_index) {
table.set(format!("{}", i), format!("value{}", i));
}
table
})
})
.collect();
for handle in handles {
let _table = handle.join().unwrap();
// тут добавляем таблицы в менеджер
}
let elapsed = start.elapsed();
println!("Time taken to set 3,000,000 keys: {:?}", elapsed);
}
и даже это можно заморочиться и улучшить, например запускать потоков не больше std::thread::available_parallelism()
или оптимизировать счетчик для каждой таблицы ((0..3000000).filter(|i| (i % 4 + 1) == table_index)
), но это я оставлю Вам в качестве д/з.fn main() {
let a = "x";
println!("1. {:?}", a); // "x"
let b = format!("{}y", a);
println!("2. {:?}", b); // "xy"
}
fn main() {
let mut s = "x".to_string();
println!("1. {}", s); // "x"
s += "y";
println!("2. {}", s); // "xy"
}
use std::slice;
fn main() {
let range = (0..u8::MAX).collect::<Vec<_>>();
let v = range
.iter()
.map(|x| std::str::from_utf8(slice::from_ref(x)).unwrap_or("Err"))
.collect::<Vec<_>>();
println!("{v:?}")
}
fn main() {
let src: Vec<[u8; 1]> = (0..u8::MAX).map(|i| [i]).collect();
let mut info = Vec::<&str>::with_capacity(u8::MAX.into());
for u in &src {
let t = std::str::from_utf8(&*u).unwrap_or("Err");
info.push(t);
}
println!("\n{:#?}", info);
}
info
будет связан лайфтаймом с src
, чтоб избавится от этого нужно хранить в нём не &str
а String
или Box<str>
fn main() {
let info: Vec<Box<str>> = (0..u8::MAX).map(|i| {
let u = [i];
let t = std::str::from_utf8(&u).unwrap_or("Err");
t.into()
}).collect();
println!("\n{:#?}", info);
}
Vec<&str>
элементы которого ссылаются на исходный вектор байт и в каждом элементе строка из 1 символа:let u01 = vec![59, 13, 10, 32, 47, 42];
let u01_str = std::str::from_utf8(&u01).expect("invalid utf8");
let mut u02 = Vec::with_capacity(u01.len());
let mut i0 = 0;
for (i, _) in u01_str.char_indices().skip(1) {
u02.push(&u01_str[i0..i]);
i0 = i;
}
u02.push(&u01_str[i0..]);
println!("u02 = {:?}", u02);
fn test() {
let mut channels = Arc::new(Mutex::new(Vec::with_capacity(PAR)));
let mut joins = Vec::with_capacity(PAR);
for _ in 0..N / PAR {
for _ in 0..PAR {
let mut channels = Arc::clone(&channels);
joins.push(thread::spawn(move || {
get(channels.lock().unwrap());
}));
}
}
for j in joins {
j.join().unwrap();
}
}
#[inline(always)]
fn get(mut channels: MutexGuard<Vec<mpsc::Sender<i32>>>) -> i32 {
let (tx, rx) = mpsc::channel();
channels.push(tx);
if channels.len() == PAR {
exec(channels);
} else {
drop(channels); // drop гварда отпускает мьютекс
}
rx.recv().unwrap()
}
#[inline(always)]
fn exec(mut channels: MutexGuard<Vec<mpsc::Sender<i32>>>) {
let mut i = 0;
for c in channels.iter() {
i += 1;
c.send(1).unwrap();
}
println!("{}", i);
channels.clear();
// а здесь гвард дропнется сам
}
fn test() {
let (tx, rx) = mpsc::channel::<mpsc::Sender<i32>>();
let mut handles = Vec::with_capacity(N + 1);
handles.push(thread::spawn(move || exec(rx)));
for _ in 0..N {
let tx = tx.clone();
handles.push(thread::spawn(move || {
get(tx);
}))
}
drop(tx);
for handle in handles {
handle.join().unwrap();
}
}
fn get(sender: mpsc::Sender<mpsc::Sender<i32>>) -> i32 {
let (tx, rx) = mpsc::channel();
sender.send(tx).unwrap();
rx.recv().unwrap()
}
fn exec(receiver: mpsc::Receiver<mpsc::Sender<i32>>) {
let mut channels = Vec::with_capacity(PAR);
loop {
for _ in 0..PAR {
let Ok(tx) = receiver.recv() else {
return;
};
channels.push(tx);
}
let mut i = 0;
for c in channels.iter() {
i += 1;
c.send(1).unwrap();
}
println!("{}", i);
channels.clear();
}
}
Но особо это профита не дает, так как основной пожиратель перфоманса - switch context в ОС. Тысячи потоков делают только хуже. Запускать потоков сильно больше чем есть ядер - это вообще плохая идея. Просто ради интереса переписал еще раз на асинхронных каналах tokio:async fn test() {
let (tx, rx) = mpsc::unbounded_channel::<mpsc::UnboundedSender<i32>>();
let mut handles = Vec::with_capacity(N + 1);
handles.push(tokio::spawn(async move { exec(rx).await }));
for _ in 0..N {
let tx = tx.clone();
handles.push(tokio::spawn(async move {
get(tx).await;
}))
}
drop(tx);
for handle in handles {
handle.await.unwrap();
}
}
async fn get(sender: mpsc::UnboundedSender<mpsc::UnboundedSender<i32>>) -> i32 {
let (tx, mut rx) = mpsc::unbounded_channel();
sender.send(tx).unwrap();
rx.recv().await.unwrap()
}
async fn exec(mut receiver: mpsc::UnboundedReceiver<mpsc::UnboundedSender<i32>>) {
let mut channels = Vec::with_capacity(PAR);
loop {
for _ in 0..PAR {
let Some(tx) = receiver.recv().await else {
return;
};
channels.push(tx);
}
let mut i = 0;
for c in channels.iter() {
i += 1;
c.send(1).unwrap();
}
println!("{}", i);
channels.clear();
}
}
и запустил в многопоточном рантайме в дефолтной конфигурации (количество воркеров == количеству ядер), работает быстрее в 19 раз.let cwd = std::env::current_dir().unwrap().parent().unwrap();
меня очень смущает постоянно выделять буффер
use std::mem::MaybeUninit;
let mut buffer = unsafe {
#[allow(invalid_value)]
MaybeUninit::<[_; 1024]>::uninit().assume_init()
};
Но я бы так не делал. Во-первых чистота кода не стоит этих копеек производительности, а во-вторых немного накосячите с чтением и будет UB. use std::sync::Arc;
fn main() {
let data = Arc::new(vec![1, 2, 3]);
let thread_1 = std::thread::spawn({
let data = Arc::clone(&data);
move || {
println!("Thread 1 data: {:?}", data);
}
});
let thread_2 = std::thread::spawn({
let data = Arc::clone(&data);
move || {
println!("Thread 2 data: {:?}", data);
}
});
thread_1.join().unwrap();
thread_2.join().unwrap();
}
Я обернул создание отдельного потока в функцию и так передавал в поток данные. Удобно, что такую функцию могу вынести в отдельный файл-модуль. Но не смог такое сделать динамически в цикле для группы потоков. Хочу подойти к варианту, когда поток, который закончил выполнение своего кода (раньше других), можно опять запустить из main и передать ему новую задачу (новые данные), - структуру данных, которую привёл в основном вопросе.Если правильно понял, то Вам нужен thread pool. Можно использовать из библиотеки rayon: https://docs.rs/rayon/1.7.0/rayon/struct.ThreadPoo...
use std::{
sync::{
mpsc::{self, Sender},
Arc, Mutex,
},
thread::{self, JoinHandle},
};
type Task = Box<dyn FnOnce() + Send + 'static>;
pub struct ThreadPool {
handles: Vec<JoinHandle<()>>,
task_sender: Sender<Task>,
}
impl ThreadPool {
pub fn new() -> Self {
let threads_count = thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(2);
let (task_sender, task_receiver) = mpsc::channel::<Task>();
let task_receiver = Arc::new(Mutex::new(task_receiver));
let mut handles = Vec::with_capacity(threads_count);
for _ in 0..threads_count {
let task_receiver = Arc::clone(&task_receiver);
handles.push(thread::spawn(move || loop {
let task_receiver = task_receiver.lock().unwrap_or_else(|e| e.into_inner());
let Ok(task) = task_receiver.recv() else {
return;
};
drop(task_receiver);
task();
}));
}
Self {
handles,
task_sender,
}
}
pub fn spawn_task<F: FnOnce() + Send + 'static>(&self, f: F) {
self.task_sender.send(Box::new(f)).expect("All threads ended");
}
pub fn join(self) -> thread::Result<()> {
drop(self.task_sender);
for handle in self.handles {
handle.join()?;
}
Ok(())
}
}
#[test]
fn thread_pool() {
let pool = ThreadPool::new();
for i in 0..500 {
pool.spawn_task(move || {
println!("Task {} working on thread {:?}", i, thread::current().id());
});
}
pool.join().unwrap();
}
use chrono::prelude::*;
use std::sync::mpsc;
use std::sync::mpsc::{Receiver, Sender};
use std::{thread, time};
fn main() {
println!("- - - - -");
let mut children = Vec::with_capacity(3);
for id in 0..children.capacity() {
let child = thread::spawn(move || {
let mut date_times = Vec::with_capacity(5);
for i in 0..date_times.capacity() {
let t: DateTime<Local> = Local::now();
date_times.push(t);
println!("{:?}_ поток, задача _{:?}, время: {:?}", id, i, t);
thread::sleep(time::Duration::from_millis(3));
}
(id, date_times)
});
children.push(child);
}
for child in children {
let (id, date_times) = child.join().expect("Дочерний поток паникует");
println!("thd_{} = {:?}", id, date_times);
}
println!("- - - - -");
}
mod m01;
mod m02;
mod m03;
fn main() {
m01::f1();
m02::f2();
let (i, p) = m03::f3();
println!("i = {:?}", i);
println!("p = {:?}", p);
}
pub fn f1() {
let num: u8 = 12;
println!("num = {:?}", num);
}
pub fn f2() {
let s: String = "abc".to_string();
println!("s = {:?}", s);
}
pub fn f3() -> (u8, String) {
let i: u8 = 88;
let p: String = "xyz".to_string();
(i, p)
}
macro_rules! f4 {
() => {
let i: u8 = 88;
let p: String = "xyz".to_string();
}
}
macro_rules! f4 {
($a: ident, $b: ident) => {
let $a: u8 = 88;
let $b: String = "xyz".to_string();
}
}
fn main() {
f4!(i, p);
println!("{} {}", i, p);
}