Prevent zombie processes in multi-process mode. (#36329)

This introduces a process manager that holds for each process a
"lifeline": this is the receiving end of a ipc channel that is not used
to send anything, but only to monitor the process presence. We turn that
ipc receiver into a crossbeam one to integrate the monitoring into the
constellation run loop. The sender side is made part of the initial
"UnprivilegedContent" data structure sent to the new process, both for
content and for service worker processes.
When a process dies we currently wait() on it to let the OS do a clean
shutdown.

Signed-off-by: webbeef <me@webbeef.org>

components/constellation/constellation.rs

@@ -113,7 +113,7 @@ use constellation_traits::{
     AnimationTickType, CompositorHitTestResult, ConstellationMsg as FromCompositorMsg, LogEntry,
     PaintMetricEvent, ScrollState, TraversalDirection, WindowSizeType,
 };
-use crossbeam_channel::{Receiver, Sender, select, unbounded};
+use crossbeam_channel::{Receiver, Select, Sender, unbounded};
 use devtools_traits::{
     ChromeToDevtoolsControlMsg, DevtoolsControlMsg, DevtoolsPageInfo, NavigationState,
     ScriptToDevtoolsControlMsg,
@@ -171,6 +171,7 @@ use crate::browsingcontext::{
 };
 use crate::event_loop::EventLoop;
 use crate::pipeline::{InitialPipelineState, Pipeline};
+use crate::process_manager::ProcessManager;
 use crate::serviceworker::ServiceWorkerUnprivilegedContent;
 use crate::session_history::{
     JointSessionHistory, NeedsToReload, SessionHistoryChange, SessionHistoryDiff,
@@ -469,6 +470,9 @@ pub struct Constellation<STF, SWF> {
 
     /// User content manager
     user_content_manager: UserContentManager,
+
+    /// The process manager.
+    process_manager: ProcessManager,
 }
 
 /// State needed to construct a constellation.
@@ -735,6 +739,7 @@ where
                     active_media_session: None,
                     rippy_data,
                     user_content_manager: state.user_content_manager,
+                    process_manager: ProcessManager::new(),
                 };
 
                 constellation.run();
@@ -996,6 +1001,12 @@ where
             );
         }
 
+        if let Some((lifeline_receiver, process)) = pipeline.lifeline {
+            let crossbeam_receiver =
+                route_ipc_receiver_to_new_crossbeam_receiver_preserving_errors(lifeline_receiver);
+            self.process_manager.add(crossbeam_receiver, process);
+        }
+
         assert!(!self.pipelines.contains_key(&pipeline_id));
         self.pipelines.insert(pipeline_id, pipeline.pipeline);
     }
@@ -1114,6 +1125,7 @@ where
             BackgroundHangMonitor(HangMonitorAlert),
             Compositor(FromCompositorMsg),
             FromSWManager(SWManagerMsg),
+            RemoveProcess(usize),
         }
         // Get one incoming request.
         // This is one of the few places where the compositor is
@@ -1126,26 +1138,49 @@ where
         // produces undefined behaviour, resulting in the destructor
         // being called. If this happens, there's not much we can do
         // other than panic.
+        let mut sel = Select::new();
+        sel.recv(&self.namespace_receiver);
+        sel.recv(&self.script_receiver);
+        sel.recv(&self.background_hang_monitor_receiver);
+        sel.recv(&self.compositor_receiver);
+        sel.recv(&self.swmanager_receiver);
+
+        self.process_manager.register(&mut sel);
+
         let request = {
+            let oper = sel.select();
+            let index = oper.index();
+
             #[cfg(feature = "tracing")]
             let _span =
                 tracing::trace_span!("handle_request::select", servo_profiling = true).entered();
-            select! {
-                recv(self.namespace_receiver) -> msg => {
-                    msg.expect("Unexpected script channel panic in constellation").map(Request::PipelineNamespace)
-                }
-                recv(self.script_receiver) -> msg => {
-                    msg.expect("Unexpected script channel panic in constellation").map(Request::Script)
-                }
-                recv(self.background_hang_monitor_receiver) -> msg => {
-                    msg.expect("Unexpected BHM channel panic in constellation").map(Request::BackgroundHangMonitor)
-                }
-                recv(self.compositor_receiver) -> msg => {
-                    Ok(Request::Compositor(msg.expect("Unexpected compositor channel panic in constellation")))
-                }
-                recv(self.swmanager_receiver) -> msg => {
-                    msg.expect("Unexpected SW channel panic in constellation").map(Request::FromSWManager)
-                }
+            match index {
+                0 => oper
+                    .recv(&self.namespace_receiver)
+                    .expect("Unexpected script channel panic in constellation")
+                    .map(Request::PipelineNamespace),
+                1 => oper
+                    .recv(&self.script_receiver)
+                    .expect("Unexpected script channel panic in constellation")
+                    .map(Request::Script),
+                2 => oper
+                    .recv(&self.background_hang_monitor_receiver)
+                    .expect("Unexpected BHM channel panic in constellation")
+                    .map(Request::BackgroundHangMonitor),
+                3 => Ok(Request::Compositor(
+                    oper.recv(&self.compositor_receiver)
+                        .expect("Unexpected compositor channel panic in constellation"),
+                )),
+                4 => oper
+                    .recv(&self.swmanager_receiver)
+                    .expect("Unexpected SW channel panic in constellation")
+                    .map(Request::FromSWManager),
+                _ => {
+                    // This can only be a error reading on a closed lifeline receiver.
+                    let process_index = index - 5;
+                    let _ = oper.recv(self.process_manager.receiver_at(process_index));
+                    Ok(Request::RemoveProcess(process_index))
+                },
             }
         };
 
@@ -1168,6 +1203,7 @@ where
             Request::FromSWManager(message) => {
                 self.handle_request_from_swmanager(message);
             },
+            Request::RemoveProcess(index) => self.process_manager.remove(index),
         }
     }
 
