hyperactor_remote/
proc_spawner.rs

1/*
2 * Copyright (c) Meta Platforms, Inc. and affiliates.
3 * All rights reserved.
4 *
5 * This source code is licensed under the BSD-style license found in the
6 * LICENSE file in the root directory of this source tree.
7 */
8
9//! Remote proc spawning protocol.
10//!
11//! [`ProcSpawnerEndpoint`] is the proc-level counterpart to
12//! [`crate::ActorSpawnerEndpoint`]. A proc-spawner endpoint accepts [`SpawnProc`],
13//! starts a new proc, creates a proc-local [`SpawnActor`] endpoint, and links that
14//! actor-spawn endpoint back to the caller through the same remote supervision
15//! machinery used for remote actor spawns.
16
17use hyperactor::ActorRef;
18use hyperactor::Endpoint;
19use hyperactor::Label;
20use hyperactor::Location;
21use hyperactor::OncePortHandle;
22use hyperactor::ProcAddr;
23use hyperactor::ProcId;
24use hyperactor::Uid;
25use hyperactor::context;
26
27use crate::KeepaliveLink;
28use crate::SpawnProc as SpawnProcMessage;
29use crate::SupervisionOptions;
30use crate::Supervisor;
31use crate::actor_spawner::SpawnActor;
32
33pub mod unix;
34
35// Behavior implemented by actors that expose proc spawning.
36hyperactor::behavior!(SpawnProc, SpawnProcMessage);
37
38/// Convenience methods for endpoints that accept [`SpawnProc`] requests.
39pub trait ProcSpawnerEndpoint {
40    /// Spawn a proc with a fresh proc uid.
41    fn spawn_proc(&self, cx: &impl context::Actor) -> anyhow::Result<ActorRef<SpawnActor>>
42    where
43        Self: Clone + Send + 'static,
44        for<'a> &'a Self: Endpoint<SpawnProcMessage>,
45    {
46        self.spawn_proc_uid(cx, Uid::anonymous())
47    }
48
49    /// Spawn a proc with an explicit proc uid.
50    fn spawn_proc_uid(
51        &self,
52        cx: &impl context::Actor,
53        uid: Uid,
54    ) -> anyhow::Result<ActorRef<SpawnActor>>
55    where
56        Self: Clone + Send + 'static,
57        for<'a> &'a Self: Endpoint<SpawnProcMessage>,
58    {
59        self.spawn_proc_uid_with_link(cx, uid, KeepaliveLink::default())
60    }
61
62    /// Spawn a proc and report when it becomes reachable.
63    ///
64    /// The returned [`ActorRef`] names the proc's actor-spawn endpoint, but the
65    /// proc has to launch and join before that endpoint can route — messages
66    /// sent earlier are dropped. A bare readiness signal is posted to `ready`
67    /// once the proc has joined, so callers that need to use the endpoint can
68    /// wait for that signal instead of guessing with a delay. The endpoint
69    /// address is already known from the synchronously returned [`ActorRef`], so
70    /// `ready` carries no payload.
71    fn spawn_proc_uid_with_ready(
72        &self,
73        cx: &impl context::Actor,
74        uid: Uid,
75        ready: OncePortHandle<()>,
76    ) -> anyhow::Result<ActorRef<SpawnActor>>
77    where
78        Self: Clone + Send + 'static,
79        for<'a> &'a Self: Endpoint<SpawnProcMessage>,
80    {
81        self.spawn_proc_uid_with_link_and_ready(cx, uid, KeepaliveLink::default(), Some(ready))
82    }
83
84    /// Spawn a proc with an explicit proc uid and liveness link.
85    fn spawn_proc_uid_with_link(
86        &self,
87        cx: &impl context::Actor,
88        uid: Uid,
89        liveness: KeepaliveLink,
90    ) -> anyhow::Result<ActorRef<SpawnActor>>
91    where
92        Self: Clone + Send + 'static,
93        for<'a> &'a Self: Endpoint<SpawnProcMessage>,
94    {
95        self.spawn_proc_uid_with_link_and_ready(cx, uid, liveness, None)
96    }
97
98    /// Spawn a proc with an explicit proc uid, liveness link, and optional
99    /// readiness port. This is the general form behind the other `spawn_proc*`
100    /// methods; see [`spawn_proc_uid_with_ready`](Self::spawn_proc_uid_with_ready)
101    /// for the readiness semantics.
102    fn spawn_proc_uid_with_link_and_ready(
103        &self,
104        cx: &impl context::Actor,
105        uid: Uid,
106        liveness: KeepaliveLink,
107        ready: Option<OncePortHandle<()>>,
108    ) -> anyhow::Result<ActorRef<SpawnActor>>
109    where
110        Self: Clone + Send + 'static,
111        for<'a> &'a Self: Endpoint<SpawnProcMessage>,
112    {
113        anyhow::ensure!(uid.is_instance(), "spawned procs cannot be singletons");
114
115        let actor_spawner = spawned_actor_spawner(self, &uid);
116        let proc_spawner = self.clone();
117        cx.instance().spawn(Supervisor::bootstrap_uid(
118            liveness,
119            SupervisionOptions::default(),
120            Uid::anonymous(),
121            actor_spawner.actor_addr().clone(),
122            ready,
123            move |cx, supervise| {
124                proc_spawner.post(cx, SpawnProcMessage { uid, supervise });
125                Ok(())
126            },
127        ));
128        Ok(actor_spawner)
129    }
130}
131
132impl<T> ProcSpawnerEndpoint for T where for<'a> &'a T: Endpoint<SpawnProcMessage> {}
133
134/// Well-known singleton name of a proc's actor-spawn endpoint.
135pub(crate) const ACTOR_SPAWNER_NAME: &str = "spawner";
136
137/// The singleton uid of a proc's actor-spawn endpoint. The spawner
138/// attests the endpoint ref at this uid, and the child proc spawns its
139/// `ActorSpawner` under it, so the two agree without a round trip.
140pub fn actor_spawner_uid() -> Uid {
141    Uid::singleton(Label::strip(ACTOR_SPAWNER_NAME))
142}
143
144pub(crate) fn actor_spawner_ref(proc_addr: &ProcAddr) -> ActorRef<SpawnActor> {
145    ActorRef::attest(proc_addr.actor_addr_uid(actor_spawner_uid()))
146}
147
148fn spawned_proc_addr(spawner: impl Endpoint<SpawnProcMessage>, uid: &Uid) -> ProcAddr {
149    let spawner_addr = spawner.endpoint_location().actor_addr().proc_addr();
150    spawned_proc_addr_for_spawner_addr(&spawner_addr, uid)
151}
152
153pub(crate) fn spawned_proc_addr_for_spawner_addr(spawner: &ProcAddr, uid: &Uid) -> ProcAddr {
154    let location = append_inner_via(spawner.location().clone(), uid.clone());
155    ProcAddr::new(ProcId::new(uid.clone(), None), location)
156}
157
158fn spawned_actor_spawner(
159    spawner: impl Endpoint<SpawnProcMessage>,
160    uid: &Uid,
161) -> ActorRef<SpawnActor> {
162    actor_spawner_ref(&spawned_proc_addr(spawner, uid))
163}
164
165fn append_inner_via(location: Location, uid: Uid) -> Location {
166    match location {
167        Location::Addr(addr) => Location::Addr(addr).with_via(uid),
168        Location::Via(via, inner) => Location::Via(via, Box::new(append_inner_via(*inner, uid))),
169    }
170}
171
172#[cfg(test)]
173mod tests {
174    use hyperactor::Label;
175    use hyperactor::channel::ChannelAddr;
176
177    use super::*;
178
179    #[test]
180    fn spawned_proc_addr_preserves_spawner_via_route() {
181        let spawner_uid = Uid::instance(Label::new("spawner").unwrap());
182        let outer = Uid::instance(Label::new("outer").unwrap());
183        let inner = Uid::instance(Label::new("inner").unwrap());
184        let child = Uid::instance(Label::new("child").unwrap());
185        let terminal = ChannelAddr::Local(123);
186        let spawner = ProcAddr::new(
187            ProcId::new(spawner_uid, None),
188            Location::from(terminal.clone())
189                .with_via(inner.clone())
190                .with_via(outer.clone()),
191        );
192
193        let spawned = spawned_proc_addr_for_spawner_addr(&spawner, &child);
194
195        let (outer_uid, location) = spawned
196            .location()
197            .as_via()
198            .expect("outer via must be preserved");
199        assert_eq!(outer_uid, &outer);
200        let (inner_uid, location) = location
201            .as_via()
202            .expect("inner via must be preserved before child");
203        assert_eq!(inner_uid, &inner);
204        let (child_uid, location) = location
205            .as_via()
206            .expect("child via must be appended inside spawner route");
207        assert_eq!(child_uid, &child);
208        assert_eq!(location.as_addr(), Some(&terminal));
209    }
210}
hyperactor_remote/proc_spawner.rs

hyperactor_remote/
proc_spawner.rs
