compio_actor/cluster/
spawn.rs

1use std::{
2    borrow::Cow,
3    error::Error,
4    fmt,
5    future::{Future, IntoFuture},
6    num::NonZeroUsize,
7    pin::Pin,
8    task::{Context, Poll, ready},
9};
10
11use futures_channel::oneshot;
12use futures_util::FutureExt;
13
14use super::Cluster;
15use crate::{
16    Actor, Handler, Mailbox,
17    actor::{ActorExit, ActorHandle, finish, run},
18    mailbox::{DEFAULT_MAILBOX_CAPACITY, Name, make_mailbox},
19    supervisor::{Supervision, SupervisionEvent},
20};
21
22/// The result returned when an actor starts successfully.
23pub type SpawnResult<A> =
24    Result<(Mailbox<A>, ActorHandle<<A as Actor>::Error>), SpawnError<<A as Actor>::Error>>;
25
26/// An error encountered while starting an actor.
27#[derive(Debug, PartialEq, Eq)]
28pub enum SpawnError<E: Send + 'static> {
29    /// The cluster is no longer accepting actors.
30    Unavailable,
31    /// Another actor is registered under this name.
32    NameTaken(Cow<'static, str>),
33    /// The actor's startup hook failed.
34    Start(E),
35    /// The worker stopped before startup completed.
36    WorkerStopped,
37}
38
39impl<E: Send + 'static> fmt::Display for SpawnError<E> {
40    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
41        match self {
42            Self::Unavailable => f.write_str("actor cluster is unavailable"),
43            Self::NameTaken(name) => write!(f, "actor name {name:?} is already registered"),
44            Self::Start(_) => f.write_str("actor startup failed"),
45            Self::WorkerStopped => f.write_str("actor worker stopped during startup"),
46        }
47    }
48}
49
50impl<E: Error + Send + 'static> Error for SpawnError<E> {}
51
52/// A configurable actor spawn operation.
53///
54/// Returned by [`Cluster::spawn`].
55#[must_use = "actors are not spawned until this builder is awaited"]
56pub struct Spawn<'a, A, F>
57where
58    A: Actor,
59    F: FnOnce() -> A + Send + 'static,
60{
61    cluster: &'a Cluster,
62    factory: F,
63    arguments: A::Arguments,
64    name: Option<Cow<'static, str>>,
65    capacity: NonZeroUsize,
66    supervisor: Option<Supervision<A>>,
67}
68
69impl<'a, A, F> Spawn<'a, A, F>
70where
71    A: Actor,
72    F: FnOnce() -> A + Send + 'static,
73{
74    pub(super) fn new(cluster: &'a Cluster, factory: F, arguments: A::Arguments) -> Self {
75        Self {
76            cluster,
77            factory,
78            arguments,
79            name: None,
80            capacity: DEFAULT_MAILBOX_CAPACITY,
81            supervisor: None,
82        }
83    }
84
85    /// Registers the actor under `name` after startup succeeds.
86    pub fn with_name(mut self, name: impl Into<Cow<'static, str>>) -> Self {
87        self.name = Some(name.into());
88        self
89    }
90
91    /// Sets the actor's bounded mailbox capacity.
92    pub fn with_capacity(mut self, capacity: NonZeroUsize) -> Self {
93        self.capacity = capacity;
94        self
95    }
96
97    /// Sends actor lifecycle events to `supervisor`.
98    pub fn with_supervisor<S>(mut self, supervisor: &Mailbox<S>) -> Self
99    where
100        S: Handler<SupervisionEvent<A>>,
101    {
102        self.supervisor = Some(Supervision::new(supervisor));
103        self
104    }
105}
106
107impl<A, F> IntoFuture for Spawn<'_, A, F>
108where
109    A: Actor,
110    F: FnOnce() -> A + Send + 'static,
111{
112    type IntoFuture = SpawnFuture<A>;
113    type Output = SpawnResult<A>;
114
115    fn into_future(self) -> Self::IntoFuture {
116        self.cluster.start(
117            self.factory,
118            self.arguments,
119            self.name,
120            self.capacity,
121            self.supervisor,
122        )
123    }
124}
125
126/// The future produced by [`Spawn`].
127pub enum SpawnFuture<A: Actor> {
128    #[doc(hidden)]
129    Ready(Option<SpawnResult<A>>),
130    #[doc(hidden)]
131    Pending {
132        mailbox: Mailbox<A>,
133        result: oneshot::Receiver<Result<ActorExit<A::Error>, ()>>,
134        started: oneshot::Receiver<Result<(), A::Error>>,
135    },
136}
137
138impl<A: Actor> SpawnFuture<A> {
139    fn ready(result: SpawnResult<A>) -> Self {
140        Self::Ready(Some(result))
141    }
142
143    fn pending(
144        mailbox: Mailbox<A>,
145        result: oneshot::Receiver<Result<ActorExit<A::Error>, ()>>,
146        started: oneshot::Receiver<Result<(), A::Error>>,
147    ) -> Self {
148        Self::Pending {
149            mailbox,
150            result,
151            started,
152        }
153    }
154}
155
156impl<A: Actor> Future for SpawnFuture<A> {
157    type Output = SpawnResult<A>;
158
159    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
160        let this = self.get_mut();
161        let started = match this {
162            Self::Ready(result) => {
163                return Poll::Ready(result.take().expect("spawn future polled after completion"));
164            }
165            Self::Pending { started, .. } => ready!(started.poll_unpin(cx)),
166        };
167        let (mailbox, result) = match std::mem::replace(this, Self::Ready(None)) {
168            Self::Pending {
169                mailbox, result, ..
170            } => (mailbox, result),
171            Self::Ready(_) => unreachable!(),
172        };
173        Poll::Ready(match started {
174            Ok(Ok(())) => Ok((mailbox, ActorHandle { result })),
175            Ok(Err(error)) => Err(SpawnError::Start(error)),
176            Err(_) => Err(SpawnError::WorkerStopped),
177        })
178    }
179}
180
181impl<A: Actor> Unpin for SpawnFuture<A> {}
182
183impl Cluster {
184    fn start<A, F>(
185        &self,
186        factory: F,
187        arguments: A::Arguments,
188        name: Option<Cow<'static, str>>,
189        capacity: NonZeroUsize,
190        supervisor: Option<Supervision<A>>,
191    ) -> SpawnFuture<A>
192    where
193        A: Actor,
194        F: FnOnce() -> A + Send + 'static,
195    {
196        let (name, reg) = match name {
197            Some(name) => {
198                let name = Name::from(name);
199                match self.inner.registry.reserve(name.clone()) {
200                    Ok(registration) => (Some(name), Some(registration)),
201                    Err(name) => {
202                        return SpawnFuture::ready(Err(SpawnError::NameTaken(name.into_cow())));
203                    }
204                }
205            }
206            None => (None, None),
207        };
208
209        let (mailbox, receiver) = make_mailbox::<A>(name, capacity);
210        let actor_ref = mailbox.clone();
211        let (started_tx, started_rx) = oneshot::channel();
212        let cluster = self.clone();
213        let result = {
214            let dispatcher = self.inner.dispatcher.lock().unwrap();
215            let Some(dispatcher) = dispatcher.as_ref() else {
216                return SpawnFuture::ready(Err(SpawnError::Unavailable));
217            };
218            dispatcher.dispatch(move || {
219                cluster.drive(async move {
220                    let mut reg = reg;
221                    let actor = factory();
222                    let mut state = match actor.pre_start(&actor_ref, arguments).await {
223                        Ok(state) => state,
224                        Err(error) => {
225                            reg.take();
226                            started_tx.send(Err(error)).ok();
227                            return Err(());
228                        }
229                    };
230                    if let Some(registration) = &reg {
231                        registration.activate(&actor_ref);
232                    }
233
234                    if started_tx.send(Ok(())).is_err() {
235                        let exit =
236                            finish(&actor, &actor_ref, receiver, &mut state, ActorExit::Stopped)
237                                .await;
238                        return Ok(exit);
239                    }
240
241                    let exit = run(
242                        actor,
243                        actor_ref.clone(),
244                        receiver,
245                        state,
246                        supervisor.as_ref(),
247                    )
248                    .await;
249                    drop(reg);
250                    if let Some(supervisor) = supervisor {
251                        match &exit {
252                            ActorExit::Stopped => supervisor.terminated(&actor_ref),
253                            ActorExit::Failed(_) => supervisor.failed(&actor_ref),
254                        }
255                    }
256                    Ok(exit)
257                })
258            })
259        };
260        let result = match result {
261            Ok(result) => result,
262            Err(_) => return SpawnFuture::ready(Err(SpawnError::Unavailable)),
263        };
264
265        SpawnFuture::pending(mailbox, result, started_rx)
266    }
267}
compio_actor/cluster/spawn.rs

compio_actor/cluster/
spawn.rs
