Continued designing language server <-> daemon communication and interfaces

Author: IsaacShelton

Cargo.lock

@@ -3,11 +3,12 @@ resolver = "3"
 members = [
     "src/adept",
     "src/daemon",
-    "src/daemon_scheduler",
+    "src/daemon_init",
     "src/language_server",
     "src/lock_file",
     "src/idle_tracker",
     "src/lsp_connection",
+    "src/lsp_message",
 ]
 
 [workspace.dependencies]

Cargo.toml

@@ -4,7 +4,7 @@ version = "0.1.0"
 edition = "2024"
 
 [dependencies]
-daemon = { version = "0.1.0", path = "../daemon" }
+daemon_init = { version = "0.1.0", path = "../daemon_init" }
 language_server = { version = "0.1.0", path = "../language_server" }
 smol.workspace = true
 

src/adept/Cargo.toml

@@ -5,7 +5,7 @@ fn main() -> ExitCode {
 
     match args.peek().map(String::as_str) {
         Some("-h" | "--help") | None => show_help(),
-        Some("--daemon") => daemon::start(),
+        Some("--daemon") => daemon_init::start(),
         Some("--language-server") => language_server::start(),
         _ => show_help(),
     }

src/adept/src/main.rs

@@ -5,6 +5,9 @@ edition = "2024"
 
 [dependencies]
 lock_file = { version = "0.1.0", path = "../lock_file" }
-daemon_scheduler = { version = "0.1.0", path = "../daemon_scheduler" }
-smol.workspace = true
+lsp_message = { version = "0.1.0", path = "../lsp_message" }
 thiserror.workspace = true
+derive_more.workspace = true
+fern.workspace = true
+humantime.workspace = true
+log.workspace = true

src/daemon/Cargo.toml

@@ -0,0 +1,23 @@
+use lsp_message::LspMessage;
+use std::io::BufReader;
+#[cfg(target_family = "unix")]
+use std::os::unix::net::UnixStream;
+#[cfg(target_family = "unix")]
+use std::{io, sync::Mutex};
+
+pub struct Connection {
+    #[cfg(target_family = "unix")]
+    pub stream: Mutex<UnixStream>,
+}
+
+impl Connection {
+    pub fn wait_for_message(&self) -> io::Result<Option<LspMessage>> {
+        let stream = self.stream.lock().unwrap();
+        LspMessage::read(&mut BufReader::new(&*stream))
+    }
+
+    pub fn send(&self, message: LspMessage) -> io::Result<()> {
+        let stream = self.stream.lock().unwrap();
+        message.write(&mut &*stream)
+    }
+}

src/daemon/src/connection.rs

@@ -0,0 +1,44 @@
+use crate::Queue;
+use lsp_message::LspMessage;
+#[cfg(target_family = "unix")]
+use std::os::unix::net::UnixListener;
+use std::{
+    io::{self, BufReader},
+    os::unix::net::UnixStream,
+    sync::atomic::{AtomicBool, Ordering},
+};
+
+pub struct Daemon {
+    #[cfg(target_family = "unix")]
+    pub listener: UnixListener,
+
+    pub should_exit: AtomicBool,
+    pub queue: Queue,
+}
+
+impl Daemon {
+    #[cfg(target_family = "unix")]
+    pub fn new(listener: UnixListener) -> Self {
+        Self {
+            listener,
+            should_exit: false.into(),
+            queue: Queue::default(),
+        }
+    }
+
+    pub fn wait_for_message(&self, client_stream: &UnixStream) -> io::Result<Option<LspMessage>> {
+        LspMessage::read(&mut BufReader::new(client_stream))
+    }
+
+    pub fn send(message: LspMessage, mut client_stream: &UnixStream) -> io::Result<()> {
+        message.write(&mut client_stream)
+    }
+
+    pub fn intend_exit(&self) {
+        self.should_exit.store(true, Ordering::SeqCst);
+    }
+
+    pub fn should_exit(&self) -> bool {
+        self.should_exit.load(Ordering::SeqCst)
+    }
+}

src/daemon/src/daemon.rs

@@ -1,71 +1,84 @@
-mod error;
+mod connection;
+mod daemon;
+mod logger;
+mod queue;
 
-pub use error::*;
-use smol::{Timer, net::TcpStream};
+pub use crate::{connection::Connection, daemon::Daemon};
+use lsp_message::LspMessage;
+pub use queue::*;
 use std::{
-    fs::remove_file,
-    io,
-    process::{Command, ExitCode},
+    io::{self, BufReader, ErrorKind},
+    os::unix::net::{SocketAddr, UnixStream},
+    sync::Arc,
     time::Duration,
 };
 
-pub fn start() -> ExitCode {
-    match try_become() {
-        Ok(()) => ExitCode::SUCCESS,
-        Err(error) => {
-            eprintln!("{}", error);
-            ExitCode::FAILURE
-        }
+pub fn main_loop(daemon: Daemon) -> io::Result<()> {
+    let daemon = Arc::new(daemon);
+
+    if logger::setup().is_err() {
+        eprintln!("Failed to create daemon log file");
     }
-}
 
-/// Become the daemon process
-pub fn try_become() -> io::Result<()> {
-    let cwd = std::env::current_dir().expect("Failed to get current directory");
-    let filepath = cwd.join("adeptd.lock");
+    daemon
+        .listener
+        .set_nonblocking(true)
+        .expect("Failed to set to non-blocking");
 
-    let Some(lock) = lock_file::acquire(&filepath)? else {
-        eprintln!("Daemon already running.");
-        return Ok(());
-    };
+    // Executor thread
+    std::thread::spawn(|| {});
 
-    eprintln!("Starting daemon...");
-    daemon_scheduler::main()?;
+    // Accept clients
+    #[cfg(target_family = "unix")]
+    loop {
+        match daemon.listener.accept() {
+            Ok((stream, address)) => {
+                let daemon = Arc::clone(&daemon);
+                std::thread::spawn(move || handle_client(daemon, stream, address));
+            }
+            Err(error) => {
+                if let io::ErrorKind::WouldBlock = error.kind() {
+                    // No clients ready to connect to us yet
+                } else {
+                    log::error!("Failed to accept client: {:?}", error);
+                }
+            }
+        }
 
-    eprintln!("Daemon shutting down...");
-    drop(lock);
+        if daemon.should_exit() {
+            return Ok(());
+        }
 
-    remove_file(&filepath)
+        std::thread::sleep(Duration::from_millis(50));
+    }
 }
 
-/// Tries to connect to the daemon process. If the daemon process
-/// is not running yet, this function attempts to launch it.
-pub async fn connect() -> Result<TcpStream, StartError> {
-    if let Ok(connection) = TcpStream::connect("127.0.0.1:6000").await {
-        eprintln!("Connected to existing daemon.");
-        return Ok(connection);
-    }
+fn handle_client(_daemon: Arc<Daemon>, stream: UnixStream, address: SocketAddr) {
+    log::info!("Accepted client {:?} {:?}", stream, address);
+    std::thread::sleep(Duration::from_millis(50));
 
-    spawn()?;
+    stream.set_nonblocking(false).unwrap();
+    stream
+        .set_read_timeout(Some(Duration::from_millis(50)))
+        .unwrap();
+    let reader = &mut BufReader::new(&stream);
 
-    for _ in 0..10 {
-        if let Ok(connection) = TcpStream::connect("127.0.0.1:6000").await {
-            return Ok(connection);
+    loop {
+        match LspMessage::read(reader) {
+            Ok(None) => {
+                log::info!("Shutting down connection to client");
+                break;
+            }
+            Ok(message) => {
+                log::info!("Got message {:?}", message);
+            }
+            Err(error) => {
+                if let ErrorKind::WouldBlock = error.kind() {
+                    // Nothing to do
+                } else {
+                    log::error!("Error receiving message from client - {:?}", error);
+                }
+            }
         }
-        Timer::after(Duration::from_millis(20)).await;
     }
-
-    Err(StartError::FailedToStart)
-}
-
-pub fn spawn() -> std::io::Result<()> {
-    let exe = std::env::current_exe()?;
-
-    // WARNING: SECURITY: This could lead to privilege escalation
-    // to the level the compiler is running at if an attacker
-    // overwrites the current executable.
-    // TL;DR - Don't let the compiler executable be changed
-    // by less privileged users.
-    Command::new(exe).arg("--daemon").spawn()?;
-    Ok(())
 }

src/daemon/src/lib.rs

@@ -0,0 +1,20 @@
+use std::{fs::File, time::SystemTime};
+
+pub(crate) fn setup() -> Result<(), fern::InitError> {
+    fern::Dispatch::new()
+        .format(|out, message, record| {
+            out.finish(format_args!(
+                "[{} {} {}] {}",
+                humantime::format_rfc3339_seconds(SystemTime::now()),
+                record.level(),
+                record.target(),
+                message
+            ))
+        })
+        .level(log::LevelFilter::Debug)
+        .chain(std::io::stderr())
+        .chain(File::create("adeptd.log")?)
+        .apply()?;
+    Ok(())
+}
+

src/daemon/src/logger.rs

@@ -0,0 +1,19 @@
+use std::{collections::VecDeque, sync::Mutex};
+
+// TODO
+#[derive(Default)]
+pub struct PendingQuery;
+
+#[derive(Default)]
+pub struct Queue {
+    // We can add priorities for this later,
+    // but even just running everything at the same time/duration step value
+    // is probably fully sufficient.
+    _queue: Mutex<VecDeque<PendingQuery>>,
+}
+
+impl Queue {
+    pub fn next(&self) -> Option<PendingQuery> {
+        todo!()
+    }
+}