60340


現在P2Pのシステムを構築中です。

何らかのキー入力を検知して、サブスレッドを終了させています。

ソケットの接続を待ち受けているサブスレッドが終了された後に`local_server.lock().unwrap().shutdown();`を実行したいのですが、Mutexのロックが解除されないのでlockができず、デッドロックの状態になってしまいます。

Mutexのロックが解除されないので、`local_server.lock().unwrap().shutdown();`が実行できないという自体に陥っています。

`local_server.lock().unwrap().shutdown();`を実行するために何ができるでしょうか？

追記:  
`start()`をreturnしたいのですが、`shutdown()`をトリガーにしてreturnしたいと思っています。  
しかし、そのトリガーである`shutdown()`が呼べないので困っています。

server1.rs

[code]

    use super::server_core;
    use std::net::{SocketAddr};
    use std::io::{self, BufRead};
    use std::{thread, time::Duration};
    use std::sync::mpsc::{self, TryRecvError};
    use std::sync::Arc;
    
    use std::io::Error;
    
    pub fn main(my_addr: SocketAddr, parent_addr: SocketAddr) -> Result<(), Error> {
    
        let mut server: server_core::ServerCore = server_core::ServerCore::new(my_addr, parent_addr);
        let local_server = server.inner.clone();
    
        let (tx, rx) = mpsc::channel();
    
        let handle = thread::spawn(move || loop {
            let local_server = server.inner.clone();
            local_server.lock().unwrap().start();
            match rx.recv() {
                Ok(_) => {
                    println!("Terminating.");
                    break;
                }
                Err(_) => {
                }
            }
        });   
    
        let mut line = String::new();
        let stdin = io::stdin();
        let _ = stdin.lock().read_line(&mut line);
    
        let _ = tx.send(());
    
        thread::sleep(Duration::from_secs(5));
    
        // can't call
        local_server.lock().unwrap().shutdown();
        // handle.join().unwrap();
        Ok(())
    }
    
    
[/code]

server_core.rs

[code]

    use super::connection_manager;
    use std::net::{SocketAddr, IpAddr, Ipv4Addr};
    use std::sync::{Arc, Mutex};
    
    // STATE_INIT = 1
    // STATE_STANDBY = 2
    // STATE_CONNECTED_TO_NETWORK = 3
    // STATE_SHUTTING_DOWN = 4
    
    #[derive(Debug)]
    pub struct ChildServerCore {
        pub server_state: u8,
        pub addr: SocketAddr,
        pub parent_addr: SocketAddr,
        pub cm: connection_manager::ConnectionManager
    }
    
    #[derive(Debug)]
    pub struct ServerCore {
        pub inner: Arc<Mutex<ChildServerCore>>
    }
    
    impl ServerCore {
        pub fn new(addr: SocketAddr, parent_addr: SocketAddr) -> ServerCore {
            let cm = connection_manager::ConnectionManager::new(addr);
            let server = ServerCore { 
                inner: Arc::new(
                    Mutex::new(
                        ChildServerCore {
                            server_state: 1,
                            addr: addr,
                            parent_addr: parent_addr,
                            cm: cm
                        }
                    )
                )
            };
            println!("Server IP address is set to ...{}", addr.ip());
            return server;
        }
    }
    
    impl ChildServerCore {
        pub fn start(&mut self) {
            self.server_state = 2;
            self.cm.start();
        }
    
        pub fn join_network(&mut self) -> Result<(), failure::Error> {
            let parent_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(000, 0, 0, 0)), 000);
            if self.parent_addr != parent_addr {
                self.server_state = 3;
                self.cm.join_network(self.parent_addr).unwrap();
            } else {
                println!("This server is running as Genesis Core Node ...");
            }
            return Ok(())
        }
    
        pub fn get_my_current_state(&self) -> u8 {
            return self.server_state;
        }
    
        pub fn shutdown(&mut self) {
            println!("Shutdown server...");
            // self.server_state = 4;
            // self.cm.connection_close().unwrap();
        }
    }
    
    
