Многонишково програмиране

14 ноември 2019

Преговор

Преговор

Преговор

Преговор

Преговор

Преговор

Fearless concurrency

Must be this tall

Нишки

1 2 3 4 5 6 7
use std::thread;

fn main() {
    thread::spawn(|| println!("hi from spawned thread"));

    println!("hi from main thread");
}
use std::thread;

fn main() {
    thread::spawn(|| println!("hi from spawned thread"));

    println!("hi from main thread");
}

Нишки

1 2 3 4 5 6 7
use std::thread;

fn main() {
    thread::spawn(|| println!("hi from spawned thread"));

    println!("hi from main thread");
}
use std::thread;

fn main() {
    thread::spawn(|| println!("hi from spawned thread"));

    println!("hi from main thread");
}

Нишки

1 2 3 4 5 6 7
use std::thread;

fn main() {
    thread::spawn(|| println!("hi from spawned thread"));

    println!("hi from main thread");
}
use std::thread;

fn main() {
    thread::spawn(|| println!("hi from spawned thread"));

    println!("hi from main thread");
}

Нишки

1 2 3 4 5 6 7
use std::thread;

fn main() {
    thread::spawn(|| println!("hi from spawned thread"));

    println!("hi from main thread");
}
use std::thread;

fn main() {
    thread::spawn(|| println!("hi from spawned thread"));

    println!("hi from main thread");
}

Нишки

1 2 3 4 5 6 7 8
use std::thread;

fn main() {
    let handle = thread::spawn(|| println!("hi from spawned thread"));

    println!("hi from main thread");
    let _ = handle.join();
}
hi from main thread hi from spawned thread
use std::thread;

fn main() {
    let handle = thread::spawn(|| println!("hi from spawned thread"));

    println!("hi from main thread");
    let _ = handle.join();
}

Нишки

1 2 3 4 5 6 7 8
use std::thread;

fn main() {
    let handle = thread::spawn(|| println!("hi from spawned thread"));

    println!("hi from main thread");
    let _ = handle.join();
}
hi from main thread hi from spawned thread
use std::thread;

fn main() {
    let handle = thread::spawn(|| println!("hi from spawned thread"));

    println!("hi from main thread");
    let _ = handle.join();
}

Нишки

1 2 3 4 5 6 7 8
use std::thread;

fn main() {
    let handle = thread::spawn(|| println!("hi from spawned thread"));

    println!("hi from main thread");
    let _ = handle.join();
}
hi from main thread hi from spawned thread
use std::thread;

fn main() {
    let handle = thread::spawn(|| println!("hi from spawned thread"));

    println!("hi from main thread");
    let _ = handle.join();
}

Нишки

1 2 3 4 5 6 7 8
use std::thread;

fn main() {
    let handle = thread::spawn(|| println!("hi from spawned thread"));

    println!("hi from main thread");
    let _ = handle.join();
}
hi from main thread hi from spawned thread
use std::thread;

fn main() {
    let handle = thread::spawn(|| println!("hi from spawned thread"));

    println!("hi from main thread");
    let _ = handle.join();
}

Panic

Panic

Panic

Panic

Panic

Panic

Споделяне на стойности

1 2 3 4 5 6 7 8 9 10 11 12 13
use std::thread;

fn main() {
    let nums = (0..5).collect::<Vec<_>>();

    let handle = thread::spawn(|| {
        for i in &nums {
            println!("number {}", i);
        }
    });

    let _ = handle.join();
}

Споделяне на стойности

1 2 3 4 5 6 7 8 9 10 11 12 13
use std::thread;

