advance/concurrency-with-threads/sync2

[giscus[bot]]

advance/concurrency-with-threads/sync2

https://course.rs/advance/concurrency-with-threads/sync2.html

[zhangdekui]

我看到这里了耶, 💯

[buoge]

哦豁 +1

[zippermonkey]

✌️

[jackwangwj]

+1

[sfgwfn1402]

优秀

[Nie-Tianyi]

你会再次看到这里的

[lijing-2008]

这一节跳过了

[gongxufan]

线程屏障这里，使用了引用技术。rust用共享变量处理心智负担真重啊

[wuyuz]

我用Go语言跑了一下自增的例子，大概花费了200ms，但是用rust里面的Atomic要花费4s左右，怎么差这么多哦

[WenqingZong]

我使用了--release之后，结果是2.6s，依旧没有go快？

[suncunqian]

我不信， 我自己用go写了个 耗时比Rust多

import (
	"fmt"
	"sync"
	"time"
)

const NTIMES uint64 = 10000000
const NTHRESDS uint = 10

var wg sync.WaitGroup
var mutex sync.Mutex
var R uint64 = 0

func add_n_time() {
	defer wg.Done()
	for i := uint64(0); i < NTIMES; i++ {
		mutex.Lock()
		R += 1
		mutex.Unlock()
	}
}

func main() {
	startTime := time.Now()
	for i := uint(0); i < NTHRESDS; i++ {
		wg.Add(1)
		go add_n_time()
	}
	wg.Wait()
	endTime := time.Now().Sub(startTime)
	fmt.Printf("total time spend: %s, R: %d", endTime, R)
}

total time spend: 3.9255068s, R: 100000000

同样的实现 我用Rust 1s+

[FlippedOvO]

用rust跑了一遍 时间是6s 多。 看了评论以后使用release去跑 还是有5s多

检查了一下才发现多打了一个0

[yzxz1996]

@suncunqian 你这不是用的go的mutex嘛，我猜楼主比较的是go与rust的atomic?

[W2Q3Q1]

代码发来。我怀疑你go用的是协程。

[yeyujingren]

use std::ops::Sub;
use std::sync::{Mutex, Arc};
use std::thread::{self, JoinHandle};
use std::time::Instant;
const N_TIMES: u64 = 10_000_000;
const N_THREADS: usize = 10;

fn add_n_times(n: u64, counter: Arc<Mutex<u64>>) -> JoinHandle<()> {
    thread::spawn(move || {
        for _ in 0..n {
            let mut num = counter.lock().unwrap();
            *num += 1;
        }
    })
}


fn main() {
    let mut counter: Arc<Mutex<u64>> = Arc::new(Mutex::new(0));
    let s = Instant::now();
    let mut threads = Vec::with_capacity(N_THREADS);
    for _ in 0..N_THREADS {
        threads.push(add_n_times(N_TIMES, counter.clone()))
    }

    for thread in threads {
        thread.join().unwrap();
    }
    assert_eq!(N_TIMES * N_THREADS as u64, *counter.lock().unwrap());
    println!("{:?}",Instant::now().sub(s));
}

针对与第一个例子的Mutex实现，在控制变量的情况下，执行时间相差不止40%，甚至与10倍不止，是我哪儿写的有问题吗？初学者，有些地方写的粗糙，还望包涵😂

[Wjinlei]

因为你在for循环里面，锁又不会被释放，它只有for循环结束后，才会释放，因此别的线程也拿不到，所以我认为放for循环里面，无意义。

[aardwolfchen]

放在循环内，每次都有请求锁和释放锁的开销

[cy-arctique]

我跟着@Wjinlei 修改了 层主的代码, mutex跑起来怎么比atomic的快?

[zzttzzmyswy]

@cy-arctique 我认为按照@Wjinlei 改后降级了互斥获取的消耗，但atomic每次写仍然都需要原子操作。所以mutex跑的更快是正常的。
而回答楼主的问题：为啥性能差距并不是那么适合40%。我认为这和同时并行线程数以及频繁获取锁/原子操作的次数与频率有关
这个是我的代码以及在我的机器上的测试结果

use std::ops::Sub;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::Instant;
#[derive(Clone)]
struct NInfo {
    n_times: u64,
    n_threads: usize,
}

