feat: finish initial ct rr worker communication: now with sockets

there were problems with stdio, this is just ok for now
some help from Nikola for some bugs

Author: alehander92

src/db-backend/src/main.rs

@@ -126,18 +126,25 @@ fn main() -> Result<(), Box<dyn Error>> {
     match cli.cmd {
         Commands::DapServer { socket_path, stdio } => {
             if stdio {
-                let _ = db_backend::dap_server::run_stdio();
+                // thread::spawn(move || {
+                let res = db_backend::dap_server::run_stdio();
+                if let Err(e) = res {
+                    error!("dap server run error: {e:?}");
+                }
+                // })
             } else {
                 let socket_path = if let Some(p) = socket_path {
                     p
                 } else {
                     let pid = std::process::id() as usize;
                     db_backend::dap_server::socket_path_for(pid)
                 };
-
-                info!("dap_server::run {:?}", socket_path);
-
-                let _ = db_backend::dap_server::run(&socket_path);
+                // thread::spawn(move || {
+                let res = db_backend::dap_server::run(&socket_path);
+                if let Err(e) = res {
+                    error!("dap server run error: {e:?}");
+                }
+                // })
             };
         }
         Commands::IndexDiff {

src/db-backend/src/paths.rs

@@ -1,5 +1,9 @@
+// copied from Stan's paths.rs in src/db-backend in the public codetracer repo
+//   added run_dir/ct_rr_worker_socket_path (and copied to db-backend too)
+
 use std::env;
-use std::path::PathBuf;
+use std::error::Error;
+use std::path::{Path, PathBuf};
 use std::sync::{LazyLock, Mutex};
 
 pub struct Paths {
@@ -24,3 +28,24 @@ impl Default for Paths {
 }
 
 pub static CODETRACER_PATHS: LazyLock<Mutex<Paths>> = LazyLock::new(|| Mutex::new(Paths::default()));
+
+pub fn run_dir_for(tmp_path: &Path, run_id: usize) -> Result<PathBuf, Box<dyn Error>> {
+    let run_dir = tmp_path.join(format!("run-{run_id}"));
+    std::fs::create_dir_all(&run_dir)?;
+    Ok(run_dir)
+}
+
+pub fn ct_rr_worker_socket_path(from: &str, worker_name: &str, run_id: usize) -> Result<PathBuf, Box<dyn Error>> {
+    let tmp_path: PathBuf = { CODETRACER_PATHS.lock()?.tmp_path.clone() };
+    let run_dir = run_dir_for(&tmp_path, run_id)?;
+    // eventually: TODO: unique index or better cleanup
+    //  if worker with the same name started/restarted multiple times
+    //  by the same backend instance
+    Ok(run_dir.join(format!("ct_rr_worker_{worker_name}_from_{from}.sock")))
+
+    // TODO: decide if we need to check/eventually remove or the unique run folder/paths are enough:
+    //
+    // if std::fs::metadata(&receiving_socket_path).is_ok() {
+    // let _ = std::fs::remove_file(&receiving_socket_path); // try to remove if existing: ignore error
+    // }
+}

src/db-backend/src/rr_dispatcher.rs

@@ -1,12 +1,15 @@
 use std::error::Error;
 use std::io::Write;
-use std::io::{BufRead, BufReader, Read};
+use std::io::{BufRead, BufReader};
+use std::os::unix::net::UnixStream;
 use std::path::{Path, PathBuf};
 use std::process::{Child, Command, Stdio};
 use std::thread;
+use std::time::Duration;
 
 use log::info;
 
+use crate::paths::ct_rr_worker_socket_path;
 use crate::query::CtRRQuery;
 
 #[derive(Debug)]
@@ -23,6 +26,8 @@ pub struct CtRRWorker {
     pub ct_rr_worker_exe: PathBuf,
     pub rr_trace_folder: PathBuf,
     process: Option<Child>,
+    sending_stream: Option<UnixStream>,
+    receiving_stream: Option<UnixStream>,
 }
 
 #[derive(Default)]
@@ -39,6 +44,8 @@ impl CtRRWorker {
             ct_rr_worker_exe: PathBuf::from(ct_rr_worker_exe),
             rr_trace_folder: PathBuf::from(rr_trace_folder),
             process: None,
+            sending_stream: None,
+            receiving_stream: None,
         }
     }
 
@@ -59,47 +66,70 @@ impl CtRRWorker {
             .spawn()?;
 
         self.process = Some(ct_worker);
-        thread::sleep_ms(1_000);
+        self.setup_worker_sockets()?;
         self.active = true;
         Ok(())
     }
 
+    fn setup_worker_sockets(&mut self) -> Result<(), Box<dyn Error>> {
+        // assuming that the ct rr worker creates the sockets!
+        // code copied and adapted from `connect_socket_with_backend_and_loop` in ct-rr-worker
+        //   which is itself copied/adapted/written from/based on https://emmanuelbosquet.com/2022/whatsaunixsocket/
+
+        let run_id = std::process::id() as usize;
+
+        // sending socket:
+        let sending_socket_path = ct_rr_worker_socket_path("backend", &self.name, run_id)?;
+        info!(
+            "try to connect to worker with sending socket in {}",
+            sending_socket_path.display()
+        );
+        loop {
+            if let Ok(sending_stream) = UnixStream::connect(&sending_socket_path) {
+                self.sending_stream = Some(sending_stream);
+                break;
+            }
+            thread::sleep(Duration::from_millis(1));
+            // TODO: handle different kinds of errors
+        }
+        // receiving socket:
+        let receiving_socket_path = ct_rr_worker_socket_path("worker", &self.name, run_id)?;
+        info!(
+            "try to connect to worker with receiving socket in {}",
+            receiving_socket_path.display()
+        );
+        loop {
+            if let Ok(receiving_stream) = UnixStream::connect(&receiving_socket_path) {
+                self.receiving_stream = Some(receiving_stream);
+                break;
+            }
+            thread::sleep(Duration::from_millis(1));
+            // TODO: handle different kinds of errors
+        }
+
+        Ok(())
+    }
+
     // for now: don't return a typed value here, only ok or an error
     pub fn run_query(&mut self, query: CtRRQuery) -> Result<(), Box<dyn Error>> {
-        let mut stdin = self
-            .process
-            .as_mut()
-            .expect("valid process")
-            .stdin
-            .take()
-            .expect("stdin: TODO error");
-        let mut stdout = self
-            .process
-            .as_mut()
-            .expect("valid process")
-            .stdout
-            .take()
-            .expect("stdout: TODO error");
-
         let raw_json = serde_json::to_string(&query)?;
-        let reader = BufReader::new(stdout);
 
         info!("send to worker {raw_json}\n");
-        write!(stdin, "{}\n", raw_json)?;
+        self.sending_stream
+            .as_mut()
+            .expect("valid sending stream")
+            .write(&format!("{raw_json}\n").into_bytes())?;
 
         let mut res = "".to_string();
         info!("wait to read");
 
-        for line_result in reader.lines() {
-            info!("line_result {line_result:?}");
-            if let Ok(line) = line_result {
-                res.push_str(&line);
-                res.push_str("\n");
-            } else {
-                continue;
-            }
-        }
+        let mut reader = BufReader::new(self.receiving_stream.as_mut().expect("valid receiving stream"));
+        reader.read_line(&mut res)?;
+
+        res = String::from(res.trim()); // trim newlines/whitespace!
+
         info!("res {res}");
+
         if res == "ok" {
             Ok(())
         } else {