advance-practice/shared-state

[giscus[bot]]

advance-practice/shared-state

https://course.rs/async-rust/tokio/shared-state.html

[yangs1]

锁如果在多个 .await 过程中持有 想问下，不大明白 多个 .await 过程 是怎么样的状态

[bxb100]

@SlowSlicing 是有这个现象, 但这不是死锁, 这是锁的作用域的问题, 后面应该有讲到

[wqshr12345]

@SlowSlicing 你的这个例子跟std::sync::Mutex和tokio::sync::Mutex的区别没关系，你就是用tokio::sync::Mutex一样会死锁。它俩的区别应该就是和@ttxs25830 说的一样，.await会引起task的调度，可能让一个task被调度到另一个thread执行，此时std::sync::Mutex的时候进行.await，很容易出现一个mutex在一个thread lock，在另一个thread unlock的情况，显然是错误的；另外也很容易出现，同一个thread执行的前后两个task，同时针对一个std::sync::Mutex操作出现死锁的情况。如果用tokio::sync::Mutex，它是以task为粒度进行上锁解锁，就算有thread的调度也没关系，就算同一个thread先后执行了两个task的tokio::sync::Mutex的lock也没关系

[4innocent]

简而言之就是在 async 方法中有调用其他.await，当这个 async 方法再次.await 时就会有跨多个.await了

[Justinlaau]

我的理解是。在tokio裏，他把cpu的抽象成executor，他有一個綫程池負責去運行不同的task 也就是future，如果你call了.await的方法，tokio就會把目前的綫程挂起，如果你之前持有鎖，鎖并不會釋放掉，而是隨著整個task挂起，等待.await的完成，然後繼續執行。如果有一些可以make progress的task 剛好被executor的選中工作，而又剛剛好要使用剛剛那個鎖，就會出現死鎖。而且要實現Send的原因就是 每次reschedule都不會知道任務在哪一個工作綫程中使用，所以要保證task裏所有數據綫程安全，而由於mutex實現了send trait， 所以套上mutex就可以運行了。

[Justinlaau]

Arc實現了，不是mutex

[sunzhong]

有个地方不明白，Futrue 是 Pin住的，为啥还能Send到其他线程呢

[SuanCaiYv]

我的理解是，Pin把需要Pin的数据放在堆上，每次Send Pin时只是把栈上的Pin结构体Send了，结构体指向堆的位置还是不变的，所以被Pin的数据(放在堆上)是不变的。

[Joneyao]

问一个关于use代码规范使用的问题：为什么在函数中使用use 引入外部包？是放了方便看，还是说有降低开销的原因，还是说编码习惯问题？
按一般的编程习惯将use 放在代码的开始问题，方便统一管理。

[Joneyao]

问一个关于use代码规范使用的问题：为什么在函数中使用use 引入外部包？是放了方便看，还是说有降低开销的原因，还是说编码习惯问题？
按一般的编程习惯将use 放在代码的开始问题，方便统一管理。

[Rustln]

问一个关于use代码规范使用的问题：为什么在函数中使用use 引入外部包？是放了方便看，还是说有降低开销的原因，还是说编码习惯问题？ 按一般的编程习惯将use 放在代码的开始问题，方便统一管理。

正常情况下，use 放在代码的顶部，放在代码顶部的 use 一般使用频率为 2次以上，但有时候直接放在函数中有下面几种考虑：

1. 这个外部包只在这个函数中用到，那么没必要放到顶部了
2. 工具库类的函数，方便后续拷贝迁移
3. 单元测试，Rust 很多库都喜欢在代码的最底部编写单元测试，如果把 use 放到函数外，显然不妥

[duanyuluo]

use是静态加载，会在编译阶段处理掉，不是在运行阶段。所以放在函数里面不会有性能差异。
rust支持动态加载，参考https://doc.rust-lang.org/reference/linkage.html

[ichsonx]

而真相是：Tokio 提供的异步锁只应该在跨多个 .await调用时使用
tokio::spawn(async move {
process(socket, db).await;
});

