folk_runtime_fork/
master.rs1use std::os::unix::io::{AsRawFd, OwnedFd};
4
5use anyhow::{Context, Result, bail};
6use folk_protocol::{FrameCodec, RpcMessage};
7use futures_util::{SinkExt, StreamExt};
8use rmpv::Value;
9use tokio::net::UnixStream;
10use tokio_util::codec::Framed;
11use tracing::{debug, info};
12
13use crate::scm_rights::send_fds;
14
15pub struct PreforkMaster {
17 control: Framed<UnixStream, FrameCodec>,
18 pid: u32,
19 fork_socket: UnixStream,
21}
22
23impl PreforkMaster {
24 pub async fn spawn(php: &str, script: &str, boot_timeout: std::time::Duration) -> Result<Self> {
27 let spawned =
30 folk_runtime_pipe::spawn::spawn_worker_with_runtime(php, script, "fork")
31 .context("spawn prefork master")?;
32 let mut control = Framed::new(spawned.control_master, FrameCodec::new());
33 let pid = spawned.child.id().unwrap_or(0);
34
35 let ready = tokio::time::timeout(boot_timeout, control.next())
37 .await
38 .context("boot timeout")?
39 .context("EOF before fork-ready")?
40 .context("decode fork-ready")?;
41
42 match ready {
43 RpcMessage::Notify { ref method, .. } if method == "control.fork-ready" => {
44 info!(pid, "prefork master ready");
45 },
46 other => bail!("expected control.fork-ready, got {other:?}"),
47 }
48
49 Ok(Self {
50 control,
51 pid,
52 fork_socket: spawned.task_master,
53 })
54 }
55
56 #[allow(unsafe_code)]
59 pub async fn fork_worker(&mut self, task_child: &OwnedFd, ctrl_child: &OwnedFd) -> Result<u32> {
60 unsafe {
62 send_fds(
63 self.fork_socket.as_raw_fd(),
64 &[task_child.as_raw_fd(), ctrl_child.as_raw_fd()],
65 )?;
66 }
67
68 self.control
70 .send(RpcMessage::notify("fork.spawn", Value::Nil))
71 .await?;
72
73 let reply = tokio::time::timeout(std::time::Duration::from_secs(10), self.control.next())
75 .await
76 .context("fork reply timeout")?
77 .context("EOF from master")?
78 .context("decode fork reply")?;
79
80 match reply {
81 RpcMessage::Notify { method, params } if method == "fork.spawned" => {
82 #[allow(clippy::cast_possible_truncation)]
83 let pid = params
84 .as_map()
85 .and_then(|m| m.iter().find(|(k, _)| k.as_str() == Some("pid")))
86 .and_then(|(_, v)| v.as_u64())
87 .context("missing pid in fork.spawned")? as u32;
88 debug!(child_pid = pid, "child forked");
89 Ok(pid)
90 },
91 other => bail!("unexpected reply to fork.spawn: {other:?}"),
92 }
93 }
94
95 #[allow(unsafe_code, clippy::cast_possible_wrap)]
97 pub async fn shutdown(&mut self) -> Result<()> {
98 let _ = self
99 .control
100 .send(RpcMessage::notify("control.shutdown", Value::Nil))
101 .await;
102 unsafe {
103 libc::kill(self.pid as libc::pid_t, libc::SIGTERM);
104 }
105 Ok(())
106 }
107}