synit/synit-pid1/src/main.rs

use clap::Clap;

use nix::sys::signal::{killpg, Signal};
use nix::sys::wait;
use nix::unistd;

use std::convert::TryInto;
use std::sync::Arc;

use syndicate::actor::*;
use syndicate::relay;
use syndicate::sturdy;
use syndicate::value::NestedValue;
use syndicate::value::Value;

use tokio::process;
use tokio::select;
use tokio::signal::unix::{signal, SignalKind};

#[derive(Clap, Clone, Debug)]
#[clap(version)]
pub struct CommandLine {
    #[clap(long, default_value="/usr/bin/syndicate-server")]
    server_path: String,

    #[clap(long, default_value="/sbin/synit-log")]
    log: String,
}

pub struct Pid1Listener {
    // config: Arc<CommandLine>,
}

impl Pid1Listener {
    fn new(_config: Arc<CommandLine>) -> Self {
        Self {
            // config,
        }
    }
}

impl Entity<AnyValue> for Pid1Listener {
}

async fn handle_sigchld_and_waitpid() -> Result<(), Box<dyn std::error::Error>> {
    let mut sigchlds = signal(SignalKind::child())?;
    let mut sigints = signal(SignalKind::interrupt())?;
    let mut sigterms = signal(SignalKind::terminate())?;

    tracing::info!("Awaiting signals...");
    loop {
        select! {
            _ = sigchlds.recv() => {
                loop {
                    match wait::waitpid(None, Some(wait::WaitPidFlag::WNOHANG)) {
                        Ok(wait::WaitStatus::StillAlive) => {
                            tracing::debug!("No child processes to reap at this time");
                            break;
                        }
                        Ok(status) =>
                            tracing::debug!("Child process reaped: {:?}", status),
                        Err(nix::errno::Errno::ECHILD) => {
                            tracing::debug!("waitpid(2) yielded ECHILD");
                            break;
                        }
                        Err(e) =>
                            Err(e)?,
                    }
                }
            }
            _ = sigints.recv() => {
                tracing::debug!("Received SIGINT");
                let result = killpg(unistd::getpgrp(), Some(Signal::SIGINT));
                tracing::debug!("killpg result: {:?}", result);
                break;
            }
            _ = sigterms.recv() => {
                tracing::debug!("Received SIGTERM");
                let result = killpg(unistd::getpgrp(), Some(Signal::SIGTERM));
                tracing::debug!("killpg result: {:?}", result);
                break;
            }
        }
    }

    Ok(())
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    syndicate::convenient_logging()?;

    match unistd::setsid() {
        Ok(_pid) => tracing::info!("setsid(2): new session is {}", _pid),
        Err(e) => tracing::info!("setsid(2) failed: {:?}", &e),
    }

    match unistd::setpgid(unistd::Pid::from_raw(0), unistd::Pid::from_raw(0)) {
        Ok(()) => tracing::info!("setpgid(2) succeeded"),
        Err(e) => tracing::info!("setpgid(2) failed: {:?}", &e),
    }

    let config = Arc::new(CommandLine::parse());

    Actor::top(None, move |t| {
        let server = process::Command::new(&config.server_path)
            .arg("--inferior")
            .arg("--config")
            .arg("/etc/syndicate/boot")
            .stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped())
            .spawn()
            .expect("Could not spawn main server");

        let server_stderr: std::process::Stdio = server.stderr
            .expect("fetching server stderr")
            .try_into()
            .expect("converting server stderr to Stdio");

        if config.log.len() > 0 {
            process::Command::new(&config.log)
                .stdin(server_stderr)
                .spawn()
                .expect("Could not spawn log program");
        }

        let listener = t.create(Pid1Listener::new(config));
        let from_server = server.stdout.expect("Missing dataspace server stdout");
        let to_server = server.stdin.expect("Missing dataspace server stdin");

        let ds = &relay::TunnelRelay::run(t,
                                          relay::Input::Bytes(Box::pin(from_server)),
                                          relay::Output::Bytes(Box::pin(to_server)),
                                          None,
                                          Some(sturdy::Oid(0.into())),
                                          false)
            .expect("Missing reference to dataspace")
            .underlying;
        t.assert(ds, Value::simple_record1("init", AnyValue::domain(Cap::new(&listener))).wrap());
        Ok(())
    });

    handle_sigchld_and_waitpid().await?;

    Ok(())
}