请问，这个”跨“字应该怎么理解？代码中，db也是给了多个process().await用，我理解这里也应该也是”跨“

[gnauhtt]

我的理解是，let mut db = db.lock().unwrap(); 获取到的MutexGuard不能载drop前跨越.await，你可以尝试把process方法的let mut db = db.lock().unwrap(); 移动一下位置，看一下编译的错误提示，或许会理解得更深刻一些。
async fn process(socket: TcpStream, db: Db) { // mini-redis提供的便利函数，使用返回的connection` 可以用于从 socket 中读取数据并解析为数据帧
let mut connection = Connection::new(socket);

// 使用 `read_frame` 方法从连接获取一个数据帧：一条redis命令 + 相应的数据
while let Some(frame) = connection.read_frame().await.unwrap() {
    let mut db = db.lock().unwrap(); // <=== 获取锁移动到这里，会发生编译错误
    let response = match Command::from_frame(frame).unwrap() {
        Set(cmd) => {
            // let mut db = db.lock().unwrap();
            db.insert(cmd.key().to_string(), cmd.value().clone());
            Frame::Simple("OK".to_string())
        }
        Get(cmd) => {
            // let db = db.lock().unwrap();
            if let Some(value) = db.get(cmd.key()) {
                // `Frame::Bulk` 期待数据的类型是 `Bytes`， 该类型会在后面章节讲解，
                // 此时，你只要知道 `&Vec<u8>` 可以使用 `into()` 方法转换成 `Bytes` 类型
                Frame::Bulk(value.clone().into())
            } else {
                Frame::Null
            }
        }
        cmd => panic!("unimplemented {:?}", cmd),
    };

    // 回复
    connection.write_frame(&response).await.unwrap();
}

}`

[ichsonx]

我的理解是，let mut db = db.lock().unwrap(); 获取到的MutexGuard不能载drop前跨越.await，你可以尝试把process方法的let mut db = db.lock().unwrap(); 移动一下位置，看一下编译的错误提示，或许会理解得更深刻一些。 async fn process(socket: TcpStream, db: Db) { // mini-redis提供的便利函数，使用返回的connection` 可以用于从 socket 中读取数据并解析为数据帧 let mut connection = Connection::new(socket);

