我正在開發一個專案，以在 Rust 中實作分布式鍵值存盤。我已經使用 Tokio 的異步運行時制作了服務器端代碼。我遇到了一個問題，似乎我的異步代碼被阻塞了，所以當我有多個連接到服務器時，只處理一個 TcpStream。我是實作async代碼的新手，無論是在一般情況下還是在 rust 上，但我認為如果給定的 tcp 流上沒有活動，其他流將被接受和處理。
我對異步的理解是錯誤的還是我錯誤地使用了 tokio？
這是我的切入點：
use std::error::Error;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::{Arc, Mutex};
use env_logger;
use log::{debug, info};
use structopt::StructOpt;
use tokio::net::TcpListener;
extern crate blue;
use blue::ipc::message;
use blue::store::args;
use blue::store::cluster::{Cluster, NodeRole};
use blue::store::deserialize::deserialize_store;
use blue::store::handler::handle_stream;
use blue::store::wal::WriteAheadLog;
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();
let opt = args::Opt::from_args();
let addr = SocketAddr::from_str(format!("{}:{}", opt.host, opt.port).as_str())?;
let role = NodeRole::from_str(opt.role.as_str()).unwrap();
let leader_addr = match role {
NodeRole::Leader => addr,
NodeRole::Follower => SocketAddr::from_str(opt.follow.unwrap().as_str())?,
};
let wal_name = addr.to_string().replace(".", "").replace(":", "");
let wal_full_name = format!("wal{}.log", wal_name);
let wal_path = PathBuf::from(wal_full_name);
let mut wal = match wal_path.exists() {
true => {
info!("Existing WAL found");
WriteAheadLog::open(&wal_path)?
}
false => {
info!("Creating WAL");
WriteAheadLog::new(&wal_path)?
}
};
debug!("WAL: {:?}", wal);
let store_name = addr.to_string().replace(".", "").replace(":", "");
let store_pth = format!("{}.pb", store_name);
let store_path = Path::new(&store_pth);
let mut store = match store_path.exists() {
true => deserialize_store(store_path)?,
false => message::Store::default(),
};
let listener = TcpListener::bind(addr).await?;
let cluster = Cluster::new(addr, &role, leader_addr, &mut wal, &mut store).await?;
let store_path = Arc::new(store_path);
let store = Arc::new(Mutex::new(store));
let wal = Arc::new(Mutex::new(wal));
let cluster = Arc::new(Mutex::new(cluster));
info!("Blue launched. Waiting for incoming connection");
loop {
let (stream, addr) = listener.accept().await?;
info!("Incoming request from {}", addr);
let store = Arc::clone(&store);
let store_path = Arc::clone(&store_path);
let wal = Arc::clone(&wal);
let cluster = Arc::clone(&cluster);
handle_stream(stream, store, store_path, wal, cluster, &role).await?;
}
}
下面是我的處理程式（handle_stream來自上面）。我排除了所有處理程式，match input因為我認為它們不需要證明這一點（該部分的完整代碼在這里：https : //github.com/matthewmturner/Bradfield-Distributed-Systems/blob/main/blue/ src/store/handler.rs如果它真的有幫助）。
特別是阻塞的點是線 let input = async_read_message::<message::Request>(&mut stream).await;
這是服務器等待來自客戶端或集群中另一臺服務器的通信的地方。我目前看到的行為是，在使用客戶端連接到服務器后，服務器沒有收到任何將其他節點添加到集群的請求——它只處理客戶端流。
use std::io;
use std::net::{SocketAddr, TcpStream};
use std::path::Path;
use std::str::FromStr;
use std::sync::{Arc, Mutex};
use log::{debug, error, info};
use serde_json::json;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpStream as asyncTcpStream;
use super::super::ipc::message;
use super::super::ipc::message::request::Command;
use super::super::ipc::receiver::async_read_message;
use super::super::ipc::sender::{async_send_message, send_message};
use super::cluster::{Cluster, NodeRole};
use super::serialize::persist_store;
use super::wal::WriteAheadLog;
// TODO: Why isnt async working? I.e. connecting servers after client is connected stays on client stream.
pub async fn handle_stream<'a>(
mut stream: asyncTcpStream,
store: Arc<Mutex<message::Store>>,
store_path: Arc<&Path>,
wal: Arc<Mutex<WriteAheadLog<'a>>>,
cluster: Arc<Mutex<Cluster>>,
role: &NodeRole,
) -> io::Result<()> {
loop {
info!("Handling stream: {:?}", stream);
let input = async_read_message::<message::Request>(&mut stream).await;
debug!("Input: {:?}", input);
match input {
...
}
}
}
這是代碼 async_read_message
pub async fn async_read_message<M: Message Default>(
stream: &mut asyncTcpStream,
) -> io::Result<M> {
let mut len_buf = [0u8; 4];
debug!("Reading message length");
stream.read_exact(&mut len_buf).await?;
let len = i32::from_le_bytes(len_buf);
let mut buf = vec![0u8; len as usize];
debug!("Reading message");
stream.read_exact(&mut buf).await?;
let user_input = M::decode(&mut buf.as_slice())?;
debug!("Received message: {:?}", user_input);
Ok(user_input)
}
uj5u.com熱心網友回復：
您的問題在于您在客戶端連接后如何處理訊息：
handle_stream(stream, store, store_path, wal, cluster, &role).await?;
這.await意味著您的偵聽回圈將等待handle_stream回傳，但是（進行一些假設）此函式在客戶端斷開連接之前不會回傳。你想要的是tokio::spawn一個可以獨立運行的新任務：
tokio::spawn(handle_stream(stream, store, store_path, wal, cluster, &role));
您可能必須更改某些引數型別以避免生命周期；tokio::spawn需要，'static因為任務的生命周期與它產生的范圍分離。
轉載請註明出處，本文鏈接：https://www.uj5u.com/caozuo/318774.html