fn add_n_times(n: u64, counter: Arc<Mutex<u64>>) -> JoinHandle<()> {
    thread::spawn(move || {
        for _ in 0..n {
            let mut num = counter.lock().unwrap();
            *num += 1;
        }
    })
}
fn add_n_times_atomic(n: u64, counter: Arc<AtomicU64>) -> JoinHandle<()> {
    thread::spawn(move || {
        for _ in 0..n {
            counter.fetch_add(1, Ordering::Relaxed);
        }
    })
}

fn mutex_test(test_info:NInfo) -> std::time::Duration {
    let counter: Arc<Mutex<u64>> = Arc::new(Mutex::new(0));
    let mut threads = Vec::with_capacity(test_info.n_threads);
    let s = Instant::now();
    for _ in 0..test_info.n_threads {
        threads.push(add_n_times(test_info.n_times, counter.clone()))
    }

    for thread in threads {
        thread.join().unwrap();
    }
    let use_time: std::time::Duration = Instant::now().sub(s);
    assert_eq!(test_info.n_times * test_info.n_threads as u64, *counter.lock().unwrap());
    use_time
}

fn atomic_test(test_info:NInfo) -> std::time::Duration {
    let counter: Arc<AtomicU64> = Arc::new(AtomicU64::new(0));
    let mut threads = Vec::with_capacity(test_info.n_threads);
    let s = Instant::now();
    for _ in 0..test_info.n_threads {
        threads.push(add_n_times_atomic(test_info.n_times, counter.clone()))
    }

    for thread in threads {
        thread.join().unwrap();
    }
    let use_time = Instant::now().sub(s);
    assert_eq!(test_info.n_times * test_info.n_threads as u64, counter.load(Ordering::Relaxed));
    use_time
}

fn test(test_info:NInfo) {
    let mut d_time=0f64;
    for _ in 0..10 {
        let time_mutex= mutex_test(test_info.clone());
        let time_atomic = atomic_test(test_info.clone());
        d_time += (time_mutex.as_micros() as f64) / (time_atomic.as_micros() as f64);
    }
    d_time /= 10 as f64;
    println!("start test times: {}, threads: {}, time_mutex/time_atomic: {d_time:?}",test_info.n_times,test_info.n_threads);
}

fn main() {
    test(NInfo{n_times:100,n_threads:10});
    test(NInfo{n_times:100,n_threads:1_000});
    test(NInfo{n_times:500,n_threads:10});
    test(NInfo{n_times:500,n_threads:1_000});
    test(NInfo{n_times:800,n_threads:10});
    test(NInfo{n_times:800,n_threads:1_000});
    test(NInfo{n_times:1000,n_threads:10});
    test(NInfo{n_times:1000,n_threads:1_000});
    test(NInfo{n_times:1500,n_threads:10});
    test(NInfo{n_times:1500,n_threads:1_000});
}

$ ./target/release/test1
start test times: 100, threads: 10, time_mutex/time_atomic: 1.0904351785611595
start test times: 100, threads: 1000, time_mutex/time_atomic: 1.0173308709809075
start test times: 500, threads: 10, time_mutex/time_atomic: 1.2163311505322552
start test times: 500, threads: 1000, time_mutex/time_atomic: 1.2318013220956812
start test times: 800, threads: 10, time_mutex/time_atomic: 2.312161001986228
start test times: 800, threads: 1000, time_mutex/time_atomic: 2.2129482151782534
start test times: 1000, threads: 10, time_mutex/time_atomic: 2.713148500213278
start test times: 1000, threads: 1000, time_mutex/time_atomic: 3.212438092710164
start test times: 1500, threads: 10, time_mutex/time_atomic: 5.892830378536266
start test times: 1500, threads: 1000, time_mutex/time_atomic: 4.068150602162824

[zzttzzmyswy]

当然了，这个测试代码并没有考虑到线程创建等其他无关消耗。只能做一个参考。并不能很有力地说明观点

[wreqwewqe]

突然就蹦出来一个变量，也不知道是啥，用方法也不说是干啥的

[ghost]

确实

[ygnh-leo]

创建线程的性能

因此线程的创建耗时并不是不可忽略的
-> 因此线程的创建耗时并不是完全可忽略的

根据上下文的语境，这里应该是不应该完全忽略。

[horsefaced]

这章最后一个例子用的不好, 莫明其妙的上了之前段落都没有提到过的自旋锁.

[kingwingfly]