// 使用 `read_frame` 方法从连接获取一个数据帧：一条redis命令 + 相应的数据
while let Some(frame) = connection.read_frame().await.unwrap() {
    let mut db = db.lock().unwrap(); // <=== 获取锁移动到这里，会发生编译错误
    let response = match Command::from_frame(frame).unwrap() {
        Set(cmd) => {
            // let mut db = db.lock().unwrap();
            db.insert(cmd.key().to_string(), cmd.value().clone());
            Frame::Simple("OK".to_string())
        }
        Get(cmd) => {
            // let db = db.lock().unwrap();
            if let Some(value) = db.get(cmd.key()) {
                // `Frame::Bulk` 期待数据的类型是 `Bytes`， 该类型会在后面章节讲解，
                // 此时，你只要知道 `&Vec<u8>` 可以使用 `into()` 方法转换成 `Bytes` 类型
                Frame::Bulk(value.clone().into())
            } else {
                Frame::Null
            }
        }
        cmd => panic!("unimplemented {:?}", cmd),
    };

    // 回复
    connection.write_frame(&response).await.unwrap();
}

}`

原来如此，感谢。我继续再 看看

[ichsonx]

是我搞混了await, 和tokio中任务的概念了，已经解决，谢谢

[haohanyuzmx]

我理解是异步的调度时机只有在.await才发生，其余时间是没有抢占了

[yAntPower]

我的理解是await就是执行一次poll方法，如果完成就是ready，没完成就pending。当pending时会调度其他的任务执行，这时候就是跨.await调用了。但是上面的代码 对锁的使用是在Set和Get分支中，相当于是提前释放了 所以使用标准库的锁没问题。如果你想把锁放到Set和Get分支之外 那么就要用tokio中的锁了。 这是我个人的理解 如果不对还请告知。

[fjchen7]

use std::sync::Mutex;

struct CanIncrement {
    mutex: Mutex<i32>,
}
impl CanIncrement {
    // 该方法不是 `async`
    fn increment(&self) {
        let mut lock = self.mutex.lock().unwrap();
        *lock += 1;
    }
}

async fn increment_and_do_stuff(can_incr: &CanIncrement) {
    can_incr.increment();
    do_something_async().await;
}

文章中的这个例子可以通过编译，是因为只需要在async fn里初始化的变量实现Send，但对传进的参数没有这个要求？

[duanyuluo]

increment不是async函数，所以，中间过程不会被切走执行权，因此这个函数里面的锁，不会因为被切走，而持续占用。

[zqlpaopao]

但是increment_and_do_stuff 是async的，会不会在执行完let mut lock = self.mutex.lock().unwrap(); 就被切走了，然后下一个async 在执行的时候，还是会死锁吧

[478619644h]

我理解是在 .await 出现前没有实现Send特征的变量要小于其生命周期

[yurnery]

例如某个任务刚获取完锁，还没使用完就因为 .await 让出了当前线程的所有权，结果下个任务又去获取了锁，造成死锁
这段话说的, 为什么会死锁呢? 我这么理解对吗?

1. 因为任务 1 被挂起之后, 他就让出了当前线程的执行权, 但是它持有一个锁;
2. 执行了另一个任务2, 他去获取这个锁的时候, 因为锁还是被任务 1 持有的, 所以他只能干等, 但是它等又占用着 CPU, 导致线程没法去执行任务 1. 形成死锁
3. 如果任务 2 是在另一个线程中执行的, 就不会发生死锁.

[mumumusuc]

任务1所await的任务排在了任务2之后就会死锁

[NicoleRobin]

反馈一个语句不顺的小问题：
一个常见的错误无条件地使用

[KonnyakuMatcha]

新人问一下这里为什么不用hashmap包裹锁，例如HashMap<String, Arc<Mutex<i32>>>这种

[2hang3hu1in]

这样什么也锁不住

[2506575085]

因为就是为了锁hashmap呀

[HongJiyu]

process() 函数，使用std::sync::Mutex，同时在db.lock().unwrap() 获取锁后，还有await操作： connection.write_frame(&response).await.unwrap(); ，这代码可能会出现死锁的问题吧

[liuhh666233]

我理解这里获取锁的生命周期是仅在match的代码块内, 在reponse赋值完成后, 获取到的锁就已经结束了

[HongJiyu]

邮件已收到！我会尽快处理的，请耐心等待。

[cctyl]

理解了“锁如果在多个 .await 过程中持有” 的意思。就是在锁的生命周期内，锁还没结束，不能再次.await。
如果.await，就有可能切换线程，那就必须实现Send，但是锁没有Send，因此报错。
举两个例子说明如下：

 示例1：

        async fn increment_and_do_stuff(mutex: &Mutex<i32>) {
            let mut lock: MutexGuard<i32> = mutex.lock().unwrap();
            *lock += 1;

            do_something_async().await;//锁还活着，但是.await就会切换线程，就要保存锁，就要Send，结论是会报错
        } // 锁在这里超出作用域

    示例2：
        async fn process(socket: TcpStream,db:Db) {

            ......;
            //这里获取了锁
            let mut mutex_guard = db.lock().unwrap();

            while let Some(frame) = ... {
                //使用锁
                .....;


                //这里调用了.await，但是锁的生命周期还没有结束
                //那就有可能发生线程的切换
                //切换就需要Send，但是锁没有实现Send
                connection.write_frame(&resp).await.unwrap();
            }


            //锁在这里结束生命周期
        }

解决办法就两个：

在.await之前就结束锁的生命周期
            使用{} 包裹，或者封装为一个单独的非异步方法
            总之不要让.await 带走锁

 使用 Tokio 提供的异步锁

[HongJiyu]

邮件已收到！我会尽快处理的，请耐心等待。