Add joining to increase reliability when closing.

Author: john-michaelburke

console_backend/src/constants.rs

@@ -8,6 +8,10 @@ pub(crate) const APPLICATION_ORGANIZATION: &str = "swift-nav";
 pub(crate) const APPLICATION_NAME: &str = "swift_navigation_console";
 // CLI constants.
 
+// Server constants.
+#[allow(dead_code)]
+pub(crate) const FETCH_MESSAGE_TIMEOUT_MS: u64 = 1;
+
 // Process Message constants.
 pub(crate) const PAUSE_LOOP_SLEEP_DURATION_MS: u64 = 100;
 

console_backend/src/process_messages.rs

@@ -202,6 +202,7 @@ where
     }
     if conn.close_when_done() {
         shared_state.set_running(false, client_send.clone());
+        shared_state.stop_server_running();
         close_frontend(&mut client_send);
     }
     Ok(())

console_backend/src/server.rs

@@ -8,32 +8,36 @@ use async_logger_log::Logger;
 use log::error;
 use std::{
     io::{BufReader, Cursor},
+    ops::Drop,
     path::PathBuf,
     str::FromStr,
     sync::mpsc,
     thread,
+    time::Duration,
 };
 
 use crate::cli_options::*;
+use crate::common_constants::ApplicationStates;
 use crate::connection::ConnectionState;
 use crate::console_backend_capnp as m;
-use crate::constants::LOG_WRITER_BUFFER_MESSAGE_COUNT;
+use crate::constants::{FETCH_MESSAGE_TIMEOUT_MS, LOG_WRITER_BUFFER_MESSAGE_COUNT};
 use crate::errors::*;
 use crate::log_panel::{splitable_log_formatter, LogLevel, LogPanelWriter};
 use crate::output::{CsvLogging, SbpLogging};
-use crate::types::{ClientSender, FlowControl, RealtimeDelay, SharedState};
+use crate::types::{ArcBool, ClientSender, FlowControl, RealtimeDelay, SharedState};
 use crate::utils::{refresh_loggingbar, refresh_navbar};
 
-/// The backend server
-#[pyclass]
-struct Server {
-    client_recv: Option<mpsc::Receiver<Vec<u8>>>,
-}
-
 #[pyclass]
 struct ServerEndpoint {
     server_send: Option<mpsc::Sender<Vec<u8>>>,
 }
+impl Drop for ServerEndpoint {
+    fn drop(&mut self) {
+        if let Some(sender) = self.server_send.take() {
+            drop(sender);
+        }
+    }
+}
 
 #[pymethods]
 impl ServerEndpoint {
@@ -101,23 +105,44 @@ fn handle_cli(opt: CliOptions, connection_state: &ConnectionState, shared_state:
     }
 }
 