[/code]

connection_manager.rs

[code]

    use std::io::{Read, Write};
    use std::net::{TcpListener, TcpStream, SocketAddr, IpAddr, Ipv4Addr, Shutdown};
    use std::{str, thread, time::Duration};
    use serde_json::{Value};
    use std::sync::{Arc, Mutex};
    use crossbeam;
    
    use super::message;
    
    // const PROTOCOL_NAME: &'static str = "bthereum";
    // const VERSION: &'static str = "0.1.0";
    
    // const MSG_ADD: u8 = 1;
    // const MSG_REMOVE: u8 = 2;
    // const MSG_CORE_LIST: u8 = 3;
    // const MSG_REQUEST_CORE_LIST: u8 = 4;
    // const MSG_PING: u8 = 5;
    // const MSG_ADD_EDGE: u8 = 6;
    // const MSG_REMOVE_EDGE: u8 = 7;
    // const NULL = 8;
    
    // const ERR_PROTOCOL_UNMATCH: u8 = 1;
    // const ERR_VERSION_UNMATCH: u8 = 2;
    // const OK_WITH_PAYLOAD: u8 = 3;
    // const OK_WITHOUT_PAYLOAD: u8 = 4;
    
    #[derive(Debug)]
    pub struct ChildConnectionManager {
        addr: SocketAddr,
        parent_addr: SocketAddr,
        core_node_set: Vec<SocketAddr>
    }
    #[derive(Debug)]
    pub struct ConnectionManager {
        inner: Arc<Mutex<ChildConnectionManager>>
    }
    
    impl ConnectionManager {
        pub fn new(addr: SocketAddr) -> ConnectionManager {
            println!("Initializing ConnectionManager...");
            let parent_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(000, 0, 0, 0)), 000);
            let manager = ConnectionManager {
                inner: Arc::new(
                    Mutex::new(
                        ChildConnectionManager {
                            addr: addr,
                            parent_addr: parent_addr,
                            core_node_set: Vec::new()
                        }
                    )
                )
            };
            let local_self = manager.inner.clone();
            local_self.lock().unwrap()._add_peer(addr).unwrap();
            return manager;
        }
    
        pub fn start(&mut self) {
            self._wait_for_access().unwrap();
        }
    
        fn _wait_for_access(&mut self) -> Result<(), failure::Error> {
            let _local_self = self.inner.clone();
            let listener = TcpListener::bind(_local_self.lock().unwrap().addr).unwrap();
            loop {
                let local_self = self.inner.clone();
                println!("Waiting for the connection...");
                let (stream, addr) = listener.accept()?;
                println!("Connected by... {}", addr);
                let handle = thread::spawn(move|| {
                    local_self.lock().unwrap()._handle_message(stream, addr).unwrap();
                });
                handle.join().unwrap();
            }
        }
    
        pub fn connection_close(&mut self) -> Result<(), failure::Error> {
            let local_self = self.inner.clone();
            let stream = TcpStream::connect(local_self.lock().unwrap().addr)?;
            stream.shutdown(Shutdown::Both).expect("shutdown call failed");
            let temp_vec: Vec<SocketAddr> = Vec::new();
            // MSG_REMOVE
            let msg = message::build(2, local_self.lock().unwrap().addr.port(), &temp_vec).unwrap();
            let result = local_self.lock().unwrap().send_msg(local_self.lock().unwrap().parent_addr, &msg).unwrap();
            if result != () {
                local_self.lock().unwrap()._remove_peer(local_self.lock().unwrap().parent_addr).unwrap();
            }
            return Ok(());
        }
    
        pub fn join_network(&mut self, address: SocketAddr) -> Result<(), failure::Error> {
            let local_self = self.inner.clone();
            local_self.lock().unwrap().parent_addr = address;
            self._connect_to_p2pnw(address).unwrap();
            return Ok(());
        }
    
        fn _connect_to_p2pnw(&mut self, address: SocketAddr) -> Result<(), failure::Error> {
            let local_self = self.inner.clone();
            let mut stream = TcpStream::connect(address)?;
            let temp_vec: Vec<SocketAddr> = Vec::new();
            // MSG_ADD
            let msg = message::build(1, local_self.lock().unwrap().addr.port(), &temp_vec).unwrap();
            let string: String = msg.to_string();
            // ここでエラー
            stream.write_all(string.as_bytes())?;
            stream.shutdown(Shutdown::Both).expect("shutdown call failed");
            return Ok(());
        }
    }
    
    impl ChildConnectionManager {
    
        fn _handle_message(&mut self, mut stream: TcpStream, addr: SocketAddr) -> Result<(), failure::Error> {
            let mut buffer = [0u8; 1024];
            loop {
                let nbytes = stream.read(&mut buffer)?;
                if nbytes == 0 {
                    println!("Connection closed.");
                    return Ok(());
                }
                // print!("{}", str::from_utf8(&buffer[..nbytes])?);
                let data: &str = str::from_utf8(&buffer[..nbytes])?;
                let message: Value = serde_json::from_str(data).unwrap();
                let (result, reason, cmd, peer_port, payload) = message::parse(&message);
                // ホストとメッセージ受付用のポートをがっちゃんこ
                let connect_addr = SocketAddr::new(addr.ip(), peer_port);
                println!("result: {}, reason: {}, cmd: {}, peer_port: {}, payload: {:?}", result, reason, cmd, peer_port, payload);
                let status = (result, reason);
                // ERR_PROTOCOL_UNMATCH
                if status == ("error".to_string(), 1) {
                    println!("Error: Protocol name is not matched");
                    return Ok(());
                }
                // ERR_VERSION_UNMATCH
                else if status == ("error".to_string(), 2) {
                    println!("Error: Protocol version is not matched");
                    return Ok(());
                }
                // OK_WITHOUT_PAYLOAD
                else if status == ("ok".to_string(), 4) {
                    // MSG_ADD
                    if cmd == 1 {
                        println!("ADD node request was received!!");
                        self._add_peer(connect_addr).unwrap();
                        if addr == self.addr {
                            return Ok(());
                        }
                        else {
                            // MSG_CORE_LIST
                            let msg = message::build(3, addr.port(), &self.core_node_set).unwrap();
                            self.send_msg_to_all_peers(&msg);
                            return Ok(())
                        }
                    }
                    // MSG_REMOVE
                    else if cmd == 2 {
                        println!("REMOVE node request was received!! from: {}", addr);
                        self._remove_peer(connect_addr).unwrap();
                        // MSG_CORE_LIST
                        let msg = message::build(3, addr.port(), &self.core_node_set).unwrap();
                        self.send_msg_to_all_peers(&msg);
                        return Ok(())
                    }
                    // MSG_PING
                    else if cmd == 5 {
                        return Ok(())
                    }
                    // MSG_REQUEST_CORE_LIST
                    else if cmd == 4 {
                        println!("List for Core nodes was requested!!");
                        // MSG_CORE_LIST
                        let msg = message::build(3, addr.port(), &self.core_node_set).unwrap();
                        let socket_address = SocketAddr::new(addr.ip(), peer_port);
                        let result = self.send_msg(socket_address, &msg).unwrap();
                        if result != () {
                            self._remove_peer(socket_address).unwrap();
                        }
                        return Ok(())
                    }
                    else {
                        println!("received unknown command: {}", cmd);
                        return Ok(());
                    }
                }
                // OK_WITH_PAYLOAD
                else if status == ("ok".to_string(), 3) {
                    // MSG_CORE_LIST
                    if cmd == 3 {
                        // 受信したリストを上書きしてるのでセキュアではない
                        println!("Refresh the core node list...");
                        let new_core_set = payload;
                        println!("latest core node list:{:?}", new_core_set);
                        self.core_node_set = new_core_set;
                        return Ok(())
                    }
                    else {
                        println!("received unknown command: {}", cmd);
                        return Ok(())
                    }
                }
                else {
                    println!("Unexpected status: {}, {}", status.0, status.1);
                }
            }
        }
    
        fn _add_peer(&mut self, peer: SocketAddr) -> Result<(), failure::Error> {
            println!("Adding peer: ({})", peer);
            self.core_node_set.push(peer);
            println!("Current Core List: {:?}", self.core_node_set);
            return Ok(())
        }
    
        fn _remove_peer(&mut self, peer: SocketAddr) -> Result<(), failure::Error> {
            if self.core_node_set.contains(&peer) {
                for i in 0..self.core_node_set.len() {
                    if self.core_node_set[i] == peer {
                        println!("Removing peer: ({})", peer);
                        self.core_node_set.remove(i);
                        println!("Current Core List: {:?}", self.core_node_set);
                    }
                }
            }
            return Ok(())
        }
    
        fn _check_peers_connection(&mut self) {
            println!("check_peers_connection was called");
            let mut changed: bool = false;
            let mut dead_core_node_set: Vec<SocketAddr> = Vec::new();
            for i in 0..self.core_node_set.len() {
                let result = self._is_alive(self.core_node_set[i]).unwrap();
                if result != true {
                    dead_core_node_set.push(self.core_node_set[i]);
                    self.core_node_set.remove(i);
                }
            }
            if dead_core_node_set.len() != 0 {
                changed = true;
                println!("Removed {:#?}", dead_core_node_set);
            }
            println!("current core node list: {:#?}", self.core_node_set);
    
            if changed {
                // MSG_CORE_LIST 
                let msg = message::build(3, self.addr.port(), &self.core_node_set).unwrap();
                self.send_msg_to_all_peers(&msg);
            }
            // 定期的に_check_peers_connectionを呼び出し
            thread::sleep(Duration::from_secs(1800));
            self._check_peers_connection();
        }
    
        fn _is_alive(&mut self, target: SocketAddr) -> Result<bool, failure::Error> {
            let mut stream = TcpStream::connect(target)?;
            // MSG_PING
            let temp_port: u16 = 0;
            let temp_vec: Vec<SocketAddr> = Vec::new();
            let msg = message::build(5, temp_port, &temp_vec).unwrap();
            let string: &str = msg.as_str().unwrap();
            stream.write_all(string.as_bytes())?;
            stream.shutdown(Shutdown::Both).expect("shutdown call failed");
            return Ok(true);
        }
    
        fn send_msg_to_all_peers(&mut self, msg: &Value) {
            println!("send_msg_to_all_peers was called!");
            for i in 0..self.core_node_set.len() {
                if self.core_node_set[i] != self.addr {
                    println!("message will be sent to ...{}", self.core_node_set[i]);
                    let result = self.send_msg(self.core_node_set[i], &msg).unwrap();
                    if result != () {
                        self._remove_peer(self.core_node_set[i]).unwrap();
                    }
                }
            }
        }
    
        fn send_msg(&mut self, socket_address: SocketAddr, msg: &Value) -> Result<(), failure::Error> {
            let mut stream = TcpStream::connect(socket_address)?;
            let string: String = msg.to_string();
            stream.write_all(string.as_bytes())?;
            stream.shutdown(Shutdown::Both).expect("shutdown call failed");
            return Ok(())
        }
    }
    
    
