Как передать ссылку на переменную стека в поток?

Я пишу сервер WebSocket, где веб-клиент подключается для игры в шахматы против многопоточного компьютерного ИИ. Сервер WebSocket хочет передать объект Logger в код AI. Объект Logger будет собирать строки журнала с ИИ на веб-клиент. Logger должен содержать ссылку на клиентское соединение.

Я смущен тем, как жизнь взаимодействует с потоками. Я воспроизвел проблему с конструкцией Wrapper, параметризованной типом. Функция run_thread пытается развернуть значение и записать его.

use std::fmt::Debug;
use std::thread;

struct Wrapper<T: Debug> {
    val: T,
}

fn run_thread<T: Debug>(wrapper: Wrapper<T>) {
    let thr = thread::spawn(move || {
        println!("{:?}", wrapper.val);
    });

    thr.join();
}

fn main() {
    run_thread(Wrapper::<i32> { val: -1 });
}

Аргумент Wrapper живет в стеке, и его срок службы не распространяется за рамки фрейма run_thread, даже если поток будет соединен до окончания стека. Я мог бы скопировать значение из стека:

use std::fmt::Debug;
use std::thread;

struct Wrapper<T: Debug + Send> {
    val: T,
}

fn run_thread<T: Debug + Send + 'static>(wrapper: Wrapper<T>) {
    let thr = thread::spawn(move || {
        println!("{:?}", wrapper.val);
    });

    thr.join();
}

fn main() {
    run_thread(Wrapper::<i32> { val: -1 });
}

Это не сработает, если T является ссылкой на большой объект, который я не хочу скопировать:

use std::fmt::Debug;
use std::thread;

struct Wrapper<T: Debug + Send> {
    val: T,
}

fn run_thread<T: Debug + Send + 'static>(wrapper: Wrapper<T>) {
    let thr = thread::spawn(move || {
        println!("{:?}", wrapper.val);
    });

    thr.join();
}

fn main() {
    let mut v = Vec::new();
    for i in 0..1000 {
        v.push(i);
    }

    run_thread(Wrapper { val: &v });
}

Результат:

error: `v` does not live long enough
  --> src/main.rs:22:32
   |
22 |     run_thread(Wrapper { val: &v });
   |                                ^ does not live long enough
23 | }
   | - borrowed value only lives until here
   |
   = note: borrowed value must be valid for the static lifetime...

Единственное решение, о котором я могу думать, это использовать Arc.

use std::fmt::Debug;
use std::sync::Arc;
use std::thread;

struct Wrapper<T: Debug + Send + Sync + 'static> {
    arc_val: Arc<T>,
}

fn run_thread<T: Debug + Send + Sync + 'static>(wrapper: &Wrapper<T>) {
    let arc_val = wrapper.arc_val.clone();
    let thr = thread::spawn(move || {
        println!("{:?}", *arc_val);
    });

    thr.join();
}

fn main() {
    let mut v = Vec::new();
    for i in 0..1000 {
        v.push(i);
    }

    let w = Wrapper { arc_val: Arc::new(v) };
    run_thread(&w);

    println!("{}", (*w.arc_val)[0]);
}

В моей реальной программе кажется, что и Logger, и объект соединения должны быть помещены в обертки Arc. Кажется раздражающим, что клиент должен подключить соединение в Arc, когда он является внутренним для библиотеки, который распараллелен. Это особенно раздражает, потому что срок службы соединения гарантированно будет больше, чем срок службы рабочих потоков.

Я что-то пропустил?

Ответы

Ответ 1

Поддержка потоков в стандартной библиотеке позволяет созданным потокам пережить созданный ими поток; это хорошо! Однако если вы передадите ссылку на выделенную в стеке переменную одному из этих потоков, нет гарантии, что эта переменная будет действительна к тому моменту, когда поток выполнится. На других языках это позволило бы потоку обращаться к недопустимой памяти, создавая кучу проблем безопасности памяти.

К счастью, мы не ограничены стандартной библиотекой. По крайней мере два ящика предоставляют потоки с областью видимости - потоки, которые гарантированно завершат работу до окончания определенной области. Это может гарантировать, что переменные стека будут доступны на протяжении всего потока:

Существуют также ящики, которые абстрагируют низкоуровневые детали "потоков", но позволяют вам достичь ваших целей:

Вот примеры каждого. Каждый пример порождает несколько потоков и мутирует локальный вектор на месте без блокировки, без Arc и без клонирования. Обратите внимание, что мутация имеет sleep вызов, чтобы убедиться, что вызовы происходят параллельно.

Вы можете расширить примеры, чтобы поделиться ссылкой на любой тип, который реализует Sync, например, Mutex или Atomic*. Однако их использование может привести к блокировке.

область видимости-ThreadPool

use scoped_threadpool::Pool; // 0.1.9
use std::{thread, time::Duration};

fn main() {
    let mut vec = vec![1, 2, 3, 4, 5];
    let mut pool = Pool::new(vec.len() as u32);

    pool.scoped(|scoped| {
        for e in &mut vec {
            scoped.execute(move || {
                thread::sleep(Duration::from_secs(1));
                *e += 1;
            });
        }
    });

    println!("{:?}", vec);
}

коромысло

use crossbeam; // 0.6.0
use std::{thread, time::Duration};

fn main() {
    let mut vec = vec![1, 2, 3, 4, 5];

    crossbeam::scope(|scope| {
        for e in &mut vec {
            scope.spawn(move |_| {
                thread::sleep(Duration::from_secs(1));
                *e += 1;
            });
        }
    })
    .expect("A child thread panicked");

    println!("{:?}", vec);
}

вискоза

use rayon::iter::{IntoParallelRefMutIterator, ParallelIterator}; // 1.0.3
use std::{thread, time::Duration};

fn main() {
    let mut vec = vec![1, 2, 3, 4, 5];

    vec.par_iter_mut().for_each(|e| {
        thread::sleep(Duration::from_secs(1));
        *e += 1;
    });

    println!("{:?}", vec);
}

клиент должен запаковать соединение в Arc когда оно является внутренним по отношению к библиотеке, код распараллеливается

Возможно, вы сможете скрыть свой параллелизм лучше? Не могли бы вы принять регистратор и затем обернуть его в Arc/Mutex прежде чем передать его своим потокам?