GPT打死不承认Atomic跨线程需要Arc，可恶

[cowboyliao]

这是rust程序员不会被取代的意思吗

[kingwingfly]

gpt4.0没用过，不知道，但3.5确实有很多进阶的东西是不会的。这憨憨玩意儿离了程序员啥都干不好。而且它还会信誓旦旦地乱搞……

[CubeSugarCheese]

确实不需要，因为 Arc<T> 只是用来满足 'static 的约束罢了
用 thread::scoped 就能不使用 Arc<T>

let atomic_bool = AtomicBool::new(true);
std::thread::scope(|s| {
    s.spawn(||{ let b = atomic_bool.load(Ordering::Relaxed);println!("thread1:{b}") });
    s.spawn(||{ let b = atomic_bool.load(Ordering::Relaxed);println!("thread2:{b}") });
});

[kingwingfly]

你这里是 move 了

[kingwingfly]

Oh，好像确实，我脑子瓦特了

[qianyiwangmeng]

Barrier是什么，为什么每一个循环都要 clone一下，这个b.wait()有什么用吗

[yuhao00]

这些例子是std文档注释里的...

[cdog-shen]

使用 Arc::clone(&barrier) 进行克隆，这样每个线程都有自己的 Arc 实例，能够安全地调用 wait 方法。不克隆的话arc的所有权会转移，其他线程用不了

[etigan]

能详细说下 thread_local! 宏么？
1.为什么闭包需要使用 move 传入FOO的所有权，去掉 move 也可以正常跑呀
2.线程结束 FOO 会不会被 drop 掉，如果不好，那么为什么子线程改为 3 了，main函数里面 为什么还是2
3.注释说 每个线程开始时都会拿到线程局部变量的FOO的初始值 是每个线程都会有一个 FOO 变量么？
只给一句注释，太笼统了

[etigan]

对于某些没使用或者没讲过的对象能不能给个注释呀，这点可能对新手不太友好，每次一边看书，还要一边百度

[iamtaojin]

不科学，为什么我的使用原子类型耗时多？

use std::ops::Sub;
use std::sync::{
    atomic::{AtomicU64, Ordering},
    Arc, Mutex,
};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};

const N_TIMES: u64 = 10000000;
const N_THREADS: usize = 10;

static R: AtomicU64 = AtomicU64::new(0);

fn add_n_times(n: u64) -> JoinHandle<()> {
    thread::spawn(move || {
        for _ in 0..n {
            R.fetch_add(1, Ordering::Relaxed);
        }
    })
}

fn main() {
    println!("use mutex: {:?}", use_mutex());
    println!("use automic: {:?}", use_automic());
}

fn use_automic() -> Duration {
    let s = Instant::now();
    let mut threads = Vec::with_capacity(N_THREADS);

    for _ in 0..N_THREADS {
        threads.push(add_n_times(N_TIMES));
    }

    for thread in threads {
        thread.join().unwrap();
    }

    assert_eq!(N_TIMES * N_THREADS as u64, R.load(Ordering::Relaxed));

    Instant::now().sub(s)
}
fn use_mutex() -> Duration {
    let s = Instant::now();
    let mut threads = Vec::with_capacity(N_THREADS);
    let counter: Arc<Mutex<u64>> = Arc::new(Mutex::new(0));

    for _ in 0..N_THREADS {
        let lock = counter.clone();
        threads.push(thread::spawn(move || {
            let mut num = lock.lock().unwrap();
            while *num < N_TIMES * N_THREADS as u64 {
                *num += 1;
            }
        }));
    }

    for thread in threads {
        thread.join().unwrap();
    }

    assert_eq!(N_TIMES * N_THREADS as u64, *counter.lock().unwrap());

    Instant::now().sub(s)
}

[superggn]

我本地跑出来的数据有点夸张了... CAS 版本 2s 左右， Mutex 版本 20s+...

use std::ops::Sub;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::thread::{self, JoinHandle};
use std::time::Instant;

const N_TIMES: u64 = 10000000;
const N_THREADS: usize = 10;

static R: AtomicU64 = AtomicU64::new(0);

