0

我正在监听 udp 数据包，一旦第一个数据包到达，我就开始监听更多数据包。

如果收到一个数据包，则服务器将开始全速运行（即使应该有延迟let duration = Duration::from_millis(1300);？），并最终被杀死

MVE代码如下：

服务器.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 {}", 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);
            }
        });
    }
}

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, 2, 3]).await.expect("Unable to talk to network");
    time::sleep(Duration::from_millis(1200)).await;
    socket.send(&[2, 2, 3]).await.expect("Unable to talk to network");
    socket.send(&[3, 2, 3]).await.expect("Unable to talk to network");
    socket.send(&[4, 2, 3]).await.expect("Unable to talk to network");
    socket.send(&[5, 2, 3]).await.expect("Unable to talk to network");
    socket.send(&[6, 2, 3]).await.expect("Unable to talk to network");
    socket.send(&[7, 2, 3]).await.expect("Unable to talk to network");
    time::sleep(Duration::from_millis(1200)).await;
    socket.send(&[8, 2, 3]).await.expect("Unable to talk to network");
    time::sleep(Duration::from_millis(3200)).await;
    socket.send(&[9, 2, 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, 0, 0, 0, 0, 0, 0, 0, 0]
Bytes len: 3 from 127.0.0.1:8000
1 id packet received:[1, 1, 0, 0, 0, 0, 0, 0, 0, 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, 1, 1, 0, 0, 0, 0, 0, 0, 0]
3 id packet received:[1, 1, 1, 1, 0, 0, 0, 0, 0, 0]
4 id packet received:[1, 1, 1, 1, 1, 0, 0, 0, 0, 0]
5 id packet received:[1, 1, 1, 1, 1, 1, 0, 0, 0, 0]
6 id packet received:[1, 1, 1, 1, 1, 1, 1, 0, 0, 0]
7 id packet received:[1, 1, 1, 1, 1, 1, 1, 1, 0, 0]
Bytes len: 3 from 127.0.0.1:8000
8 id packet received:[1, 1, 1, 1, 1, 1, 1, 1, 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, 1, 1, 1, 1, 1, 1, 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

我不知道如何在不被杀死的情况下保持服务器平稳运行，非常感谢任何帮助。

4

1 回答 1

1

感谢@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);
         /*   }
        }); */
    }
于 2021-04-01T01:17:44.263 回答