Add broker-coordinated child process startup

litebox/src/litebox.rs

@@ -6,6 +6,7 @@
 use alloc::sync::Arc;
 
 use litebox_broker_local::BrokerLocal;
+use litebox_broker_protocol::ProcessId;
 use litebox_broker_protocol::message::BrokerNotification;
 use litebox_broker_transport::channel::LocalCallChannel;
 use litebox_platform::time::TimeProvider;
@@ -14,6 +15,7 @@ use crate::{
     broker,
     fd::Descriptors,
     sync::{RawSyncPrimitivesProvider, RwLock},
+    thread::Thread,
 };
 
 /// A full LiteBox system.
@@ -48,16 +50,49 @@ impl<Platform: RawSyncPrimitivesProvider> LiteBox<Platform> {
         platform: &'static Platform,
         broker_local: BrokerLocal<Channel>,
     ) -> Self
+    where
+        Platform: TimeProvider,
+        Channel: LocalCallChannel + Send + Sync + 'static,
+    {
+        Self::new_with_broker_local_inner(platform, broker_local).0
+    }
+
+    fn new_with_broker_local_inner<Channel>(
+        platform: &'static Platform,
+        broker_local: BrokerLocal<Channel>,
+    ) -> (Self, Arc<dyn broker::BrokerControl>)
     where
         Platform: TimeProvider,
         Channel: LocalCallChannel + Send + Sync + 'static,
     {
         let broker_pollables = Arc::new(broker::BrokerPollableRegistry::new());
-        let broker_control = Arc::new(broker::BrokerLocalControl::<Platform, Channel>::new(
-            broker_local,
-            Arc::clone(&broker_pollables),
-        ));
-        Self::new_inner(platform, Some(broker_control), broker_pollables)
+        let broker_control: Arc<dyn broker::BrokerControl> =
+            Arc::new(broker::BrokerLocalControl::<Platform, Channel>::new(
+                broker_local,
+                Arc::clone(&broker_pollables),
+            ));
+        let litebox = Self::new_inner(
+            platform,
+            Some(Arc::clone(&broker_control)),
+            broker_pollables,
+        );
+        (litebox, broker_control)
+    }
+
+    /// Creates a broker-backed process and its negotiated initial thread.
+    pub fn new_process_with_broker_local<Channel>(
+        platform: &'static Platform,
+        broker_local: BrokerLocal<Channel>,
+    ) -> (Self, ProcessId, Thread)
+    where
+        Platform: TimeProvider,
+        Channel: LocalCallChannel + Send + Sync + 'static,
+    {
+        let process_id = broker_local.process_id();
+        let initial_thread_id = broker_local.initial_thread_id();
+        let (litebox, broker) = Self::new_with_broker_local_inner(platform, broker_local);
+        let initial_thread = Thread::from_broker(initial_thread_id, broker);
+        (litebox, process_id, initial_thread)
     }
 
     fn new_inner(

litebox/src/thread.rs

@@ -50,6 +50,10 @@ pub struct Thread {
 }
 
 impl Thread {
+    pub(crate) fn from_broker(id: ThreadId, broker: Arc<dyn BrokerControl>) -> Self {
+        Self { id, broker }
+    }
+
     /// Returns the assigned thread ID.
     #[must_use]
     pub const fn id(&self) -> u32 {

litebox_broker_core/src/lib.rs

@@ -52,6 +52,7 @@ pub use policy::{
 use process::ObjectReference;
 pub use process::{
     AssociationCancellation, BrokerProcess, BrokerThread, CallerCredential, ObjectRights,
+    ProcessLifecycleSink, ProcessShutdown,
 };
 use random::RandomProvider;
 use socket::{BrokerSocketPorts, SocketProvider};
@@ -213,10 +214,17 @@ pub struct BrokerCore {
     pub(crate) socket_provider: Arc<dyn SocketProvider>,
     pub(crate) fs: Arc<dyn FileService>,
     pub(crate) socket_ports: BrokerSocketPorts,
+    pub(crate) process_lifecycle_sink: Arc<dyn ProcessLifecycleSink>,
 }
 
 static BROKER_CORE_CREATED: AtomicBool = AtomicBool::new(false);
 
+struct NoopProcessLifecycleSink;
+
+impl ProcessLifecycleSink for NoopProcessLifecycleSink {
+    fn changed(&self) {}
+}
+
 impl BrokerCore {
     /// Creates the broker core with broker-wide service providers.
     pub fn new(
@@ -266,9 +274,25 @@ impl BrokerCore {
             socket_provider,
             fs,
             socket_ports: BrokerSocketPorts::default(),
+            process_lifecycle_sink: Arc::new(NoopProcessLifecycleSink),
         })
     }
 
+    /// Returns a broker handle that publishes process lifecycle changes to `sink`.
+    #[must_use]
+    pub fn with_process_lifecycle_sink(&self, sink: Arc<dyn ProcessLifecycleSink>) -> Self {
+        Self {
+            process_lifecycle_sink: sink,
+            ..self.clone()
+        }
+    }
+
+    /// Returns whether any broker process remains registered.
+    #[must_use]
+    pub fn has_processes(&self) -> bool {
+        !self.processes.read().is_empty()
+    }
+
     /// Returns the configured authority-state limits.
     #[must_use]
     pub const fn limits(&self) -> BrokerCoreLimits {
@@ -301,7 +325,7 @@ impl BrokerCore {
         Ok((first, second))
     }
 
-    /// Allocates and registers one authenticated broker process.
+    /// Allocates one authenticated process awaiting association activation.
     ///
     /// # Panics
     ///
@@ -310,6 +334,7 @@ impl BrokerCore {
     pub fn create_process(
         &self,
         caller_credential: CallerCredential,
+        parent_id: Option<ProcessId>,
     ) -> Result<Arc<BrokerProcess>> {
         let mut processes = self.processes.write();
         if processes.len() >= self.limits.max_processes {
@@ -323,7 +348,7 @@ impl BrokerCore {
         let process = Arc::new(BrokerProcess::new(
             self.clone(),
             id,
-            None,
+            parent_id,
             caller_credential,
         ));
         assert!(