advance-practice/spawning

[giscus[bot]]

advance-practice/spawning

https://course.rs/advance-practice/spawning.html

[EluvK]

本节内容最开始，把之前的客户端代码放进[example]里，少了说明：
要添加以下example说明到Cargo.toml文件里：

[[example]]
name = "hello-redis"
path = "examples/hello-redis.rs"

毕竟这部分内容是在第7章Cargo里才介绍到，这里最好补充一下简要说明和跳转链接吧

[EluvK]

我增加了一点说明，顺手改掉了一点笔误，请Review. #889

[wentaotoyou]

Send约束rc的这个例子，感觉编译器把人掰弯过来后，再给你掰弯回去。毕竟前面借用检查都实现了Non-Lexical Lifetimes(NLL)--专门用于找到某个引用在作用域(})结束前就不再被使用的代码位置，这个地方是不是也可以更进一步呢？

[NicholasFirst]

    Set(cmd) => {
        // 值被存储为 `Vec<u8>` 的形式
        db.insert(cmd.key().to_string(), cmd.value().to_vec());
        Frame::Simple("OK".to_string())
     }

一定得是大写的OK

[LovesAsuna]

我也想知道这里的NLL去哪了

[runzhi214]

深有同感

[lgln0]

编译器不够强大，如果是这样相当于实现内存自动回收了，人家python、java动态编译才实现的特性

[ichsonx]

本想尝试下异步的表现，可是怎么弄都是同步的逻辑。
我在异步函数sing()中使用task休眠3秒，理论上应该会让出线程继续执行dance()，但结果并没有。
结果打印依然是顺序打印，在等待sing()休眠完后才继续往下执行。
请问这段代码怎么达到预期的异步效果呢？

use tokio;
use tokio::{time};

#[tokio::main]
async fn main() {
sing().await; //里面有休眠3秒的逻辑，按道理休眠后，会异步调用dance()乃至say_to_world()都有可能
dance().await;

let s = say_to_world().await;
println!("{}", s)
}

async fn get_Song() -> String{
println!("finish the song");
String::from("song name abcd...")
}

async fn sing(){
let s = get_Song().await;
time::sleep(time::Duration::from_secs(3)).await; // 这里进行了休眠3秒
println!("it`s singing song: {}", s)
}

async fn dance(){
println!("now is dancing...")
}

async fn say_to_world() -> String {
String::from("world")
}

[idlercloud]

上面部分改成

use tokio::{join, time};

#[tokio::main]
async fn main() {
    // 同时运行三个 future
    // 花费的时间约等于最久的那一个
    join!(sing(), dance(), async {
        let s = say_to_world().await;
        println!("{}", s)
    });
}

实际上 async/await 的用法你可能有点误解了。

你首先可以把整个 main 函数看作一个最大的 future。

其下有三个小的 future，分别是 sing、dance 和 print。对小的 future 使用 .await 意味着「父 future 执行时，等待子 future 完成」。

也就是说，这个时候实际上是 main 进行 .await，等待着一个子 future 完成，这时控制权是可以交给和 main 以外的 future 的，不过这个例子里面只有 main 这么一个最大的 future，所以表现出来就是一直到 sing 完成了，dance 才开始。

上面的言辞可能不是那么准确，但大概的意思是这样的。join! 宏就是相当于帮你同时运行几个 future，在它们之间进行调度。

[LovesAsuna]

为什么子future之间不能直接转移控制权，我写kotlin的协程的时候用launch就是这种自动切换任务的形式，所以学rust的时候给我整懵了

[ichsonx]

感谢你的回复，我已经了解了，现在才看到你的回复。
很多文章或书本没有很深入去解释 .await 、 join!宏等等的机制，所以可能会有不少人在使用上会出现各种理解之外的情况。

像@LovesAsuna说的其实也可以理解，既然要异步并行了，直接点全都调度就好，当然很方便，但rust这样做，可能是为了保持异步使用的多样性来适用不同的场景需求。

我的问题其实还可以使用tokio::task 或者 tokio::spawn的函数进行封装调用，这样他们之间就会在任务等待时释放cpu资源，在多任务间并发调度了。不过相对.await，这样做就不会【等待完成】，执行是会执行但不知道什么时候执行，main也不会等你执行完才结束，可能你们没执行完我就先下班，你们也被迫下班了

