如何解决Tokio 服务器在过热后被杀死
我正在监听 udp 数据包,一旦第一个数据包到达,我就开始监听更多数据包。
如果收到一个数据包,那么服务器将开始全速运行(即使应该有延迟 let duration = Duration::from_millis(1300);
?),并最终被杀死
MVE 代码如下:
server.rs
use std::net::SocketAddr;
use std::time::Duration;
use tokio::net::UdpSocket;
use tokio::{sync::mpsc,task,time}; // 1.4.0
use std::env;
use std::sync::Arc;
const UDP_HEADER: usize = 8;
const IP_HEADER: usize = 20;
const AG_HEADER: usize = 4;
const MAX_DATA_LENGTH: usize = (64 * 1024 - 1) - UDP_HEADER - IP_HEADER;
const MAX_CHUNK_SIZE: usize = MAX_DATA_LENGTH - AG_HEADER;
const MAX_DATAGRAM_SIZE: usize = 0x10000;
const ADDRESS: &str = "127.0.0.1:8080";
const ADDRESS_CLIENT: &str = "127.0.0.1:8000";
#[tokio::main]
async fn main() {
server().await;
}
async fn server() {
eprintln!("Starting the server");
let addr = env::args().nth(1).unwrap_or_else(|| ADDRESS.to_string());
let socket = UdpSocket::bind(&addr).await.unwrap();
let arc = Arc::new(socket);
let mut buf = [0u8; MAX_DATA_LENGTH];
let (debounce_tx,mut debounce_rx) = mpsc::channel::<Vec<u8>>(MAX_DATAGRAM_SIZE);
let _debouncer = task::spawn(async move {
let mut _packet_ids: Vec<i32> = Vec::new();
_packet_ids = vec![0; 10];
let duration = Duration::from_millis(1300);
loop {
match time::timeout(duration,debounce_rx.recv()).await {
Ok(Some(bytes)) => {
let id: u8 = bytes.clone()[0];
_packet_ids[id as usize] = 1;
eprintln!("{} id packet received:{:?}",id,_packet_ids);
if _packet_ids.iter().all(|x| x == &1i32) {
println!("All packets have been received,stop program ");
// break;
}
}
Ok(None) => {
eprintln!("Done: {:?}",_packet_ids);
break;
}
Err(_) => {
eprintln!("No activity for 1.3sd");
}
}
}
});
// Listen for first packet
let result = arc.clone().recv_from(&mut buf).await;
match result {
Ok((len,addr)) => {
eprintln!("Bytes len: {} from {}",len,addr);
debounce_tx
.send(buf.to_vec())
.await
.expect("Unable to talk to debounce");
}
Err(_) => {
eprintln!("Couldnt get datagram");
}
}
// listen for other packets
loop {
let thread_socket = arc.clone();
let _server = task::spawn({
let debounce_tx = debounce_tx.clone();
async move {
while let result = thread_socket.recv_from(&mut buf).await {
match result {
Ok((len,addr)) => {
eprintln!("Bytes len: {} from {}",addr);
debounce_tx
.send(buf.to_vec())
.await
.expect("Unable to talk to debounce");
}
Err(_) => {
eprintln!("Couldnt get datagram");
}
}
}
// Prevent deadlocks
drop(debounce_tx);
}
});
}
}
client.rs(用于测试目的)
async fn client() {
eprintln!("Starting the client");
let remote_addr: SocketAddr = env::args()
.nth(2)
.unwrap_or_else(|| ADDRESS.into()) // cargo run --example udp-client -- 127.0.0.1:8080
.parse()
.unwrap();
// We use port 0 to let the operating system allocate an available port for us.
let local_addr: SocketAddr = if remote_addr.is_ipv4() {
ADDRESS_CLIENT // "0.0.0.0:0" //
} else {
"[::]:0"
}
.parse()
.unwrap();
let socket = UdpSocket::bind(ADDRESS_CLIENT).await.unwrap();
socket.connect(&remote_addr).await.unwrap();
socket.send(&[0,2,3]).await.expect("Unable to talk to network");
socket.send(&[1,3]).await.expect("Unable to talk to network");
time::sleep(Duration::from_millis(1200)).await;
socket.send(&[2,3]).await.expect("Unable to talk to network");
socket.send(&[3,3]).await.expect("Unable to talk to network");
socket.send(&[4,3]).await.expect("Unable to talk to network");
socket.send(&[5,3]).await.expect("Unable to talk to network");
socket.send(&[6,3]).await.expect("Unable to talk to network");
socket.send(&[7,3]).await.expect("Unable to talk to network");
time::sleep(Duration::from_millis(1200)).await;
socket.send(&[8,3]).await.expect("Unable to talk to network");
time::sleep(Duration::from_millis(3200)).await;
socket.send(&[9,3]).await.expect("Unable to talk to network"); // stop when n1 = 0
eprintln!("Client done");
}
输出(从客户端发送 10 个数据包,其中第一个字节的范围从 0 到 9,注意:我现在只是想让它工作,我知道缓冲区溢出):: >
Starting the server
Bytes len: 3 from 127.0.0.1:8000
0 id packet received:[1,0]
Bytes len: 3 from 127.0.0.1:8000
1 id packet received:[1,1,0]
Bytes len: 3 from 127.0.0.1:8000
Bytes len: 3 from 127.0.0.1:8000
Bytes len: 3 from 127.0.0.1:8000
Bytes len: 3 from 127.0.0.1:8000
Bytes len: 3 from 127.0.0.1:8000
Bytes len: 3 from 127.0.0.1:8000
2 id packet received:[1,0]
3 id packet received:[1,0]
4 id packet received:[1,0]
5 id packet received:[1,0]
6 id packet received:[1,0]
7 id packet received:[1,0]
Bytes len: 3 from 127.0.0.1:8000
8 id packet received:[1,0]
No activity for 1.3sd
No activity for 1.3sd
Bytes len: 3 from 127.0.0.1:8000
9 id packet received:[1,1]
All packets have been received,stop program
No activity for 1.3sd
No activity for 1.3sd
No activity for 1.3sd
Killed
我不知道如何保持服务器平稳运行而不被杀死,非常感谢任何帮助。
解决方法
感谢@Frxstrem 和@transistor 的评论解决了这个问题,请参阅代码对更改的评论:
loop {
let thread_socket = arc.clone();
let debounce_tx = debounce_tx.clone(); // moved up
/* let _server = task::spawn({
async move { */
if let result = thread_socket.recv_from(&mut buf).await { // previously while
match result {
Ok((len,addr)) => {
eprintln!("Bytes len: {} from {}",len,addr);
debounce_tx
.send(buf.to_vec())
.await
.expect("Unable to talk to debounce");
}
Err(_) => {
eprintln!("Couldnt get datagram");
}
}
}
// Prevent deadlocks
drop(debounce_tx);
/* }
}); */
}
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。