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();
}
}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();
}
}Как я понимаю, этот .catch должен перехватить реджект главного
промиса, но эта часть в принципе не выполняется и выходит
ошибка UnhandledPromiseRejection
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()
};FROM rust:latest as build
WORKDIR /test-tcp
COPY ./Cargo.lock ./Cargo.lock
COPY ./Cargo.toml ./Cargo.toml
COPY ./src ./src
RUN cargo build --release
FROM debian:buster-slim
COPY --from=build /test-tcp/target/release/test-tcp /usr/src/test-tcp
CMD ["/usr/src/test-tcp"]RUN USER=root cargo new --bin test-tcp
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!("- - - - -");
}#[tokio::main(flavor = "multi_thread", worker_threads = 1024)]async fn set(&mut self, name: String, value: String) {
self.data.insert(name, value);
}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() -> u8 - это указатель на функцию. В него можно записать или обычную функцию с подходящей сигнатурой или замыкание, которое ничего не замыкает:fn some_func() -> u8 {
0
}
let f: fn() -> u8 = some_func;
let f: fn() -> u8 = || 0;fn main() {
let mut funcs = Vec::with_capacity(3);
for i in 1u8..=3 {
funcs.push(move || i);
}
for i in 0..3 {
println!("{}", funcs[i]());
}
}std::ops::RangeInclusive::<u8>::new
1. Как происходит связь между кодом js, c++. На самом гитхабе Nodejs можно открыть папки с функциями, которые я импортирую в коде. А также их реализации в файлах .cc расширения. Но не могли бы подробнее описать механизм, которые позволяет 2 разных языка использовать.JS движок (например V8) предоставляет платформе (Node.js) некоторый C++ api, через который платформа может движком управлять. Например через это api можно дать движку строчку с JS кодом и попросить его выполнить, а в ответ получить указатель на результат выполнения (V8 например возвращает результат последнего выражения в коде). Так же, если у нас есть указатель на JS функцию, можно попросить движок ее выполнить с определенными аргументами и this, а в ответ получить результат этого выполнения (функция что-то вернула или бросила исключение). Ну и наконец, мы можем дать этому api указатель на C/C++ функцию, а движок сделает из нее JS функцию и вернет указатель на нее, когда JS функция вызывается движок вызовет нашу C/C++ функцию и даст ей на вход контекст выполнения, из которого можно извлечь this и аргументы функции, так же через контекст можно установить результат выполнения JS функции, указав что JS функция должна вернуть некоторое JS значение или бросить некоторое исключение.
2. V8 - движок для исполнения JS кода, то есть превращения его в машинный код для исполнения комманд. Libuv написана на с++. Они же не работают в паре? Я так понимаю, что часть кода исполняется на движке v8, а различные асинхронные вызовы, которые через Nodejs api делегируются библиотеке Libuv - уже от неё превращаются в низкоуровневый машинный код, который исполняется компьютеров?Можете чуть пролить свет на исполнения уже между этими компонентами Libuv, V8.V8 ничего не знает о libuv, как и libuv ничего не знает о V8.
3. После того, как Nodejs передает асинхронный вызов Libuv на исполнение. То callback функция помещается позже в стэк для исполнения, а исполнятся начинается после того, как система просигнализирует о завершении процесса, например считывания файла. Или вначале считывается файл, а потом система об этом сигнализирует и помещается callback функция в стэк Event Loop'a для исполнения?Выше уже расписал как Node.js взаимодействует с libuv в теле event loop, так же расписал как JS код может вызвать C++ функцию. Рассмотрим на сильно упрощенном примере, что происходит когда в JS коде на Node.js мы вызываем
fs.readFile(filepath, callback):4. Можете более подробней рассказать про исполнение кода на инстансах приложения в зависимости от количества ядер процессора и управления потоками:Каждый инстанс Node.js - это отдельный процесс операционной системы. У каждого процесса своя изолированная память. И да, каждый может запускать несколько потоков (у Node.js кстати 4 потока - это минимум, при большой нагрузке она может добавить в пул до 128 потоков), но в случае с Node.js большая часть потоков почти всегда что-то ждут (пока тот же файл прочитается например) и почти не тратят ресурсы CPU, а активно работает лишь один - тот в котором event loop.
4.1 Есть к примеру приложений на Nodejs. На компьютере 4 ядра. Я на каждом ядре могу запустить инстанс этого приложения. И у каждого инстанса в pool thread'e будет по 4 потока дополнительных, которые создаются по умолчанию?
4.2 Если отойти от ноды и запустить приложение на php фреймворке на Apache и этот сервер выделяет под каждый запрос свой поток.В данном случае - это такой же условно поток, который может выделяться и библиотекой Libuv или это абсолютно разные понятия? И сколько инстансов php приложения запущено, когда Apache под каждый создает свой поток? Как соотносится поток с инстансом приложения на php.С точки зрения операционной системы - да, это такие же потоки. Разница заключается в том, что в этих потоках происходит. В php только 1 поток, который то что-то полезное делает, то ждет ответа от БД/файловой системы/еще чего-то. А в Node.js это разделено, 1 поток занят только полезной работой (выполнением JS кода) и несколько только ждут ответа.
4.3 Может в дополнение расскажите про потоки в системе, их ограничения и т.дТут целую лекцию можно читать. Думаю это стоит вынести в отдельный вопрос.
Есть сборщик gulpТо что под gulp есть сборщики не делает сам gulp сборщиком. Gulp - task runner, запускалка задач. То есть сборщик под капотом может быть любой, хоть тот же vite.