@@ -2398,13 +2434,24 @@ where
                     own_sender: own_sender.clone(),
                     receiver,
                 };
-                let content = ServiceWorkerUnprivilegedContent::new(sw_senders, origin);
 
                 if opts::get().multiprocess {
-                    if content.spawn_multiprocess().is_err() {
+                    let (sender, receiver) =
+                        ipc::channel().expect("Failed to create lifeline channel for sw");
+                    let content =
+                        ServiceWorkerUnprivilegedContent::new(sw_senders, origin, Some(sender));
+
+                    if let Ok(process) = content.spawn_multiprocess() {
+                        let crossbeam_receiver =
+                            route_ipc_receiver_to_new_crossbeam_receiver_preserving_errors(
+                                receiver,
+                            );
+                        self.process_manager.add(crossbeam_receiver, process);
+                    } else {
                         return warn!("Failed to spawn process for SW manager.");
                     }
                 } else {
+                    let content = ServiceWorkerUnprivilegedContent::new(sw_senders, origin, None);
                     content.start::<SWF>();
                 }
                 entry.insert(own_sender)

components/constellation/lib.rs

@@ -12,6 +12,7 @@ mod constellation;
 mod event_loop;
 mod logging;
 mod pipeline;
+mod process_manager;
 mod sandboxing;
 mod serviceworker;
 mod session_history;

components/constellation/pipeline.rs

@@ -46,6 +46,7 @@ use webrender_api::DocumentId;
 use webrender_traits::CrossProcessCompositorApi;
 
 use crate::event_loop::EventLoop;
+use crate::process_manager::Process;
 use crate::sandboxing::{UnprivilegedContent, spawn_multiprocess};
 
 /// A `Pipeline` is the constellation's view of a `Window`. Each pipeline has an event loop
@@ -201,6 +202,7 @@ pub struct InitialPipelineState {
 pub struct NewPipeline {
     pub pipeline: Pipeline,
     pub bhm_control_chan: Option<IpcSender<BackgroundHangMonitorControlMsg>>,
+    pub lifeline: Option<(IpcReceiver<()>, Process)>,
 }
 
 impl Pipeline {
@@ -210,7 +212,7 @@ impl Pipeline {
     ) -> Result<NewPipeline, Error> {
         // Note: we allow channel creation to panic, since recovering from this
         // probably requires a general low-memory strategy.
-        let (script_chan, bhm_control_chan) = match state.event_loop {
+        let (script_chan, (bhm_control_chan, lifeline)) = match state.event_loop {
             Some(script_chan) => {
                 let new_layout_info = NewLayoutInfo {
                     parent_info: state.parent_pipeline_id,
@@ -226,7 +228,7 @@ impl Pipeline {
                 {
                     warn!("Sending to script during pipeline creation failed ({})", e);
                 }
-                (script_chan, None)
+                (script_chan, (None, None))
             },
             None => {
                 let (script_chan, script_port) = ipc::channel().expect("Pipeline script chan");
@@ -292,17 +294,21 @@ impl Pipeline {
                     player_context: state.player_context,
                     rippy_data: state.rippy_data,
                     user_content_manager: state.user_content_manager,
+                    lifeline_sender: None,
                 };
 
                 // Spawn the child process.
                 //
                 // Yes, that's all there is to it!
-                let bhm_control_chan = if opts::get().multiprocess {
+                let multiprocess_data = if opts::get().multiprocess {
                     let (bhm_control_chan, bhm_control_port) =
                         ipc::channel().expect("Sampler chan");
                     unprivileged_pipeline_content.bhm_control_port = Some(bhm_control_port);
-                    unprivileged_pipeline_content.spawn_multiprocess()?;
-                    Some(bhm_control_chan)
+                    let (sender, receiver) =
+                        ipc::channel().expect("Failed to create lifeline channel");
+                    unprivileged_pipeline_content.lifeline_sender = Some(sender);
+                    let process = unprivileged_pipeline_content.spawn_multiprocess()?;
+                    (Some(bhm_control_chan), Some((receiver, process)))
                 } else {
                     // Should not be None in single-process mode.
                     let register = state
@@ -313,10 +319,10 @@ impl Pipeline {
                         state.layout_factory,
                         register,
                     );
-                    None
+                    (None, None)
                 };
 
-                (EventLoop::new(script_chan), bhm_control_chan)
+                (EventLoop::new(script_chan), multiprocess_data)
             },
         };
 
@@ -333,6 +339,7 @@ impl Pipeline {
         Ok(NewPipeline {
             pipeline,
             bhm_control_chan,
+            lifeline,
         })
     }
 
@@ -498,6 +505,7 @@ pub struct UnprivilegedPipelineContent {
     player_context: WindowGLContext,
     rippy_data: Vec<u8>,
     user_content_manager: UserContentManager,
+    lifeline_sender: Option<IpcSender<()>>,
 }
 
 impl UnprivilegedPipelineContent {
@@ -558,7 +566,7 @@ impl UnprivilegedPipelineContent {
         }
     }
 
-    pub fn spawn_multiprocess(self) -> Result<(), Error> {
+    pub fn spawn_multiprocess(self) -> Result<Process, Error> {
         spawn_multiprocess(UnprivilegedContent::Pipeline(self))
     }
 

components/constellation/process_manager.rs

@@ -0,0 +1,68 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+use std::process::Child;
+
+use crossbeam_channel::{Receiver, Select};
+use log::{debug, warn};
+
+pub enum Process {
+    Unsandboxed(Child),
+    Sandboxed(u32),
+}
+
+impl Process {
+    fn pid(&self) -> u32 {
+        match self {
+            Self::Unsandboxed(child) => child.id(),
+            Self::Sandboxed(pid) => *pid,
+        }
+    }
+
+    fn wait(&mut self) {
+        match self {
+            Self::Unsandboxed(child) => {
+                let _ = child.wait();
+            },
+            Self::Sandboxed(_pid) => {
+                // TODO: use nix::waitpid() on supported platforms.
+                warn!("wait() is not yet implemented for sandboxed processes.");
+            },
+        }
+    }
+}
+
+type ProcessReceiver = Receiver<Result<(), ipc_channel::Error>>;
+
+pub(crate) struct ProcessManager {
+    processes: Vec<(Process, ProcessReceiver)>,
+}
+
+impl ProcessManager {
+    pub fn new() -> Self {
+        Self { processes: vec![] }
+    }
+
+    pub fn add(&mut self, receiver: ProcessReceiver, process: Process) {
+        debug!("Adding process pid={}", process.pid());
+        self.processes.push((process, receiver));
+    }
+
+    pub fn register<'a>(&'a self, select: &mut Select<'a>) {
+        for (_, receiver) in &self.processes {
+            select.recv(receiver);
+        }
+    }
+
+    pub fn receiver_at(&self, index: usize) -> &ProcessReceiver {
+        let (_, receiver) = &self.processes[index];
+        receiver
+    }
+
+    pub fn remove(&mut self, index: usize) {
+        let (mut process, _) = self.processes.swap_remove(index);
+        debug!("Removing process pid={}", process.pid());
+        process.wait();
+    }
+}

components/constellation/sandboxing.rs

@@ -25,6 +25,7 @@ use servo_config::opts::Opts;
 use servo_config::prefs::Preferences;
 
 use crate::pipeline::UnprivilegedPipelineContent;
+use crate::process_manager::Process;
 use crate::serviceworker::ServiceWorkerUnprivilegedContent;
 
 #[derive(Deserialize, Serialize)]
@@ -146,7 +147,7 @@ pub fn content_process_sandbox_profile() {
     target_arch = "arm",
     all(target_arch = "aarch64", not(target_os = "windows"))
 ))]
-pub fn spawn_multiprocess(content: UnprivilegedContent) -> Result<(), Error> {
+pub fn spawn_multiprocess(content: UnprivilegedContent) -> Result<Process, Error> {
     use ipc_channel::ipc::{IpcOneShotServer, IpcSender};
     // Note that this function can panic, due to process creation,
     // avoiding this panic would require a mechanism for dealing
@@ -158,15 +159,14 @@ pub fn spawn_multiprocess(content: UnprivilegedContent) -> Result<(), Error> {
     let mut child_process = process::Command::new(path_to_self);
     setup_common(&mut child_process, token);
 
-    #[allow(clippy::zombie_processes)]
-    let _ = child_process
+    let child = child_process
         .spawn()
         .expect("Failed to start unsandboxed child process!");
 
     let (_receiver, sender) = server.accept().expect("Server failed to accept.");
     sender.send(content)?;
 
-    Ok(())
+    Ok(Process::Unsandboxed(child))
 }
 
 #[cfg(all(
@@ -177,7 +177,7 @@ pub fn spawn_multiprocess(content: UnprivilegedContent) -> Result<(), Error> {
     not(target_arch = "arm"),
     not(target_arch = "aarch64")
 ))]
-pub fn spawn_multiprocess(content: UnprivilegedContent) -> Result<(), Error> {
+pub fn spawn_multiprocess(content: UnprivilegedContent) -> Result<Process, Error> {
     use gaol::sandbox::{self, Sandbox, SandboxMethods};
     use ipc_channel::ipc::{IpcOneShotServer, IpcSender};
 
@@ -208,33 +208,37 @@ pub fn spawn_multiprocess(content: UnprivilegedContent) -> Result<(), Error> {
         .expect("Failed to create IPC one-shot server.");
 
     // If there is a sandbox, use the `gaol` API to create the child process.
-    if content.opts().sandbox {
+    let process = if content.opts().sandbox {
         let mut command = sandbox::Command::me().expect("Failed to get current sandbox.");
         setup_common(&mut command, token);
 
         let profile = content_process_sandbox_profile();
-        let _ = Sandbox::new(profile)
-            .start(&mut command)
-            .expect("Failed to start sandboxed child process!");
+        Process::Sandboxed(
+            Sandbox::new(profile)
+                .start(&mut command)
+                .expect("Failed to start sandboxed child process!")
+                .pid as u32,
+        )
     } else {
         let path_to_self = env::current_exe().expect("Failed to get current executor.");
         let mut child_process = process::Command::new(path_to_self);
         setup_common(&mut child_process, token);
 
-        #[allow(clippy::zombie_processes)]
-        let _ = child_process
-            .spawn()
-            .expect("Failed to start unsandboxed child process!");
-    }
+        Process::Unsandboxed(
+            child_process
+                .spawn()
+                .expect("Failed to start unsandboxed child process!"),
+        )
+    };
 
     let (_receiver, sender) = server.accept().expect("Server failed to accept.");
     sender.send(content)?;
 
-    Ok(())
+    Ok(process)
 }
 
 #[cfg(any(target_os = "windows", target_os = "ios"))]
-pub fn spawn_multiprocess(_content: UnprivilegedContent) -> Result<(), Error> {
+pub fn spawn_multiprocess(_content: UnprivilegedContent) -> Result<Process, Error> {
     log::error!("Multiprocess is not supported on Windows or iOS.");
     process::exit(1);
 }

components/constellation/serviceworker.rs

@@ -3,6 +3,7 @@
  * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
 
 use ipc_channel::Error;
+use ipc_channel::ipc::IpcSender;
 use script_traits::{SWManagerSenders, ServiceWorkerManagerFactory};
 use serde::{Deserialize, Serialize};
 use servo_config::opts::{self, Opts};
@@ -10,6 +11,7 @@ use servo_config::prefs;
 use servo_config::prefs::Preferences;
 use servo_url::ImmutableOrigin;
 
+use crate::process_manager::Process;
 use crate::sandboxing::{UnprivilegedContent, spawn_multiprocess};
 
 /// Conceptually, this is glue to start an agent-cluster for a service worker agent.
@@ -20,18 +22,21 @@ pub struct ServiceWorkerUnprivilegedContent {
     prefs: Box<Preferences>,
     senders: SWManagerSenders,
     origin: ImmutableOrigin,
+    lifeline_sender: Option<IpcSender<()>>,
 }
 
 impl ServiceWorkerUnprivilegedContent {
     pub fn new(
         senders: SWManagerSenders,
         origin: ImmutableOrigin,
+        lifeline_sender: Option<IpcSender<()>>,
     ) -> ServiceWorkerUnprivilegedContent {
         ServiceWorkerUnprivilegedContent {
             opts: (*opts::get()).clone(),
             prefs: Box::new(prefs::get().clone()),
             senders,
             origin,
+            lifeline_sender,
         }
     }
 
@@ -44,7 +49,7 @@ impl ServiceWorkerUnprivilegedContent {
     }
 
     /// Start the agent-cluster in it's own process.
-    pub fn spawn_multiprocess(self) -> Result<(), Error> {
+    pub fn spawn_multiprocess(self) -> Result<Process, Error> {
         spawn_multiprocess(UnprivilegedContent::ServiceWorker(self))
     }