[Justinlaau]

你的代碼是sequential地進行，并沒有利用到異步并發的優勢

[iscas-zac]

这个 process 函数里的 match 臂，顺序是有影响的。因为前面的 db 需要推断泛型参数，而只有 Set 分支能推断出完整的 HashMap<String, Vec<u8>> 类型。如果先遇到 Get 分支就推断不全。语义上 match 各分支是没有先后的，类型推断也应当尽力而为，感觉这是语言的一个缺陷。

用的是 rust 1.67.1 版本。

EDIT：搜了一下发现从 2015 年传承至今。rust-lang/rust#25165

[runzhi214]

用std::sync::Arc和tokio::sync::Mutex实现一个连接共享的HashMap还是比较容易的吧？

use std::collections::HashMap;
use tokio::net::{TcpListener, TcpStream};
use mini_redis::{Connection, Frame};
use mini_redis::Command::{Get,Set,self};
use tokio::sync::Mutex;
use std::sync::Arc;


#[tokio::main]
async fn main(){
    let listener = TcpListener::bind("127.0.0.1:6379").await.expect("连接redis失败");

    let mut db:Arc<Mutex<HashMap<String,Vec<u8>>>> = Arc::new(Mutex::new(HashMap::new()));

    loop {
        let (socket,_) = listener.accept().await.expect("监听失败");
        let move_db = Arc::clone(&db);
        tokio::spawn(async move {
            process(socket, move_db).await;
        });
    }
}

async fn process(socket: TcpStream, db: Arc<Mutex<HashMap<String, Vec<u8>>>>) {
    let mut connection = Connection::new(socket);

    while let Some(frame) = connection.read_frame().await.unwrap() {
        let response = match Command::from_frame(frame).unwrap() {
            Set(cmd) => {
                db.lock().await.insert(cmd.key().to_string(), cmd.value().to_vec());
                Frame::Simple("OK".to_string())
            }
            Get(cmd) => {
                match db.lock().await.get(cmd.key()) {
                    None => {
                        Frame::Null
                    }
                    Some(v) => {
                        Frame::Bulk(v.clone().into())
                    }
                }
            }
            cmd => panic!("unimplemented commands {:?}",cmd),
        };
        connection.write_frame(&response).await.expect("Fail to write response");
    }
}

然而这么做的缺点就是读写都共享同一把锁，要是可以做成无锁设计或者是写锁就好了.....

[runzhi214]

tokio::sync::Mutex改成tokio::sync::RwLock的话性能应该会更强

[maojinhua]

example 在运行的时候连接断开了- - 服务端要怎么样保持连接不断开呢？
use mini_redis::{client, Result};

#[tokio::main]
async fn main() ->Result<()>{
let mut client = client::connect("127.0.0.1:6379").await?;

client.set("hello","world".into()).await.expect("set failed");


// let mut client = client::connect("127.0.0.1:6999").await?;
let result = client.get("hello").await.expect("get failed");

println!("从服务器获取到结果={:?}",result);
Ok(())

}

[codecoron]

mini-redis 好像不支持持久化

[zhfs]

求大佬解答，为什么服务端set返回值必须是大写的OK？感谢

[Jimmy2Angel]

在 client.rs 里搜一下 "OK" 就可以看到了

/// The core `SET` logic, used by both `set` and `set_expires.
async fn set_cmd(&mut self, cmd: Set) -> crate::Result<()> {
    // Convert the `Set` command into a frame
    let frame = cmd.into_frame();

    debug!(request = ?frame);

    // Write the frame to the socket. This writes the full frame to the
    // socket, waiting if necessary.
    self.connection.write_frame(&frame).await?;

    // Wait for the response from the server. On success, the server
    // responds simply with `OK`. Any other response indicates an error.
    match self.read_response().await? {
        Frame::Simple(response) if response == "OK" => Ok(()),
        frame => Err(frame.to_error()),
    }
}

[heqyu]

这里hashmap的编译器类型推导很有意思。在match里面，如果对换了Set和Get会导致编译不通过：只有db.insert才可以推导出类型，而先Get的话，db.get不知道是个什么。 然而写代码容易会先写出先get后set的习惯