Дмитрий Беляев

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();
    }
}

let cwd = std::env::current_dir().unwrap().parent().unwrap();

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"]

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 и передать ему новую задачу (новые данные), - структуру данных, которую привёл в основном вопросе.

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!("- - - - -");
}

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]());
    }
}

use std::rc::{Rc, Weak};

pub struct Node {
    parent: Weak<Node>,
    data: u32,
    pub left: Option<Rc<Node>>,
    pub right: Option<Rc<Node>>,
}

#[derive(Debug)]
pub struct SetChildError;

impl Node {
    pub fn new(data: u32) -> Self {
        Self {
            parent: Weak::new(),
            data,
            left: None,
            right: None,
        }
    }

    pub fn set_left(self: &mut Rc<Self>, mut child: Rc<Self>) -> Result<(), SetChildError> {
        let child_mut = Rc::get_mut(&mut child).ok_or(SetChildError)?;
        child_mut.parent = Rc::downgrade(self);

        let self_mut = Rc::get_mut(self).ok_or(SetChildError)?;
        self_mut.left = Some(child);

        Ok(())
    }
}

#[no_mangle]
pub fn test() {
    // Вот тут на стеке выделено 4 байта
    let mut idx: i32 = 10;
    // Указатель на эти 4 байта
    let ptr = &mut idx as *mut i32;

    // По сути указатель на те же 4 байта, но на деле указатель на 40 байт
    let buffer = unsafe { core::slice::from_raw_parts_mut(ptr, 10) };

    // И тут мы работаем с этими 40 байтами,
    // но нормальные из них только 4 байта в начале,
    // а остальные смотрят в глубь стека и меняют его
    // то есть перетирают адрес возврата и данные на стеке вызвавшей функции
    for pixel in buffer.iter_mut() {
        *pixel = 0x7d2b7500;
    }
}