fix:fix ub when from_std

docs/rust_test-zh.md

@@ -2,9 +2,9 @@
 
 ## 1.正确的依赖
 
- 对于Rust代码，首先对于严重依赖Godot场景树和对象函数的方法，由于使用字符串调用，你需要对于Debug模式下，在`ready`函数中尝试`get`这些函数和对象，以此实现部分的自动测试
+对于Rust代码，首先对于严重依赖Godot场景树和对象函数的方法，由于使用字符串调用，你需要对于Debug模式下，在`ready`函数中尝试`get`这些函数和对象，以此实现部分的自动测试
 
-然而，由于该部分较为重复，我们提供了```#[derive::gen_debug]```和```debug_check!()```宏来简化。
+然而，由于该部分较为重复，我们提供了`#[derive::gen_debug]`和`debug_check!()`宏来简化。
 
 例子:
 
@@ -21,9 +21,9 @@ impl INode2D for xxx {
 impl xxx {
     #[debug]
     fn get_staticbody(&self) -> Gd<StaticBody2D> {
-        self.base().get_node_as::<StaticBody2D>("Collision")
+        self.base().get_node_as("Collision")
     }
 }
 ```
 
-这样，被```#[debug]```标记的函数就会在Debug模式下于```debug_check!()```宏处运行了
+这样，被`#[debug]`标记的函数就会在Debug模式下于`debug_check!()`宏处运行了

docs/rust_test.md

@@ -2,9 +2,9 @@
 
 ## 1. Correctness Assurance
 
-For Rust code, first of all, for methods that heavily rely on Godot's scene tree and object functions, since string calling is used, you need to attempt to ```get``` these functions and objects in the ```ready``` function during Debug mode, thereby implementing some automatic testing.
+For Rust code, first of all, for methods that heavily rely on Godot's scene tree and object functions, since string calling is used, you need to attempt to `get` these functions and objects in the `ready` function during Debug mode, thereby implementing some automatic testing.
 
-However,because this part is repetitive,we provide ```#[derive::gen_debug]``` and ```debug_check!()``` macros to simplify this process.
+However,because this part is repetitive,we provide `#[derive::gen_debug]` and `debug_check!()` macros to simplify this process.
 
 For example:
 
@@ -21,9 +21,9 @@ impl INode2D for xxx {
 impl xxx {
     #[debug]
     fn get_staticbody(&self) -> Gd<StaticBody2D> {
-        self.base().get_node_as::<StaticBody2D>("Collision")
+        self.base().get_node_as("Collision")
     }
 }
 ```
 
-Like this,the function marked by ```#[debug]``` will be ran in debug mode at the place of ```debug_check!()``` macro.
+Like this,the function marked by `#[debug]` will be ran in debug mode at the place of `debug_check!()` macro.

gdrust/Cargo.toml

@@ -5,7 +5,9 @@ version = "0.1.0"
 edition = "2024"
 
 [dependencies]
-godot = { git = "https://github.com/godot-rust/gdext", branch = "master" }
+godot = { git = "https://github.com/godot-rust/gdext", branch = "master", features = [
+  "experimental-threads",
+] }
 rand = "0.8"
 derive = { path = "./derive" }
 proto = { path = "proto" }

gdrust/server/src/lib.rs

@@ -21,30 +21,40 @@ use tokio::{
 struct Connection {
     stream: TcpStream,
     buffer: BytesMut,
+    shutdown: broadcast::Receiver<()>,
 }
 
 impl Connection {
-    pub fn new(stream: TcpStream) -> Self {
+    pub fn new(shutdown: broadcast::Receiver<()>, stream: TcpStream) -> Self {
         Self {
             stream,
             buffer: BytesMut::with_capacity(1024),
+            shutdown,
         }
     }
 
     pub async fn read_join(&mut self) -> anyhow::Result<Option<Join>> {
         loop {
-            if let Some(request) = self.parse_join()? {
-                return Ok(Some(request));
+            if self.shutdown.try_recv().is_ok() {
+                log::info!("Exit");
+                return Err(anyhow!("Exit"));
             }
-            if 0 == self.stream.read_buf(&mut self.buffer).await? {
+            let sz = self.stream.read_buf(&mut self.buffer).await?;
+            log::info!("Received:{}", sz);
+            if 0 == sz {
                 if self.buffer.is_empty() {
+                    log::info!("Join exiting(empty)...");
                     return Ok(None);
                 } else {
                     let msg = "Connection reset by peer";
                     log::info!("{}", msg);
                     return Err(anyhow!(msg));
                 }
             }
+            if let Some(request) = self.parse_join()? {
+                log::info!("Join exiting...");
+                return Ok(Some(request));
+            }
         }
     }
 
@@ -74,26 +84,37 @@ struct ArgsParser {
 
 async fn process_request(mut connect: Connection) -> anyhow::Result<()> {
     // 首先获取连接请求
+    log::info!("start joining");
     let join_data = connect.read_join().await?;
-    loop {}
+    log::info!("joined");
+    let request = async { loop {} };
+    select! {
+        _ = request => {},
+        _ = connect.shutdown.recv() => {
+            log::info!("Player exited")
+        }
+    }
     Ok(())
 }
 
-async fn accept_sockets(tcplistener: TcpListener, mut shutdown_receiver: broadcast::Receiver<()>) {
+async fn accept_sockets(
+    tcplistener: TcpListener,
+    shutdown_sender: broadcast::Sender<()>,
+    mut shutdown_receiver: broadcast::Receiver<()>,
+) {
     let async_loop = async move {
         loop {
             let ret = tcplistener.accept().await;
             match ret {
                 Ok((socket, _)) => {
-                    if let Err(e) = tokio::spawn(async {
-                        if let Err(e) = process_request(Connection::new(socket)).await {
+                    let shutdown = shutdown_sender.subscribe();
+                    log::info!("Connected to a socket");
+                    tokio::spawn(async move {
+                        if let Err(e) = process_request(Connection::new(shutdown, socket)).await {
                             log::error!("When processing a request:{}", e)
                         }
-                    })
-                    .await
-                    {
-                        log::error!("Async error:{}", e);
-                    }
+                        log::info!("A socket exited successful");
+                    });
                 }
                 Err(e) => {
                     log::error!("Accepting a new player failed:{}", e)
@@ -122,7 +143,11 @@ pub async fn lib_main() -> Result<(), Box<dyn Error>> {
         }
     };
     let (shutdown_sender, mut shutdown_receiver) = broadcast::channel(32);
-    tokio::spawn(accept_sockets(tcplistener, shutdown_sender.subscribe()));
+    tokio::spawn(accept_sockets(
+        tcplistener,
+        shutdown_sender.clone(),
+        shutdown_sender.subscribe(),
+    ));
     tokio::spawn(async move {
         match tokio::signal::ctrl_c().await {
             Ok(()) => {

gdrust/src/multi.rs

@@ -38,6 +38,7 @@ async fn write_loop(
     mut receiver: mpsc::Receiver<Bytes>,
     mut write_socket: OwnedWriteHalf,
 ) -> anyhow::Result<()> {
+    godot_print!("Starting writing loop");
     while let Some(data) = receiver.recv().await {
         write_socket.write_all(&data).await?;
     }
@@ -49,6 +50,7 @@ async fn read_loop(
     mut read_socket: OwnedReadHalf,
 ) -> anyhow::Result<()> {
     let mut buf = BytesMut::new();
+    godot_print!("Starting reading loop");
     loop {
         let n = read_socket.read(&mut buf).await?;
         match parse_request(&buf) {
@@ -84,14 +86,18 @@ impl MultiManagerImpl {
             godot_warn!("Socket has value,but reset")
         }
         let socket = std::net::TcpStream::connect(&ip)?;
-        let socket = TcpStream::from_std(socket)?;
-        let (read_socket, write_socket) = socket.into_split();
+        socket.set_nonblocking(true)?;
         let (sender, receiver) = mpsc::channel(32);
         let (request_sender, request_receiver) = std::sync::mpsc::channel();
         self.receiver = Some(request_receiver);
         self.socket = Some(sender);
-        get_tokio_runtime().spawn(write_loop(receiver, write_socket));
-        get_tokio_runtime().spawn(read_loop(request_sender, read_socket));
+        get_tokio_runtime().spawn(async move {
+            let socket = TcpStream::from_std(socket)?;
+            let (read_socket, write_socket) = socket.into_split();
+            get_tokio_runtime().spawn(write_loop(receiver, write_socket));
+            get_tokio_runtime().spawn(read_loop(request_sender, read_socket));
+            anyhow::Ok(())
+        });
         Ok(())
     }