fn add_n_times(n: u64) -> JoinHandle<()> {
    thread::spawn(move || {
        for _ in 0..n {
            R.fetch_add(1, Ordering::Relaxed);
        }
    })
}
fn add_n_times_mutex(n: u64, num: Arc<Mutex<u64>>) -> JoinHandle<()> {
    thread::spawn(move || {
        for _ in 0..n {
            let mut actual_num = num.lock().unwrap();
            *actual_num += 1;
        }
    })
}

fn main() {
    let s = Instant::now();
    let mut threads = Vec::with_capacity(N_THREADS);

    for _ in 0..N_THREADS {
        threads.push(add_n_times(N_TIMES));
    }

    for thread in threads {
        thread.join().unwrap();
    }

    assert_eq!(N_TIMES * N_THREADS as u64, R.load(Ordering::Relaxed));

    println!("CAS version: {:?}", Instant::now().sub(s));

    println!("================================================");

    let s2 = Instant::now();

    let res = Arc::new(Mutex::new(0));
    let mut threads = Vec::with_capacity(N_THREADS);

    for _ in 0..N_THREADS {
        let temp_res = res.clone();
        threads.push(add_n_times_mutex(N_TIMES, temp_res));
    }

    for thread in threads {
        thread.join().unwrap();
    }

    assert_eq!(N_TIMES * N_THREADS as u64, *res.lock().unwrap());
    println!("Mutex version: {:?}", Instant::now().sub(s2));
}

[lieren1922]

用release，Mutex耗时更短

[tzfun]

javaer也来学rust了，看到自增这个例子赶忙去用java写了个test对比了一下，rust release模式试了10次跑出来平均 1.9s，java试了10次跑出来平均 1.6s，这是为啥？

Rust 代码

#[test]
fn test_atomic() {
    static R: AtomicU64 = AtomicU64::new(0);
    const N: usize = 10;
    const TIMES: u64 = 10000000;
    let start = Instant::now();
    let mut threads = Vec::with_capacity(N);

    for _ in 0..N {
        threads.push(thread::spawn(move || {
            for _ in 0..TIMES {
                //  Ordering 是用于控制内存顺序
                R.fetch_add(1, Ordering::Relaxed);
            }
        }));
    }

    for thread in threads {
        thread.join().unwrap();
    }
    assert_eq!(N as u64 * TIMES, R.load(Ordering::Relaxed));
    println!("use {:?}", Instant::now().sub(start))
}

Java 代码

@Test
public void testAtomic() throws Exception {
    AtomicLong counter = new AtomicLong(0);
    final int N = 10;
    final long TIMES = 10000000;
    long start = System.currentTimeMillis();
    Thread[] threads = new Thread[N];

    for (int i = 0; i < N; i++) {
        Thread t = new Thread(() -> {
            for (long l = 0; l < TIMES; l++) {
                counter.addAndGet(1);
            }
        });
        t.start();
        threads[i] =t;
    }

    for (Thread t : threads) {
        t.join();
    }

    Assert.assertEquals(N * TIMES, counter.get());
    System.out.println("use " + (System.currentTimeMillis() - start) + " ms");
}

[ddx533534]

hello, i am 0, from main
hello, i am 0, from spawn
hello, i am 1, from main
hello, i am 1, from spawn
hello, i am 2, from main
hello, i am 2, from spawn
hello, i am 3, from main
hello, i am 3, from spawn
hello, i am 4, from main
hello, i am 4, from spawn
hello, i am 
Process finished with exit code 0

运行结果如上，真就 main线程执行完了就结束，子线程打印到一半还没打完...

[kingwingfly]

你知道莫名收到一个邮件，一会儿说自己是0，一会儿又是1，有多猎奇吗？😏

[chenchong22]

很不错的手册，很糟糕的教程

[kingwingfly]

可以 can can word。

https://github.com/kingwingfly/rust_tutorial

[MaJinjie]

厉害厉害，拨云见雾。

[ZhangJingYun2018]

条件控制那部分的执行顺序并不能保证，只是大概率主线程先执行。while有概率不执行。

[J0hNnY1ee]

6

[ZhangJingYun2018]

没有看懂内存顺序。例子和解释有点少 。。。。。。。

[PeishuaiLee]

要感谢：对于原子语义，可以在硬件层面进行说明。第一次学习到，原子指令在多核心CPU下，单个CPU在执行原子指令的时候，需要暂停其他CPU执行。这个给了我很大的启发，在硬件层面认知系统。

[cctga]

还是叫指令重排序听得懂些，内存顺序这个名字真的奇怪