[/code]


CC BY-SA 4.0
2019-11-07T16:59:13.310
60337
2019-11-08T02:42:21.003
2019-11-08T02:42:21.003
36479
36479
question
rust
Mutexのロックが解除されない


`super::server_core::ServerCore`の実装がどうなっているのかわからないので、具体的にどう直したらいいかはアドバイスできないのですが、

`local_server.lock().unwrap().start()`で設定したMutexのロックは、`start()`メソッドからリターンすれば解除されます。逆にいうと、`start()`メソッドからずっとリターンしないために、この行から先に実行が進んでないのだろうと考えられます。

まずは`local_server.lock().unwrap().start()`の行と`match rx.recv()
{`の行の間に`println!`かなにかを追加して、そのメッセージが表示されないこと（つまり、本当に`start()`メソッドからリターンしていないのかどうか）を確認してみてください。

もしそのメッセージが表示されないなら、`start()`メソッドからすぐにリターンするように、`super::server_core::ServerCore`の実装を変更してください。

**追記**

`start()`で起動し、`shutdown()`で終了できるサーバーの実装例を書いてみました。参考になれば幸いです。

[code]

    use std::io::BufRead;
    
    fn main() {
        // サーバーを作成し、スタートする
        let mut server = server::ServerCore::new();
        server.start();
    
        // ユーザーからの入力を待つ
        let mut line = String::new();
        let stdin = std::io::stdin();
        let _ = stdin.lock().read_line(&mut line);
    
        // サーバーを終了する
        server.shutdown();
    }
    
    mod server {
    
        use std::sync::{
            atomic::{AtomicBool, Ordering},
            Arc,
        };
        use std::thread;
    
        pub struct ServerCore {
            // trueの間はサーバーはwhileループを回し続ける。falseになったら終了する
            is_running: Arc<AtomicBool>,
            // サーバーを実行するスレッドへのハンドル
            thread: Option<thread::JoinHandle<()>>,
        }
    
        impl ServerCore {
            pub fn new() -> Self {
                Self {
                    is_running: Arc::new(AtomicBool::new(false)),
                    thread: None,
                }
            }
    
            // サーバーを起動する。このメソッドからはすぐにリターンする
            pub fn start(&mut self) {
                // すでにサーバーがスタートしているようなら、何もしないでリターンする
                if self.is_running.load(Ordering::Relaxed) {
                    println!("{:?}: The server is already started.", thread::current().id());
                    return;
                }
    
                println!("{:?}: Starting the server.", thread::current().id());
                let is_running = Arc::clone(&self.is_running);
                // is_runningをtrueに設定する
                is_running.store(true, Ordering::Relaxed);
                // 子スレッドを立ち上げ、その子スレッドでrun()を実行する
                self.thread = Some(thread::spawn(move || Self::run(is_running)));
                println!("{:?}: Finished starting the server.", thread::current().id())
            }
    
            // is_runningがfalseになるまでwhileループを実行する
            fn run(is_running: Arc<AtomicBool>) {
                println!("{:?}: Server started.", thread::current().id());
                while is_running.load(Ordering::Relaxed) {
                    println!("{:?}: Server is running.", thread::current().id());
                    thread::sleep(std::time::Duration::from_secs(1));
                }
                println!("{:?}: Server stopped.", thread::current().id());
            }
    
            // サーバーを停止する。このメソッドは子スレッドが終了するまでブロックされる
            pub fn shutdown(&mut self) {
                // すでにサーバーが終了しているようなら、何もしないでリターンする
                if !self.is_running.load(Ordering::Relaxed) {
                    println!("{:?}: The server is already shut down.", thread::current().id());
                    return;
                }
                println!("{:?}: Shutting down the server.", thread::current().id());
                // is_runningをfalseに設定する
                self.is_running.store(false, Ordering::Relaxed);
                // 子スレッドが終了するのを待つ
                self.thread.take().map(|t| t.join().unwrap());
                println!("{:?}: Finished shuttig down the server.", thread::current().id());
            }
        }
    }
    
[/code]


CC BY-SA 4.0
2019-11-07T22:27:33.060
60340
2019-11-08T00:40:32.473
2019-11-08T00:40:32.473
14101
14101
60337
answer