fn main() {
    let nums = (0..5).collect::<Vec<_>>();

    let handle = thread::spawn(|| {
        for i in &nums {
            println!("number {}", i);
        }
    });

    let _ = handle.join();
}
error[E0373]: closure may outlive the current function, but it borrows `nums`, which is owned by the current function --> src/bin/main_70d75afc36316cf9348e1b9a1b43baada82fd5b2.rs:6:32 | 6 | let handle = thread::spawn(|| { | ^^ may outlive borrowed value `nums` 7 | for i in &nums { | ---- `nums` is borrowed here | note: function requires argument type to outlive `'static` --> src/bin/main_70d75afc36316cf9348e1b9a1b43baada82fd5b2.rs:6:18 | 6 | let handle = thread::spawn(|| { | __________________^ 7 | | for i in &nums { 8 | | println!("number {}", i); 9 | | } 10| | }); | |______^ help: to force the closure to take ownership of `nums` (and any other referenced variables), use the `move` keyword | 6 | let handle = thread::spawn(move || { | ^^^^^^^
use std::thread;

fn main() {
    let nums = (0..5).collect::>();

    let handle = thread::spawn(|| {
        for i in &nums {
            println!("number {}", i);
        }
    });

    let _ = handle.join();
}

Споделяне на стойности

Споделяне на стойности

Споделяне на стойности

1 2 3 4
pub fn spawn<F, T>(f: F) -> JoinHandle<T> where
    F: FnOnce() -> T,
    F: Send + 'static,
    T: Send + 'static

Споделяне на стойности

Можем да преместим стойността в новата нишка

1 2 3 4 5 6 7 8 9 10 11 12 13
use std::thread;

fn main() {
    let nums = (0..5).collect::<Vec<_>>();

    let handle = thread::spawn(move || {
        for i in &nums {
            println!("number {}", i);
        }
    });

    let _ = handle.join();
}
number 0 number 1 number 2 number 3 number 4
use std::thread;

fn main() {
    let nums = (0..5).collect::>();

    let handle = thread::spawn(move || {
        for i in &nums {
            println!("number {}", i);
        }
    });

    let _ = handle.join();
}

Споделяне между няколко нишки

Как бихме споделили стойност между няколко нишки?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18
use std::thread;

fn main() {
    let nums = (0..5).collect::<Vec<_>>();

    let handles = (0..2).map(|_| {
            thread::spawn(move || {
                for i in &nums {
                    println!("number {}", i);
                }
            })
        })
        .collect::<Vec<_>>();

    for h in handles {
        let _ = h.join();
    }
}

Очевидно прехвърляне на собственост няма да работи - трябва ни референция!

Споделяне между няколко нишки

Вариант едно: референция, но статична!

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
use std::thread;

fn main() {
    let boxed_nums = (0..5).collect::<Box<[_]>>();
    let nums: &'static [_] = Box::leak(boxed_nums);

    let handles = (0..2).map(|_| {
            thread::spawn(move || {
                for i in nums {
                    println!("number {}", i);
                }
            })
        })
        .collect::<Vec<_>>();

    for h in handles {
        let _ = h.join();
    }
}
number 0 number 1 number 2 number 3 number 4 number 0 number 1 number 2 number 3 number 4
use std::thread;

fn main() {
    let boxed_nums = (0..5).collect::>();
    let nums: &'static [_] = Box::leak(boxed_nums);

    let handles = (0..2).map(|_| {
            thread::spawn(move || {
                for i in nums {
                    println!("number {}", i);
                }
            })
        })
        .collect::>();

    for h in handles {
        let _ = h.join();
    }
}

Споделяне между няколко нишки

Вариант едно: референция, но статична!

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
use std::thread;

fn main() {
    let boxed_nums = (0..5).collect::<Box<[_]>>();
    let nums: &'static [_] = Box::leak(boxed_nums);

    let handles = (0..2).map(|_| {
            thread::spawn(move || {
                for i in nums {
                    println!("number {}", i);
                }
            })
        })
        .collect::<Vec<_>>();

    for h in handles {
        let _ = h.join();
    }
}
number 0 number 1 number 2 number 3 number 4 number 0 number 1 number 2 number 3 number 4
use std::thread;

fn main() {
    let boxed_nums = (0..5).collect::>();
    let nums: &'static [_] = Box::leak(boxed_nums);

    let handles = (0..2).map(|_| {
            thread::spawn(move || {
                for i in nums {
                    println!("number {}", i);
                }
            })
        })
        .collect::>();

    for h in handles {
        let _ = h.join();
    }
}

Забележка: не правете това!

Споделяне между няколко нишки

Можем да пробваме с Rc

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23
use std::rc::Rc;
use std::thread;

fn main() {
    let nums_vec = (0..5).collect::<Vec<_>>();
    let nums = Rc::new(nums_vec);

    let handles = (0..2)
        .map(|_| {
            let nums = Rc::clone(&nums);

            thread::spawn(move || {
                for i in &*nums {
                    println!("number {}", i);
                }
            })
        })
        .collect::<Vec<_>>();

    for h in handles {
        let _ = h.join();
    }
}

Споделяне между няколко нишки

Можем да пробваме с Rc

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23
use std::rc::Rc;
use std::thread;

fn main() {
    let nums_vec = (0..5).collect::<Vec<_>>();
    let nums = Rc::new(nums_vec);

    let handles = (0..2)
        .map(|_| {
            let nums = Rc::clone(&nums);

            thread::spawn(move || {
                for i in &*nums {
                    println!("number {}", i);
                }
            })
        })
        .collect::<Vec<_>>();

    for h in handles {
        let _ = h.join();
    }
}
error[E0277]: `std::rc::Rc<std::vec::Vec<i32>>` cannot be sent between threads safely --> src/bin/main_a64edfb8be3b487ac4411ab31257712271d4ea64.rs:12:13 | 12 | thread::spawn(move || { | ^^^^^^^^^^^^^ `std::rc::Rc<std::vec::Vec<i32>>` cannot be sent between threads safely | = help: within `[closure@src/bin/main_a64edfb8be3b487ac4411ab31257712271d4ea64.rs:12:27: 16:14 nums:std::rc::Rc<std::vec::Vec<i32>>]`, the trait `std::marker::Send` is not implemented for `std::rc::Rc<std::vec::Vec<i32>>` = note: required because it appears within the type `[closure@src/bin/main_a64edfb8be3b487ac4411ab31257712271d4ea64.rs:12:27: 16:14 nums:std::rc::Rc<std::vec::Vec<i32>>]`
use std::rc::Rc;
use std::thread;

fn main() {
    let nums_vec = (0..5).collect::>();
    let nums = Rc::new(nums_vec);

    let handles = (0..2)
        .map(|_| {
            let nums = Rc::clone(&nums);

            thread::spawn(move || {
                for i in &*nums {
                    println!("number {}", i);
                }
            })
        })
        .collect::>();

    for h in handles {
        let _ = h.join();
    }
}

Send и Sync

1 2 3 4
pub fn spawn<F, T>(f: F) -> JoinHandle<T> where
    F: FnOnce() -> T,
    F: Send + 'static,
    T: Send + 'static

Send и Sync

Send и Sync

Send и Sync

Send и Sync

Send и Sync

Send и Sync

Send и Sync

Send и Sync

Аuto traits

1
pub struct Token(u32);
pub struct Token(u32);
fn main() {}

Send и Sync

Аuto traits

1
pub struct Token(u32);
pub struct Token(u32);
fn main() {}

Auto trait docs

Send и Sync

Send

Send и Sync

Send

Send и Sync

Send

Send и Sync

Send

Send и Sync

Send

Send и Sync

Send

Send + Sync

Sync

Send + Sync

Sync

Send + Sync

Sync

Send + Sync

Sync

Send + Sync

Sync

Send + Sync

Sync

Send + Sync

Sync

Send + Sync

Sync

Това значи ли че Vec<T> е Sync?

Send + Sync

Sync

Това значи ли че Vec<T> е Sync?

Send + Sync

Sync

Това значи ли че Vec<T> е Sync?

Send + Sync

Sync

Това значи ли че Vec<T> е Sync?

Send + Sync

Sync

Това значи ли че Vec<T> е Sync?

Send + Sync

Unsafe traits

1 2 3 4
struct MyBox(*mut u8);

unsafe impl Send for MyBox {}
unsafe impl Sync for MyBox {}
fn main() {}
struct MyBox(*mut u8);

unsafe impl Send for MyBox {}
unsafe impl Sync for MyBox {}

Send + Sync

Деимплементация

1 2 3 4 5 6 7
// Само на nightly
#![feature(optin_builtin_traits)]

struct SpecialToken(u8);

impl !Send for SpecialToken {}
impl !Sync for SpecialToken {}

Send + Sync

Деимплементация

1 2 3 4 5 6 7
// Само на nightly
#![feature(optin_builtin_traits)]

struct SpecialToken(u8);

impl !Send for SpecialToken {}
impl !Sync for SpecialToken {}

Send + Sync

Деимплементация

1 2 3 4 5 6 7
// Само на nightly
#![feature(optin_builtin_traits)]

struct SpecialToken(u8);

impl !Send for SpecialToken {}
impl !Sync for SpecialToken {}

Send + Sync

Деимплементация

Хак за stable

1 2 3
use std::marker::PhantomData;

struct SpecialToken(u8, PhantomData<*const ()>);
fn main() {}
use std::marker::PhantomData;

struct SpecialToken(u8, PhantomData<*const ()>);

Arc

Да се върнем на кода, който не се компилираше

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22
use std::rc::Rc;
use std::thread;

fn main() {
    let nums = Rc::new((0..5).collect::<Vec<_>>());

    let handles = (0..2)
        .map(|_| {
            let nums = Rc::clone(&nums);

            thread::spawn(move || {
                for i in &*nums {
                    println!("number {}", i);
                }
            })
        })
        .collect::<Vec<_>>();

    for h in handles {
        let _ = h.join();
    }
}
error[E0277]: `std::rc::Rc<std::vec::Vec<i32>>` cannot be sent between threads safely --> src/bin/main_8e08fc7cb2794a1557ea5e32c268dd5d930f61fb.rs:11:13 | 11 | thread::spawn(move || { | ^^^^^^^^^^^^^ `std::rc::Rc<std::vec::Vec<i32>>` cannot be sent between threads safely | = help: within `[closure@src/bin/main_8e08fc7cb2794a1557ea5e32c268dd5d930f61fb.rs:11:27: 15:14 nums:std::rc::Rc<std::vec::Vec<i32>>]`, the trait `std::marker::Send` is not implemented for `std::rc::Rc<std::vec::Vec<i32>>` = note: required because it appears within the type `[closure@src/bin/main_8e08fc7cb2794a1557ea5e32c268dd5d930f61fb.rs:11:27: 15:14 nums:std::rc::Rc<std::vec::Vec<i32>>]`
use std::rc::Rc;
use std::thread;

fn main() {
    let nums = Rc::new((0..5).collect::>());

    let handles = (0..2)
        .map(|_| {
            let nums = Rc::clone(&nums);

            thread::spawn(move || {
                for i in &*nums {
                    println!("number {}", i);
                }
            })
        })
        .collect::>();

    for h in handles {
        let _ = h.join();
    }
}

Arc

Решението е да заменим std::rc::Rc с std::sync::Arc

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22
use std::sync::Arc;
use std::thread;

fn main() {
    let nums = Arc::new((0..5).collect::<Vec<_>>());

    let handles = (0..2)
        .map(|_| {
            let nums = Arc::clone(&nums);

            thread::spawn(move || {
                for i in &*nums {
                    println!("number {}", i);
                }
            })
        })
        .collect::<Vec<_>>();

    for h in handles {
        let _ = h.join();
    }
}
number 0 number 1 number 2 number 3 number 4 number 0 number 1 number 2 number 3 number 4
use std::sync::Arc;
use std::thread;

fn main() {
    let nums = Arc::new((0..5).collect::>());

    let handles = (0..2)
        .map(|_| {
            let nums = Arc::clone(&nums);

            thread::spawn(move || {
                for i in &*nums {
                    println!("number {}", i);
                }
            })
        })
        .collect::>();

    for h in handles {
        let _ = h.join();
    }
}

Arc

Arc

Arc

Arc

Arc

Примитиви за синхронизация

Примитиви за синхронизация

Стандартния пример за грешен многонишков алгоритъм:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
let v = Arc::new((0..100).collect::<Vec<_>>());
let mut sum = 0;

let t1 = {
    let v = Arc::clone(&v);
    thread::spawn(|| for i in &v[0..50] { sum += i; })
};

let t2 = {
    let v = Arc::clone(&v);
    thread::spawn(|| for i in &v[51..100] { sum += i; })
};

let _ = t1.join();
let _ = t2.join();
println!("sum: {}", sum);
error[E0373]: closure may outlive the current function, but it borrows `v`, which is owned by the current function --> src/bin/main_9ab818289d8f7ec0e1d87f97c7a3c68bc6ef4b11.rs:9:19 | 9 | thread::spawn(|| for i in &v[0..50] { sum += i; }) | ^^ - `v` is borrowed here | | | may outlive borrowed value `v` | note: function requires argument type to outlive `'static` --> src/bin/main_9ab818289d8f7ec0e1d87f97c7a3c68bc6ef4b11.rs:9:5 | 9 | thread::spawn(|| for i in &v[0..50] { sum += i; }) | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ help: to force the closure to take ownership of `v` (and any other referenced variables), use the `move` keyword | 9 | thread::spawn(move || for i in &v[0..50] { sum += i; }) | ^^^^^^^ error[E0373]: closure may outlive the current function, but it borrows `sum`, which is owned by the current function --> src/bin/main_9ab818289d8f7ec0e1d87f97c7a3c68bc6ef4b11.rs:9:19 | 9 | thread::spawn(|| for i in &v[0..50] { sum += i; }) | ^^ --- `sum` is borrowed here | | | may outlive borrowed value `sum` | note: function requires argument type to outlive `'static` --> src/bin/main_9ab818289d8f7ec0e1d87f97c7a3c68bc6ef4b11.rs:9:5 | 9 | thread::spawn(|| for i in &v[0..50] { sum += i; }) | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ help: to force the closure to take ownership of `sum` (and any other referenced variables), use the `move` keyword | 9 | thread::spawn(move || for i in &v[0..50] { sum += i; }) | ^^^^^^^ error[E0499]: cannot borrow `sum` as mutable more than once at a time --> src/bin/main_9ab818289d8f7ec0e1d87f97c7a3c68bc6ef4b11.rs:14:19 | 9 | thread::spawn(|| for i in &v[0..50] { sum += i; }) | -------------------------------------------------- | | | | | | | first borrow occurs due to use of `sum` in closure | | first mutable borrow occurs here | argument requires that `sum` is borrowed for `'static` ... 14 | thread::spawn(|| for i in &v[51..100] { sum += i; }) | ^^ --- second borrow occurs due to use of `sum` in closure | | | second mutable borrow occurs here error[E0373]: closure may outlive the current function, but it borrows `v`, which is owned by the current function --> src/bin/main_9ab818289d8f7ec0e1d87f97c7a3c68bc6ef4b11.rs:14:19 | 14 | thread::spawn(|| for i in &v[51..100] { sum += i; }) | ^^ - `v` is borrowed here | | | may outlive borrowed value `v` | note: function requires argument type to outlive `'static` --> src/bin/main_9ab818289d8f7ec0e1d87f97c7a3c68bc6ef4b11.rs:14:5 | 14 | thread::spawn(|| for i in &v[51..100] { sum += i; }) | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ help: to force the closure to take ownership of `v` (and any other referenced variables), use the `move` keyword | 14 | thread::spawn(move || for i in &v[51..100] { sum += i; }) | ^^^^^^^ error[E0373]: closure may outlive the current function, but it borrows `sum`, which is owned by the current function --> src/bin/main_9ab818289d8f7ec0e1d87f97c7a3c68bc6ef4b11.rs:14:19 | 14 | thread::spawn(|| for i in &v[51..100] { sum += i; }) | ^^ --- `sum` is borrowed here | | | may outlive borrowed value `sum` | note: function requires argument type to outlive `'static` --> src/bin/main_9ab818289d8f7ec0e1d87f97c7a3c68bc6ef4b11.rs:14:5 | 14 | thread::spawn(|| for i in &v[51..100] { sum += i; }) | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ help: to force the closure to take ownership of `sum` (and any other referenced variables), use the `move` keyword | 14 | thread::spawn(move || for i in &v[51..100] { sum += i; }) | ^^^^^^^ error[E0502]: cannot borrow `sum` as immutable because it is also borrowed as mutable --> src/bin/main_9ab818289d8f7ec0e1d87f97c7a3c68bc6ef4b11.rs:19:21 | 9 | thread::spawn(|| for i in &v[0..50] { sum += i; }) | -------------------------------------------------- | | | | | | | first borrow occurs due to use of `sum` in closure | | mutable borrow occurs here | argument requires that `sum` is borrowed for `'static` ... 19 | println!("sum: {}", sum); | ^^^ immutable borrow occurs here
use std::sync::Arc;
use std::thread;
fn main() {
let v = Arc::new((0..100).collect::>());
let mut sum = 0;

let t1 = {
    let v = Arc::clone(&v);
    thread::spawn(|| for i in &v[0..50] { sum += i; })
};

let t2 = {
    let v = Arc::clone(&v);
    thread::spawn(|| for i in &v[51..100] { sum += i; })
};

let _ = t1.join();
let _ = t2.join();
println!("sum: {}", sum);
}

Примитиви за синхронизация

Можем ли да го накараме да работи?

Примитиви за синхронизация

Можем ли да го накараме да работи?

Примитиви за синхронизация

Можем ли да го накараме да работи?

Примитиви за синхронизация

Можем ли да го накараме да работи?

Примитиви за синхронизация

Можем да го накараме да работи

Примитиви за синхронизация

Можем да го накараме да работи

Примитиви за синхронизация

Можем да го накараме да работи

Примитиви за синхронизация

Можем да го накараме да работи

Примитиви за синхронизация

Можем да го накараме да работи

Mutex

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
use std::sync::Mutex;

fn main() {
    // мутекса опакова стойността, която предпазва
    let mutex = Mutex::new(10);

    {
        // заключваме мутекса
        // `lock` е умен указател с deref до `&T` и `&mut T`
        let mut lock = mutex.lock().unwrap();
        *lock += 32;

        // мутекса се отключва когато `lock` се деалокира
    }
}
use std::sync::Mutex;

fn main() {
    // мутекса опакова стойността, която предпазва
    let mutex = Mutex::new(10);

    {
        // заключваме мутекса
        // `lock` е умен указател с deref до `&T` и `&mut T`
        let mut lock = mutex.lock().unwrap();
        *lock += 32;

        // мутекса се отключва когато `lock` се деалокира
    }
}

Mutex

Mutex

Mutex

Mutex

Mutex

Mutex

Mutex

Mutex

Mutex

Mutex

Mutex

Mutex

Panic

Mutex

Panic

Mutex

Panic

Mutex

Panic

Mutex

Panic

Mutex

Panic

RwLock

RwLock

RwLock

RwLock

Mutex срещу RwLock

Mutex срещу RwLock

Mutex срещу RwLock

Mutex срещу RwLock

Condvar

Condvar

Condvar

Condvar

Condvar

1
let pair = Arc::new((Mutex::new(false), Condvar::new()));
use std::sync::{Arc, Mutex, Condvar};
fn main() {
    let pair = Arc::new((Mutex::new(false), Condvar::new()));
}

1 2 3 4 5 6 7
// Thread A
let (lock, cvar) = &*pair;
let mut started = lock.lock().unwrap();

while !*started {
    started = cvar.wait(started).unwrap();
}

Copy

use std::sync::{Arc, Mutex, Condvar};
fn main() {
let pair = Arc::new((Mutex::new(false), Condvar::new()));
// Thread A
let (lock, cvar) = &*pair;
let mut started = lock.lock().unwrap();

while !*started {
    started = cvar.wait(started).unwrap();
}
}

1 2 3 4 5 6
// Thread B
let (lock, cvar) = &*pair;
let mut started = lock.lock().unwrap();

*started = true;
cvar.notify_one();

Copy

use std::sync::{Arc, Mutex, Condvar};
fn main() {
let pair = Arc::new((Mutex::new(false), Condvar::new()));
// Thread B
let (lock, cvar) = &*pair;
let mut started = lock.lock().unwrap();

*started = true;
cvar.notify_one();
}

Други примитиви за синхронизация

Други примитиви за синхронизация

Други примитиви за синхронизация

Атомарни числа

Атомарни числа

Атомарни числа

Атомарни числа

Атомарни числа

Атомарни числа

Атомарни числа

Атомарни числа

Атомарни числа

Атомарни числа

Канали

MPSC

Канали

Go-lang motto

Don't communicate by sharing memory,
share memory by communicating

Канали в стандартната библиотека

1 2 3 4 5 6 7 8 9 10 11 12
use std::sync::mpsc;
use std::thread;

fn main() {
    let (sender, receiver) = mpsc::channel();

    thread::spawn(move || {
        sender.send(10).unwrap();
    });

    println!("received {}", receiver.recv().unwrap());
}
received 10
use std::sync::mpsc;
use std::thread;

fn main() {
    let (sender, receiver) = mpsc::channel();

    thread::spawn(move || {
        sender.send(10).unwrap();
    });

    println!("received {}", receiver.recv().unwrap());
}

Типове канали

Неограничен канал

Типове канали

Неограничен канал

Типове канали

Неограничен канал

Типове канали

Неограничен канал

Типове канали

Неограничен канал

Типове канали

Неограничен канал

1 2 3 4 5 6 7 8 9 10 11
let (sender, receiver) = mpsc::channel();

thread::spawn(move || {
    sender.send(1).unwrap();
    sender.send(2).unwrap();
    sender.send(3).unwrap();
});

assert_eq!(receiver.recv().unwrap(), 1);
assert_eq!(receiver.recv().unwrap(), 2);
assert_eq!(receiver.recv().unwrap(), 3);
use std::sync::mpsc;
use std::thread;
fn main() {
let (sender, receiver) = mpsc::channel();

thread::spawn(move || {
    sender.send(1).unwrap();
    sender.send(2).unwrap();
    sender.send(3).unwrap();
});

assert_eq!(receiver.recv().unwrap(), 1);
assert_eq!(receiver.recv().unwrap(), 2);
assert_eq!(receiver.recv().unwrap(), 3);
}

Типове канали

Oграничен канал

Типове канали

Oграничен канал

Типове канали

Oграничен канал

Типове канали

Oграничен канал

Типове канали

Oграничен канал

Типове канали

Ограничен канал

1 2 3 4 5 6 7 8 9 10 11 12
let (sender, receiver) = mpsc::sync_channel(1);

thread::spawn(move || {
    // записва съобщението и връща веднага
    sender.send(1).unwrap();

    // ще блокира докато главната нишка не извика `receiver.recv()`
    sender.send(2).unwrap();
});

assert_eq!(receiver.recv().unwrap(), 1);
assert_eq!(receiver.recv().unwrap(), 2);
use std::sync::mpsc;
use std::thread;
fn main() {
let (sender, receiver) = mpsc::sync_channel(1);

thread::spawn(move || {
    // записва съобщението и връща веднага
    sender.send(1).unwrap();

    // ще блокира докато главната нишка не извика `receiver.recv()`
    sender.send(2).unwrap();
});

assert_eq!(receiver.recv().unwrap(), 1);
assert_eq!(receiver.recv().unwrap(), 2);
}

Множество изпращачи

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
let (sender, receiver) = mpsc::channel();
let sender2 = sender.clone();

thread::spawn(move || {
    sender.send(1).unwrap();
    sender.send(2).unwrap();
});

thread::spawn(move || {
    sender2.send(3).unwrap();
    sender2.send(4).unwrap();
});

println!("{} {} {} {}",
    receiver.recv().unwrap(), receiver.recv().unwrap(),
    receiver.recv().unwrap(), receiver.recv().unwrap());
3 4 1 2
use std::sync::mpsc;
use std::thread;
fn main() {
let (sender, receiver) = mpsc::channel();
let sender2 = sender.clone();

thread::spawn(move || {
    sender.send(1).unwrap();
    sender.send(2).unwrap();
});

thread::spawn(move || {
    sender2.send(3).unwrap();
    sender2.send(4).unwrap();
});

println!("{} {} {} {}",
    receiver.recv().unwrap(), receiver.recv().unwrap(),
    receiver.recv().unwrap(), receiver.recv().unwrap());
}

Sender

Методи

1 2 3
// изпраща `t`
// връща грешка ако получателят е бил унищожен
fn send(&self, t: T) -> Result<(), SendError<T>>

Sender

Методи

1 2 3 4 5 6 7 8 9 10 11
let (sender, receiver) = mpsc::channel();

assert_eq!(sender.send(12), Ok(()));

// унищожаваме получателя
// съобщението `12` никога няма да бъде получено
mem::drop(receiver);

// грешка - получателя е унищожен
// можем да си върнем съобщението `23` от грешката
assert_eq!(sender.send(23), Err(SendError(23)));
use std::mem;
use std::sync::mpsc::{self, SendError};
fn main() {
let (sender, receiver) = mpsc::channel();

assert_eq!(sender.send(12), Ok(()));

// унищожаваме получателя
// съобщението `12` никога няма да бъде получено
mem::drop(receiver);

// грешка - получателя е унищожен
// можем да си върнем съобщението `23` от грешката
assert_eq!(sender.send(23), Err(SendError(23)));
}

SyncSender

Методи

1 2 3 4 5
// блокира ако буфера е пълен
fn send(&self, t: T) -> Result<(), SendError<T>>

// връща грешка ако буфера е пълен или получателят е бил унищожен
fn try_send(&self, t: T) -> Result<(), TrySendError<T>>

SyncSender

Методи

1 2 3 4 5 6 7 8
let (sender, receiver) = mpsc::sync_channel(1);

assert_eq!(sender.try_send(12), Ok(()));
assert_eq!(sender.try_send(23), Err(TrySendError::Full(23)));

mem::drop(receiver);

assert_eq!(sender.try_send(23), Err(TrySendError::Disconnected(23)));
use std::mem;
use std::sync::mpsc::{self, TrySendError};
fn main() {
let (sender, receiver) = mpsc::sync_channel(1);

assert_eq!(sender.try_send(12), Ok(()));
assert_eq!(sender.try_send(23), Err(TrySendError::Full(23)));

mem::drop(receiver);

assert_eq!(sender.try_send(23), Err(TrySendError::Disconnected(23)));
}

Множество получатели

Множество получатели

Множество получатели

Множество получатели

Receiver

Методи

1 2 3 4 5 6 7 8 9 10 11
// блокира докато не получи съобщение
// връща грешка ако всички изпращачи са унищожени
fn recv(&self) -> Result<T, RecvError>

// не блокира
// връща грешка ако всички изпращачи са унищожени или няма съобщение в опашката
fn try_recv(&self) -> Result<T, TryRecvError>

// блокира за определено време
// връща грешка ако всички изпращачи са унищожени или е изтекло времето
fn recv_timeout(&self, timeout: Duration) -> Result<T, RecvTimeoutError>

Receiver

Методи

1 2 3 4 5 6 7 8 9 10 11
let (sender, receiver) = mpsc::channel();

thread::spawn(move || {
    for i in (0..50).rev() {
        sender.send(i).unwrap();
    }
});

while let Ok(msg) = receiver.recv() {
    println!("received {}", msg);
}
use std::sync::mpsc;
use std::thread;
fn main() {
let (sender, receiver) = mpsc::channel();

thread::spawn(move || {
    for i in (0..50).rev() {
        sender.send(i).unwrap();
    }
});

while let Ok(msg) = receiver.recv() {
    println!("received {}", msg);
}
}

Receiver

Итератори

1 2 3 4 5 6 7 8 9 10 11
let (sender, receiver) = mpsc::channel();

thread::spawn(move || {
    for i in (0..50).rev() {
        sender.send(i).unwrap();
    }
});

for msg in receiver.iter() {
    println!("received {}", msg);
}
use std::sync::mpsc;
use std::thread;
fn main() {
let (sender, receiver) = mpsc::channel();

thread::spawn(move || {
    for i in (0..50).rev() {
        sender.send(i).unwrap();
    }
});

for msg in receiver.iter() {
    println!("received {}", msg);
}
}

Други структури

Други библиотеки

Въпроси