+/// The backend server
+#[pyclass]
+struct Server {
+    client_recv: Option<mpsc::Receiver<Vec<u8>>>,
+    is_running: ArcBool,
+    handle: Option<thread::JoinHandle<()>>,
+}
+impl Drop for Server {
+    fn drop(&mut self) {
+        self.is_running.set(false);
+
+        if let Some(recv) = self.client_recv.take() {
+            drop(recv);
+        }
+    }
+}
+
 #[pymethods]
 impl Server {
     #[new]
     pub fn __new__() -> Self {
-        Server { client_recv: None }
+        Server {
+            client_recv: None,
+            is_running: ArcBool::new(),
+            handle: None,
+        }
     }
 
     #[text_signature = "($self, /)"]
     pub fn fetch_message(&mut self, py: Python) -> Option<PyObject> {
         let result = py.allow_threads(move || {
             let client_recv = self.client_recv.as_ref();
             if let Some(client_recv) = client_recv {
-                let buf = client_recv.recv();
-                if let Ok(buf) = buf {
+                if let Ok(buf) =
+                    client_recv.recv_timeout(Duration::from_millis(FETCH_MESSAGE_TIMEOUT_MS))
+                {
                     Some(buf)
                 } else {
-                    println!("error receiving message: {:?}", buf);
                     None
                 }
             } else {
@@ -156,7 +181,12 @@ impl Server {
         handle_cli(opt, &connection_state, shared_state.clone());
         refresh_navbar(&mut client_send.clone(), shared_state.clone());
         refresh_loggingbar(&mut client_send.clone(), shared_state.clone());
-        thread::spawn(move || loop {
+        self.is_running.set(true);
+        let is_running = self.is_running.clone();
+        self.handle = Some(thread::spawn(move || loop {
+            if !is_running.get() {
+                break;
+            }
             let buf = server_recv.recv();
             if let Ok(buf) = buf {
                 let mut buf_reader = BufReader::new(Cursor::new(buf));
@@ -176,6 +206,19 @@ impl Server {
                     }
                 };
                 match message {
+                    m::message::Status(Ok(cv_in)) => {
+                        let status_req =
+                            cv_in.get_text().expect(CAP_N_PROTO_DESERIALIZATION_FAILURE);
+                        match ApplicationStates::from_str(status_req) {
+                            Ok(ApplicationStates::CLOSE) => {
+                                let shared_state_clone = shared_state.clone();
+                                shared_state_clone.stop_server_running();
+                                is_running.set(false);
+                            }
+                            Ok(_) => {}
+                            Err(_) => {}
+                        }
+                    }
                     m::message::ConnectRequest(Ok(conn_req)) => {
                         let request = conn_req
                             .get_request()
@@ -328,7 +371,8 @@ impl Server {
                 println!("error: {:?}", buf);
                 break;
             }
-        });
+        }
+        ));
         Ok(server_endpoint)
     }
 }

src/main/python/main.py

@@ -120,6 +120,7 @@
 PIKSI_HOST = "piksi-relay-bb9f2b10e53143f4a816a11884e679cf.ce.swiftnav.com"
 PIKSI_PORT = 55555
 
+IS_RUNNING = True
 
 MAIN_INDEX = "MAIN_INDEX"
 SUB_INDEX = "SUB_INDEX"
@@ -180,13 +181,15 @@
 
 
 def receive_messages(app_, backend, messages):
-    while True:
+    while IS_RUNNING:
         buffer = backend.fetch_message()
+        if buffer is None:
+            continue
         Message = messages.Message
         m = Message.from_bytes(buffer)
         if m.which == Message.Union.Status:
             if m.status.text == ApplicationStates.CLOSE:
-                return app_.quit()
+                break
             if m.status.text == ApplicationStates.CONNECTED:
                 NAV_BAR[Keys.CONNECTED] = True
             elif m.status.text == ApplicationStates.DISCONNECTED:
@@ -298,7 +301,8 @@ def receive_messages(app_, backend, messages):
             log_panel_lock.unlock()
         else:
             pass
-
+    
+    app_.quit()
 
 class DataModel(QObject):
 
@@ -310,6 +314,14 @@ def __init__(self, endpoint, messages):
         self.endpoint = endpoint
         self.messages = messages
 
+    def status(self, status: ApplicationStates):
+        Message = self.messages.Message
+        msg = Message()
+        msg.status = msg.init(Message.Union.Status)
+        msg.status.text = str(status)
+        buffer = msg.to_bytes()
+        self.endpoint.send_message(buffer)
+
     @Slot()  # type: ignore
     def connect(self) -> None:
         self.connect_tcp(PIKSI_HOST, PIKSI_PORT)
@@ -584,14 +596,20 @@ def handle_cli_arguments(args: argparse.Namespace, globals_: QObject):
     globals_main = globals_main.property("globals")  # type: ignore
 
     handle_cli_arguments(args_main, globals_main)
-
-    threading.Thread(
+    running_lock = threading.Lock()
+    recv_thread = threading.Thread(
         target=receive_messages,
         args=(
             app,
             backend_main,
             messages_main,
         ),
         daemon=True,
-    ).start()
-    sys.exit(app.exec_())
+    )
+    recv_thread.start()
+    app.exec_()
+    data_model.status(ApplicationStates.CLOSE)
+    with running_lock:
+        IS_RUNNING = False
+    recv_thread.join()
+    sys.exit()