hyperactor/
ref_.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//! Typed capability references for Hyperactor actors and ports.
10
11use std::cmp::Ordering;
12use std::fmt;
13use std::hash::Hash;
14use std::hash::Hasher;
15use std::marker::PhantomData;
16
17use derivative::Derivative;
18use hyperactor_config::Flattrs;
19use serde::Deserialize;
20use serde::Deserializer;
21use serde::Serialize;
22use serde::Serializer;
23use typeuri::Named;
24
25use crate::Actor;
26use crate::ActorAddr;
27use crate::ActorHandle;
28use crate::Endpoint;
29use crate::EndpointLocation;
30use crate::PortAddr;
31use crate::RemoteEndpoint;
32use crate::RemoteHandles;
33use crate::RemoteMessage;
34use crate::accum::IdleFlushReducerOpts;
35use crate::accum::ReducerSpec;
36use crate::accum::StreamingReducerOpts;
37use crate::actor::Binds;
38use crate::actor::Referable;
39use crate::context;
40use crate::context::MailboxExt;
41use crate::mailbox::DeliveryFailureReport;
42use crate::mailbox::MailboxSenderError;
43use crate::mailbox::MailboxSenderErrorKind;
44use crate::mailbox::PortSink;
45use crate::port::Port;
46
47/// ActorRefs are typed references to actors.
48#[derive(typeuri::Named)]
49pub struct ActorRef<A: Referable> {
50    pub(crate) actor_addr: ActorAddr,
51    // fn() -> A so that the struct remains Send
52    phantom: PhantomData<fn() -> A>,
53}
54
55impl<A: Referable> ActorRef<A> {
56    /// Get the remote port for message type [`M`] for the referenced actor.
57    pub fn port<M: RemoteMessage>(&self) -> PortRef<M>
58    where
59        A: RemoteHandles<M>,
60    {
61        PortRef::attest(self.actor_addr.port_addr(Port::handler::<M>()))
62    }
63
64    /// The caller guarantees that the provided actor ID is also a valid,
65    /// typed reference.  This is usually invoked to provide a guarantee
66    /// that an externally-provided actor ID (e.g., through a command
67    /// line argument) is a valid reference.
68    pub fn attest(actor_addr: ActorAddr) -> Self {
69        Self {
70            actor_addr,
71            phantom: PhantomData,
72        }
73    }
74
75    /// The actor address corresponding with this reference.
76    pub fn actor_addr(&self) -> &ActorAddr {
77        &self.actor_addr
78    }
79
80    /// Convert this actor reference into its corresponding actor address.
81    pub fn into_actor_addr(self) -> ActorAddr {
82        self.actor_addr
83    }
84
85    /// Attempt to downcast this reference into a (local) actor handle.
86    /// This will only succeed when the referenced actor is in the same
87    /// proc as the caller.
88    pub fn downcast_handle(&self, cx: &impl context::Actor) -> Option<ActorHandle<A>>
89    where
90        A: Actor,
91    {
92        cx.instance().proc().resolve_actor_ref(self)
93    }
94}
95
96impl<A, B> From<&ActorRef<A>> for ActorRef<B>
97where
98    A: Actor + Referable,
99    B: Binds<A>,
100{
101    fn from(value: &ActorRef<A>) -> Self {
102        ActorRef::attest(value.actor_addr().clone())
103    }
104}
105
106impl<A, M> Endpoint<M> for &ActorRef<A>
107where
108    A: Referable + RemoteHandles<M>,
109    M: RemoteMessage,
110{
111    fn endpoint_location(&self) -> EndpointLocation {
112        EndpointLocation::Actor(self.actor_addr.clone())
113    }
114
115    fn post<C>(self, cx: &C, message: M)
116    where
117        C: context::Actor,
118    {
119        RemoteEndpoint::post_with_headers(self, cx, Flattrs::new(), message)
120    }
121}
122
123impl<A, M> RemoteEndpoint<M> for &ActorRef<A>
124where
125    A: Referable + RemoteHandles<M>,
126    M: RemoteMessage,
127{
128    fn post_with_headers<C>(self, cx: &C, headers: Flattrs, message: M)
129    where
130        C: context::Actor,
131    {
132        RemoteEndpoint::post_with_headers(&self.port(), cx, headers, message)
133    }
134}
135
136// Implement Serialize manually, without requiring A: Serialize
137impl<A: Referable> Serialize for ActorRef<A> {
138    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
139    where
140        S: Serializer,
141    {
142        // Serialize only the fields that don't depend on A
143        self.actor_addr.serialize(serializer)
144    }
145}
146
147// Implement Deserialize manually, without requiring A: Deserialize
148impl<'de, A: Referable> Deserialize<'de> for ActorRef<A> {
149    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
150    where
151        D: Deserializer<'de>,
152    {
153        let actor_addr = <ActorAddr>::deserialize(deserializer)?;
154        Ok(ActorRef {
155            actor_addr,
156            phantom: PhantomData,
157        })
158    }
159}
160
161// Implement Debug manually, without requiring A: Debug
162impl<A: Referable> fmt::Debug for ActorRef<A> {
163    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
164        f.debug_struct("ActorRef")
165            .field("actor_addr", &self.actor_addr)
166            .field("type", &std::any::type_name::<A>())
167            .finish()
168    }
169}
170
171impl<A: Referable> fmt::Display for ActorRef<A> {
172    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
173        fmt::Display::fmt(&self.actor_addr, f)?;
174        write!(f, "<{}>", std::any::type_name::<A>())
175    }
176}
177
178// We implement Clone manually to avoid imposing A: Clone.
179impl<A: Referable> Clone for ActorRef<A> {
180    fn clone(&self) -> Self {
181        Self {
182            actor_addr: self.actor_addr.clone(),
183            phantom: PhantomData,
184        }
185    }
186}
187
188impl<A: Referable> PartialEq for ActorRef<A> {
189    fn eq(&self, other: &Self) -> bool {
190        self.actor_addr == other.actor_addr
191    }
192}
193
194impl<A: Referable> Eq for ActorRef<A> {}
195
196impl<A: Referable> PartialOrd for ActorRef<A> {
197    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
198        Some(self.cmp(other))
199    }
200}
201
202impl<A: Referable> Ord for ActorRef<A> {
203    fn cmp(&self, other: &Self) -> Ordering {
204        self.actor_addr.cmp(&other.actor_addr)
205    }
206}
207
208impl<A: Referable> Hash for ActorRef<A> {
209    fn hash<H: Hasher>(&self, state: &mut H) {
210        self.actor_addr.hash(state);
211    }
212}
213
214/// A reference to a remote port. All messages passed through
215/// PortRefs will be serialized. PortRefs are always streaming.
216#[derive(Debug, Derivative, typeuri::Named)]
217#[derivative(PartialEq, Eq, PartialOrd, Hash, Ord)]
218pub struct PortRef<M> {
219    port_addr: PortAddr,
220    #[derivative(
221        PartialEq = "ignore",
222        PartialOrd = "ignore",
223        Ord = "ignore",
224        Hash = "ignore"
225    )]
226    reducer_spec: Option<ReducerSpec>,
227    #[derivative(
228        PartialEq = "ignore",
229        PartialOrd = "ignore",
230        Ord = "ignore",
231        Hash = "ignore"
232    )]
233    streaming_opts: StreamingReducerOpts,
234    phantom: PhantomData<M>,
235    return_undeliverable: bool,
236    #[derivative(
237        PartialEq = "ignore",
238        PartialOrd = "ignore",
239        Ord = "ignore",
240        Hash = "ignore"
241    )]
242    unsplit: bool,
243}
244
245#[doc(hidden)]
246#[derive(Debug, Clone, Serialize, Deserialize, typeuri::Named)]
247pub struct PortRefRepr {
248    port_addr: PortAddr,
249    reducer_spec: Option<ReducerSpec>,
250    streaming_opts: StreamingReducerOpts,
251    return_undeliverable: bool,
252    unsplit: bool,
253}
254
255impl PortRefRepr {
256    /// This port's address.
257    pub fn port_addr(&self) -> &PortAddr {
258        &self.port_addr
259    }
260
261    /// The typehash of this port's reducer, if any.
262    pub fn reducer_spec(&self) -> &Option<ReducerSpec> {
263        &self.reducer_spec
264    }
265
266    /// This port's streaming reducer options.
267    pub fn streaming_opts(&self) -> &StreamingReducerOpts {
268        &self.streaming_opts
269    }
270
271    /// Get whether undeliverable messages should be returned to the sender.
272    pub fn get_return_undeliverable(&self) -> bool {
273        self.return_undeliverable
274    }
275
276    /// Whether the port must not be split.
277    pub fn unsplit(&self) -> bool {
278        self.unsplit
279    }
280
281    /// Update this port's address.
282    pub fn update_port_addr(&mut self, port_addr: PortAddr) {
283        self.port_addr = port_addr;
284    }
285}
286
287impl<M> TryFrom<&PortRef<M>> for PortRefRepr {
288    type Error = serde_multipart::Error;
289
290    fn try_from(port_ref: &PortRef<M>) -> serde_multipart::Result<Self> {
291        Ok(Self {
292            port_addr: port_ref.port_addr.clone(),
293            reducer_spec: port_ref.reducer_spec.clone(),
294            streaming_opts: port_ref.streaming_opts.clone(),
295            return_undeliverable: port_ref.return_undeliverable,
296            unsplit: port_ref.unsplit,
297        })
298    }
299}
300
301impl<M> TryFrom<PortRefRepr> for PortRef<M> {
302    type Error = serde_multipart::Error;
303
304    fn try_from(repr: PortRefRepr) -> serde_multipart::Result<Self> {
305        Ok(Self {
306            port_addr: repr.port_addr,
307            reducer_spec: repr.reducer_spec,
308            streaming_opts: repr.streaming_opts,
309            phantom: PhantomData,
310            return_undeliverable: repr.return_undeliverable,
311            unsplit: repr.unsplit,
312        })
313    }
314}
315
316serde_multipart::part_codec! {
317    impl<M> PortRef<M>
318    {
319        type Repr = PortRefRepr;
320    }
321}
322
323impl<M: RemoteMessage> PortRef<M> {
324    /// The caller attests that the provided port address identifies a
325    /// reachable typed port for message type `M`.
326    pub fn attest(port_addr: PortAddr) -> Self {
327        Self {
328            port_addr,
329            reducer_spec: None,
330            streaming_opts: StreamingReducerOpts::default(),
331            phantom: PhantomData,
332            return_undeliverable: true,
333            unsplit: false,
334        }
335    }
336
337    /// The caller attests that the provided port address identifies a
338    /// reachable typed port for message type `M`.
339    pub fn attest_reducible(
340        port_addr: PortAddr,
341        reducer_spec: Option<ReducerSpec>,
342        streaming_opts: StreamingReducerOpts,
343    ) -> Self {
344        Self {
345            port_addr,
346            reducer_spec,
347            streaming_opts,
348            phantom: PhantomData,
349            return_undeliverable: true,
350            unsplit: false,
351        }
352    }
353
354    /// Prevents the port from being split.
355    pub fn unsplit(mut self) -> Self {
356        self.unsplit = true;
357        self
358    }
359
360    /// The caller attests that the provided actor exposes a reachable handler
361    /// port for message type `M`.
362    pub fn attest_handler_port(actor: &ActorAddr) -> Self {
363        PortRef::<M>::attest(actor.port_addr(Port::handler::<M>()))
364    }
365
366    /// The typehash of this port's reducer, if any. Reducers
367    /// may be used to coalesce messages sent to a port.
368    pub fn reducer_spec(&self) -> &Option<ReducerSpec> {
369        &self.reducer_spec
370    }
371
372    /// This port's address.
373    pub fn port_addr(&self) -> &PortAddr {
374        &self.port_addr
375    }
376
377    /// Convert this PortRef into its corresponding port address.
378    pub fn into_port_addr(self) -> PortAddr {
379        self.port_addr
380    }
381
382    /// Convert this streaming reference into an idle-flush reference to the same port.
383    ///
384    /// For example, a reference to port `7` remains addressed to port `7`, while
385    /// `reducer_opts` tells each cast split when to flush its reduced replies.
386    pub fn into_idle_flush(self, reducer_opts: IdleFlushReducerOpts) -> IdleFlushPortRef<M> {
387        IdleFlushPortRef {
388            port_addr: self.port_addr,
389            reducer_spec: self.reducer_spec,
390            streaming_opts: self.streaming_opts,
391            reducer_opts,
392            return_undeliverable: self.return_undeliverable,
393            unsplit: self.unsplit,
394            phantom: PhantomData,
395        }
396    }
397
398    /// coerce it into OncePortRef so we can send messages to this port from
399    /// APIs requires OncePortRef.
400    pub fn into_once(self) -> OncePortRef<M> {
401        let return_undeliverable = self.return_undeliverable;
402        let unsplit = self.unsplit;
403        let mut once = OncePortRef::attest(self.into_port_addr());
404        once.return_undeliverable = return_undeliverable;
405        once.unsplit = unsplit;
406        once
407    }
408
409    /// Post a serialized message to this port, provided a sending capability, such as
410    /// [`crate::actor::Instance`].
411    pub fn post_serialized(
412        &self,
413        cx: &impl context::Actor,
414        mut headers: Flattrs,
415        message: wirevalue::Any,
416    ) {
417        crate::mailbox::headers::set_send_timestamp(&mut headers);
418        crate::mailbox::headers::set_rust_message_type::<M>(&mut headers);
419        cx.post(
420            self.port_addr.clone(),
421            headers,
422            message,
423            self.return_undeliverable,
424            context::SeqInfoPolicy::AssignNew,
425        );
426    }
427
428    /// Convert this port into a sink that can be used to send messages using the given capability.
429    pub fn into_sink<C: context::Actor>(self, cx: C) -> PortSink<C, M> {
430        PortSink::new(cx, self)
431    }
432
433    /// Get whether or not messages sent to this port that are undeliverable should
434    /// be returned to the sender.
435    pub fn get_return_undeliverable(&self) -> bool {
436        self.return_undeliverable
437    }
438
439    /// Set whether or not messages sent to this port that are undeliverable
440    /// should be returned to the sender.
441    pub fn return_undeliverable(&mut self, return_undeliverable: bool) {
442        self.return_undeliverable = return_undeliverable;
443    }
444}
445
446impl<M> Endpoint<M> for &PortRef<M>
447where
448    M: RemoteMessage,
449{
450    fn endpoint_location(&self) -> EndpointLocation {
451        EndpointLocation::Port(self.port_addr.clone())
452    }
453
454    fn post<C>(self, cx: &C, message: M)
455    where
456        C: context::Actor,
457    {
458        RemoteEndpoint::post_with_headers(self, cx, Flattrs::new(), message)
459    }
460}
461
462impl<M> RemoteEndpoint<M> for &PortRef<M>
463where
464    M: RemoteMessage,
465{
466    fn post_with_headers<C>(self, cx: &C, headers: Flattrs, message: M)
467    where
468        C: context::Actor,
469    {
470        let serialized = match wirevalue::Any::serialize(&message).map_err(|err| {
471            MailboxSenderError::new_bound(
472                self.port_addr.clone(),
473                MailboxSenderErrorKind::Serialize(err.into()),
474            )
475        }) {
476            Ok(serialized) => serialized,
477            Err(err) => {
478                cx.instance()
479                    .report_delivery_failure(DeliveryFailureReport::from_send_error::<M>(
480                        cx.mailbox().actor_addr().clone(),
481                        self.endpoint_location(),
482                        &err,
483                    ));
484                return;
485            }
486        };
487        self.post_serialized(cx, headers, serialized);
488    }
489}
490
491impl<M: RemoteMessage> Clone for PortRef<M> {
492    fn clone(&self) -> Self {
493        Self {
494            port_addr: self.port_addr.clone(),
495            reducer_spec: self.reducer_spec.clone(),
496            streaming_opts: self.streaming_opts.clone(),
497            phantom: PhantomData,
498            return_undeliverable: self.return_undeliverable,
499            unsplit: self.unsplit,
500        }
501    }
502}
503
504impl<M: RemoteMessage> fmt::Display for PortRef<M> {
505    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
506        fmt::Display::fmt(&self.port_addr, f)
507    }
508}
509
510/// A reusable remote port reference whose cast-tree proxies flush reduced updates after idling.
511#[derive(Debug, PartialEq, typeuri::Named)]
512pub struct IdleFlushPortRef<M> {
513    port_addr: PortAddr,
514    reducer_spec: Option<ReducerSpec>,
515    streaming_opts: StreamingReducerOpts,
516    reducer_opts: IdleFlushReducerOpts,
517    return_undeliverable: bool,
518    unsplit: bool,
519    phantom: PhantomData<M>,
520}
521
522#[doc(hidden)]
523#[derive(Debug, Clone, Serialize, Deserialize, typeuri::Named)]
524pub struct IdleFlushPortRefRepr {
525    port_addr: PortAddr,
526    reducer_spec: Option<ReducerSpec>,
527    streaming_opts: StreamingReducerOpts,
528    reducer_opts: IdleFlushReducerOpts,
529    return_undeliverable: bool,
530    unsplit: bool,
531}
532
533impl IdleFlushPortRefRepr {
534    /// Return the destination address; for a reference to port `7`, this returns port `7`.
535    pub fn port_addr(&self) -> &PortAddr {
536        &self.port_addr
537    }
538
539    /// Return the reducer specification; a non-reducible port returns `None`.
540    pub fn reducer_spec(&self) -> Option<&ReducerSpec> {
541        self.reducer_spec.as_ref()
542    }
543
544    /// Return the idle-flush reducer options.
545    pub fn reducer_opts(&self) -> &IdleFlushReducerOpts {
546        &self.reducer_opts
547    }
548
549    /// Return the undeliverable policy; a port configured to drop failures returns `false`.
550    pub fn get_return_undeliverable(&self) -> bool {
551        self.return_undeliverable
552    }
553
554    /// Return whether splitting is disabled; an unsplit reference returns `true`.
555    pub fn unsplit(&self) -> bool {
556        self.unsplit
557    }
558
559    /// Replace the destination address; changing port `7` to port `8` makes later sends use `8`.
560    pub fn update_port_addr(&mut self, port_addr: PortAddr) {
561        self.port_addr = port_addr;
562    }
563}
564
565impl<M> TryFrom<&IdleFlushPortRef<M>> for IdleFlushPortRefRepr {
566    type Error = serde_multipart::Error;
567
568    fn try_from(port_ref: &IdleFlushPortRef<M>) -> serde_multipart::Result<Self> {
569        Ok(Self {
570            port_addr: port_ref.port_addr.clone(),
571            reducer_spec: port_ref.reducer_spec.clone(),
572            streaming_opts: port_ref.streaming_opts.clone(),
573            reducer_opts: port_ref.reducer_opts.clone(),
574            return_undeliverable: port_ref.return_undeliverable,
575            unsplit: port_ref.unsplit,
576        })
577    }
578}
579
580impl<M> TryFrom<IdleFlushPortRefRepr> for IdleFlushPortRef<M> {
581    type Error = serde_multipart::Error;
582
583    fn try_from(repr: IdleFlushPortRefRepr) -> serde_multipart::Result<Self> {
584        Ok(Self {
585            port_addr: repr.port_addr,
586            reducer_spec: repr.reducer_spec,
587            streaming_opts: repr.streaming_opts,
588            reducer_opts: repr.reducer_opts,
589            return_undeliverable: repr.return_undeliverable,
590            unsplit: repr.unsplit,
591            phantom: PhantomData,
592        })
593    }
594}
595
596serde_multipart::part_codec! {
597    impl<M> IdleFlushPortRef<M>
598    {
599        type Repr = IdleFlushPortRefRepr;
600    }
601}
602
603impl<M: RemoteMessage> IdleFlushPortRef<M> {
604    /// The typehash of this port's reducer, if any.
605    pub fn reducer_spec(&self) -> &Option<ReducerSpec> {
606        &self.reducer_spec
607    }
608
609    /// This port's address.
610    pub fn port_addr(&self) -> &PortAddr {
611        &self.port_addr
612    }
613
614    /// Convert this IdleFlushPortRef into its corresponding port address.
615    pub fn into_port_addr(self) -> PortAddr {
616        self.port_addr
617    }
618
619    /// Post a serialized message to this port, provided a sending capability, such as
620    /// [`crate::actor::Instance`].
621    pub fn post_serialized(
622        &self,
623        cx: &impl context::Actor,
624        mut headers: Flattrs,
625        message: wirevalue::Any,
626    ) {
627        crate::mailbox::headers::set_send_timestamp(&mut headers);
628        crate::mailbox::headers::set_rust_message_type::<M>(&mut headers);
629        cx.post(
630            self.port_addr.clone(),
631            headers,
632            message,
633            self.return_undeliverable,
634            context::SeqInfoPolicy::AssignNew,
635        );
636    }
637
638    /// Get whether or not messages sent to this port that are undeliverable should
639    /// be returned to the sender.
640    pub fn get_return_undeliverable(&self) -> bool {
641        self.return_undeliverable
642    }
643
644    /// Set whether or not messages sent to this port that are undeliverable
645    /// should be returned to the sender.
646    pub fn return_undeliverable(&mut self, return_undeliverable: bool) {
647        self.return_undeliverable = return_undeliverable;
648    }
649}
650
651impl<M> Endpoint<M> for &IdleFlushPortRef<M>
652where
653    M: RemoteMessage,
654{
655    fn endpoint_location(&self) -> EndpointLocation {
656        EndpointLocation::Port(self.port_addr.clone())
657    }
658
659    fn post<C>(self, cx: &C, message: M)
660    where
661        C: context::Actor,
662    {
663        RemoteEndpoint::post_with_headers(self, cx, Flattrs::new(), message)
664    }
665}
666
667impl<M> RemoteEndpoint<M> for &IdleFlushPortRef<M>
668where
669    M: RemoteMessage,
670{
671    fn post_with_headers<C>(self, cx: &C, headers: Flattrs, message: M)
672    where
673        C: context::Actor,
674    {
675        let serialized = match wirevalue::Any::serialize(&message).map_err(|err| {
676            MailboxSenderError::new_bound(
677                self.port_addr.clone(),
678                MailboxSenderErrorKind::Serialize(err.into()),
679            )
680        }) {
681            Ok(serialized) => serialized,
682            Err(err) => {
683                cx.instance()
684                    .report_delivery_failure(DeliveryFailureReport::from_send_error::<M>(
685                        cx.mailbox().actor_addr().clone(),
686                        self.endpoint_location(),
687                        &err,
688                    ));
689                return;
690            }
691        };
692        self.post_serialized(cx, headers, serialized);
693    }
694}
695
696impl<M: RemoteMessage> Clone for IdleFlushPortRef<M> {
697    fn clone(&self) -> Self {
698        Self {
699            port_addr: self.port_addr.clone(),
700            reducer_spec: self.reducer_spec.clone(),
701            streaming_opts: self.streaming_opts.clone(),
702            reducer_opts: self.reducer_opts.clone(),
703            return_undeliverable: self.return_undeliverable,
704            unsplit: self.unsplit,
705            phantom: PhantomData,
706        }
707    }
708}
709
710/// A remote reference to a [`OncePort`]. References are serializable
711/// and may be passed to remote actors, which can then use it to send
712/// a message to this port.
713#[derive(Debug, PartialEq)]
714pub struct OncePortRef<M> {
715    port_addr: PortAddr,
716    reducer_spec: Option<ReducerSpec>,
717    return_undeliverable: bool,
718    unsplit: bool,
719    phantom: PhantomData<M>,
720}
721
722#[doc(hidden)]
723#[derive(Debug, Clone, Serialize, Deserialize, typeuri::Named)]
724pub struct OncePortRefRepr {
725    port_addr: PortAddr,
726    reducer_spec: Option<ReducerSpec>,
727    return_undeliverable: bool,
728    unsplit: bool,
729}
730
731impl OncePortRefRepr {
732    /// This port's address.
733    pub fn port_addr(&self) -> &PortAddr {
734        &self.port_addr
735    }
736
737    /// The typehash of this port's reducer, if any.
738    pub fn reducer_spec(&self) -> &Option<ReducerSpec> {
739        &self.reducer_spec
740    }
741
742    /// Get whether undeliverable messages should be returned to the sender.
743    pub fn get_return_undeliverable(&self) -> bool {
744        self.return_undeliverable
745    }
746
747    /// Whether the port must not be split.
748    pub fn unsplit(&self) -> bool {
749        self.unsplit
750    }
751
752    /// Update this port's address.
753    pub fn update_port_addr(&mut self, port_addr: PortAddr) {
754        self.port_addr = port_addr;
755    }
756}
757
758impl<M> TryFrom<&OncePortRef<M>> for OncePortRefRepr {
759    type Error = serde_multipart::Error;
760
761    fn try_from(port_ref: &OncePortRef<M>) -> serde_multipart::Result<Self> {
762        Ok(Self {
763            port_addr: port_ref.port_addr.clone(),
764            reducer_spec: port_ref.reducer_spec.clone(),
765            return_undeliverable: port_ref.return_undeliverable,
766            unsplit: port_ref.unsplit,
767        })
768    }
769}
770
771impl<M> TryFrom<OncePortRefRepr> for OncePortRef<M> {
772    type Error = serde_multipart::Error;
773
774    fn try_from(repr: OncePortRefRepr) -> serde_multipart::Result<Self> {
775        Ok(Self {
776            port_addr: repr.port_addr,
777            reducer_spec: repr.reducer_spec,
778            return_undeliverable: repr.return_undeliverable,
779            unsplit: repr.unsplit,
780            phantom: PhantomData,
781        })
782    }
783}
784
785serde_multipart::part_codec! {
786    impl<M> OncePortRef<M>
787    {
788        type Repr = OncePortRefRepr;
789    }
790}
791
792impl<M: RemoteMessage> OncePortRef<M> {
793    pub(crate) fn attest(port_addr: PortAddr) -> Self {
794        Self {
795            port_addr,
796            reducer_spec: None,
797            return_undeliverable: true,
798            unsplit: false,
799            phantom: PhantomData,
800        }
801    }
802
803    /// The caller attests that the provided PortId can be
804    /// converted to a reachable, typed once port reference.
805    pub fn attest_reducible(port_addr: PortAddr, reducer_spec: Option<ReducerSpec>) -> Self {
806        Self {
807            port_addr,
808            reducer_spec,
809            return_undeliverable: true,
810            unsplit: false,
811            phantom: PhantomData,
812        }
813    }
814
815    /// Prevents the port from being split.
816    pub fn unsplit(mut self) -> Self {
817        self.unsplit = true;
818        self
819    }
820
821    /// The typehash of this port's reducer, if any.
822    pub fn reducer_spec(&self) -> &Option<ReducerSpec> {
823        &self.reducer_spec
824    }
825
826    /// This port's address.
827    pub fn port_addr(&self) -> &PortAddr {
828        &self.port_addr
829    }
830
831    /// Convert this OncePortRef into its corresponding port address.
832    pub fn into_port_addr(self) -> PortAddr {
833        self.port_addr
834    }
835
836    /// Get whether or not messages sent to this port that are undeliverable should
837    /// be returned to the sender.
838    pub fn get_return_undeliverable(&self) -> bool {
839        self.return_undeliverable
840    }
841
842    /// Set whether or not messages sent to this port that are undeliverable
843    /// should be returned to the sender.
844    pub fn return_undeliverable(&mut self, return_undeliverable: bool) {
845        self.return_undeliverable = return_undeliverable;
846    }
847}
848
849impl<M> Endpoint<M> for OncePortRef<M>
850where
851    M: RemoteMessage,
852{
853    fn endpoint_location(&self) -> EndpointLocation {
854        EndpointLocation::Port(self.port_addr.clone())
855    }
856
857    fn post<C>(self, cx: &C, message: M)
858    where
859        C: context::Actor,
860    {
861        RemoteEndpoint::post_with_headers(self, cx, Flattrs::new(), message)
862    }
863}
864
865impl<M> RemoteEndpoint<M> for OncePortRef<M>
866where
867    M: RemoteMessage,
868{
869    fn post_with_headers<C>(self, cx: &C, mut headers: Flattrs, message: M)
870    where
871        C: context::Actor,
872    {
873        crate::mailbox::headers::set_send_timestamp(&mut headers);
874        let serialized = match wirevalue::Any::serialize(&message).map_err(|err| {
875            MailboxSenderError::new_bound(
876                self.port_addr.clone(),
877                MailboxSenderErrorKind::Serialize(err.into()),
878            )
879        }) {
880            Ok(serialized) => serialized,
881            Err(err) => {
882                cx.instance()
883                    .report_delivery_failure(DeliveryFailureReport::from_send_error::<M>(
884                        cx.mailbox().actor_addr().clone(),
885                        self.endpoint_location(),
886                        &err,
887                    ));
888                return;
889            }
890        };
891        cx.post(
892            self.port_addr.clone(),
893            headers,
894            serialized,
895            self.return_undeliverable,
896            context::SeqInfoPolicy::AssignNew,
897        );
898    }
899}
900
901impl<M: RemoteMessage> Clone for OncePortRef<M> {
902    fn clone(&self) -> Self {
903        Self {
904            port_addr: self.port_addr.clone(),
905            reducer_spec: self.reducer_spec.clone(),
906            return_undeliverable: self.return_undeliverable,
907            unsplit: self.unsplit,
908            phantom: PhantomData,
909        }
910    }
911}
912
913impl<M: RemoteMessage> fmt::Display for OncePortRef<M> {
914    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
915        fmt::Display::fmt(&self.port_addr, f)
916    }
917}
918
919impl<M: RemoteMessage> Named for OncePortRef<M> {
920    fn typename() -> &'static str {
921        wirevalue::intern_typename!(Self, "hyperactor::mailbox::OncePortRef<{}>", M)
922    }
923}
924
925#[cfg(test)]
926mod tests {
927    use std::time::Duration;
928
929    use hyperactor_config::NonZeroUsize;
930    use serde::Deserialize;
931    use serde::Serialize;
932    use serde_multipart::PartCodec;
933    use typeuri::Named;
934
935    use super::*;
936    use crate::channel::ChannelAddr;
937    use crate::id::ActorId;
938    use crate::id::Label;
939    use crate::id::PortId;
940    use crate::id::ProcId;
941
942    fn test_port_ref() -> PortRef<String> {
943        let proc_id = ProcId::singleton(Label::new("proc").unwrap());
944        let actor_id = ActorId::singleton(Label::new("actor").unwrap(), proc_id);
945        let port_id = PortId::new(actor_id, Port::from(7));
946        let port_addr = PortAddr::new(port_id, ChannelAddr::Local(42).into());
947        let mut port_ref = PortRef::<String>::attest(port_addr).unsplit();
948        port_ref.return_undeliverable(false);
949        port_ref
950    }
951
952    fn test_once_port_ref() -> OncePortRef<String> {
953        let proc_id = ProcId::singleton(Label::new("proc").unwrap());
954        let actor_id = ActorId::singleton(Label::new("actor").unwrap(), proc_id);
955        let port_id = PortId::new(actor_id, Port::from(8));
956        let port_addr = PortAddr::new(port_id, ChannelAddr::Local(43).into());
957        let mut once_port_ref = OncePortRef::<String>::attest(port_addr).unsplit();
958        once_port_ref.return_undeliverable(false);
959        once_port_ref
960    }
961
962    fn test_idle_flush_port_ref() -> IdleFlushPortRef<String> {
963        let proc_id = ProcId::singleton(Label::new("proc").unwrap());
964        let actor_id = ActorId::singleton(Label::new("actor").unwrap(), proc_id);
965        let port_id = PortId::new(actor_id, Port::from(9));
966        let port_addr = PortAddr::new(port_id, ChannelAddr::Local(44).into());
967        let mut port_ref = PortRef::<String>::attest_reducible(
968            port_addr,
969            Some(ReducerSpec {
970                typehash: 123,
971                builder_params: None,
972            }),
973            StreamingReducerOpts {
974                max_update_interval: Some(Duration::from_millis(250)),
975                initial_update_interval: Some(Duration::from_millis(5)),
976            },
977        )
978        .unsplit()
979        .into_idle_flush(IdleFlushReducerOpts {
980            idle_timeout: Duration::from_millis(50),
981            abandon_timeout: Duration::from_secs(30),
982            expected_updates_per_destination: NonZeroUsize::new(2).expect("2 is non-zero"),
983        });
984        port_ref.return_undeliverable(false);
985        port_ref
986    }
987
988    #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
989    struct PortRefEnvelope {
990        port: PortRef<String>,
991        seq: u64,
992    }
993
994    #[derive(Debug, PartialEq, Serialize, Deserialize)]
995    struct IdleFlushPortRefEnvelope {
996        port: IdleFlushPortRef<String>,
997        seq: u64,
998    }
999
1000    #[derive(Debug, PartialEq, Serialize, Deserialize)]
1001    struct OncePortRefEnvelope {
1002        port: OncePortRef<String>,
1003        seq: u64,
1004    }
1005
1006    fn assert_same_port_ref(actual: &PortRef<String>, expected: &PortRef<String>) {
1007        let actual = actual.to_repr().unwrap();
1008        let expected = expected.to_repr().unwrap();
1009        assert_eq!(actual.port_addr, expected.port_addr);
1010        assert_eq!(actual.reducer_spec, expected.reducer_spec);
1011        assert_eq!(actual.streaming_opts, expected.streaming_opts);
1012        assert_eq!(actual.return_undeliverable, expected.return_undeliverable);
1013        assert_eq!(actual.unsplit, expected.unsplit);
1014    }
1015
1016    fn assert_same_once_port_ref(actual: &OncePortRef<String>, expected: &OncePortRef<String>) {
1017        let actual = actual.to_repr().unwrap();
1018        let expected = expected.to_repr().unwrap();
1019        assert_eq!(actual.port_addr, expected.port_addr);
1020        assert_eq!(actual.reducer_spec, expected.reducer_spec);
1021        assert_eq!(actual.return_undeliverable, expected.return_undeliverable);
1022        assert_eq!(actual.unsplit, expected.unsplit);
1023    }
1024
1025    fn assert_same_idle_flush_port_ref(
1026        actual: &IdleFlushPortRef<String>,
1027        expected: &IdleFlushPortRef<String>,
1028    ) {
1029        let actual = actual.to_repr().unwrap();
1030        let expected = expected.to_repr().unwrap();
1031        assert_eq!(actual.port_addr, expected.port_addr);
1032        assert_eq!(actual.reducer_spec, expected.reducer_spec);
1033        assert_eq!(actual.streaming_opts, expected.streaming_opts);
1034        assert_eq!(actual.reducer_opts, expected.reducer_opts);
1035        assert_eq!(actual.return_undeliverable, expected.return_undeliverable);
1036        assert_eq!(actual.unsplit, expected.unsplit);
1037    }
1038
1039    #[test]
1040    fn test_port_ref_serde_multipart_part_codec() {
1041        let value = PortRefEnvelope {
1042            port: test_port_ref(),
1043            seq: 123,
1044        };
1045
1046        let message = serde_multipart::serialize_bincode(&value).unwrap();
1047        assert_eq!(message.num_parts(), 1);
1048        assert_eq!(message.parts()[0].typehash(), Some(PortRefRepr::typehash()));
1049
1050        let repr = message.parts()[0].deserialized::<PortRefRepr>().unwrap();
1051        assert_eq!(repr.port_addr, value.port.port_addr().clone());
1052        assert!(!repr.return_undeliverable);
1053        assert!(repr.unsplit);
1054
1055        let deserialized: PortRefEnvelope = serde_multipart::deserialize_bincode(message).unwrap();
1056        assert_eq!(deserialized.seq, value.seq);
1057        assert_same_port_ref(&deserialized.port, &value.port);
1058    }
1059
1060    #[test]
1061    fn test_idle_flush_port_ref_serde_multipart_part_codec() {
1062        let value = IdleFlushPortRefEnvelope {
1063            port: test_idle_flush_port_ref(),
1064            seq: 789,
1065        };
1066
1067        let message = serde_multipart::serialize_bincode(&value).unwrap();
1068        assert_eq!(message.num_parts(), 1);
1069        assert_eq!(
1070            message.parts()[0].typehash(),
1071            Some(IdleFlushPortRefRepr::typehash())
1072        );
1073
1074        let repr = message.parts()[0]
1075            .deserialized::<IdleFlushPortRefRepr>()
1076            .unwrap();
1077        let expected = value.port.to_repr().unwrap();
1078        assert_eq!(repr.port_addr, expected.port_addr);
1079        assert_eq!(repr.reducer_spec, expected.reducer_spec);
1080        assert_eq!(repr.streaming_opts, expected.streaming_opts);
1081        assert_eq!(repr.reducer_opts, expected.reducer_opts);
1082        assert_eq!(repr.return_undeliverable, expected.return_undeliverable);
1083        assert_eq!(repr.unsplit, expected.unsplit);
1084
1085        let deserialized: IdleFlushPortRefEnvelope =
1086            serde_multipart::deserialize_bincode(message).unwrap();
1087        assert_eq!(deserialized.seq, value.seq);
1088        assert_same_idle_flush_port_ref(&deserialized.port, &value.port);
1089    }
1090
1091    #[test]
1092    fn test_once_port_ref_serde_multipart_part_codec() {
1093        let value = OncePortRefEnvelope {
1094            port: test_once_port_ref(),
1095            seq: 789,
1096        };
1097
1098        let message = serde_multipart::serialize_bincode(&value).unwrap();
1099        assert_eq!(message.num_parts(), 1);
1100        assert_eq!(
1101            message.parts()[0].typehash(),
1102            Some(OncePortRefRepr::typehash())
1103        );
1104
1105        let repr = message.parts()[0]
1106            .deserialized::<OncePortRefRepr>()
1107            .unwrap();
1108        assert_eq!(repr.port_addr, value.port.port_addr().clone());
1109        assert!(!repr.return_undeliverable);
1110        assert!(repr.unsplit);
1111
1112        let deserialized: OncePortRefEnvelope =
1113            serde_multipart::deserialize_bincode(message).unwrap();
1114        assert_eq!(deserialized.seq, value.seq);
1115        assert_same_once_port_ref(&deserialized.port, &value.port);
1116    }
1117
1118    #[test]
1119    fn test_port_ref_regular_serde_uses_repr() {
1120        let value = PortRefEnvelope {
1121            port: test_port_ref(),
1122            seq: 456,
1123        };
1124
1125        let encoded = bincode::serde::encode_to_vec(&value, bincode::config::standard()).unwrap();
1126        let (deserialized, len): (PortRefEnvelope, usize) =
1127            bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap();
1128
1129        assert_eq!(len, encoded.len());
1130        assert_eq!(deserialized.seq, value.seq);
1131        assert_same_port_ref(&deserialized.port, &value.port);
1132    }
1133
1134    #[test]
1135    fn test_once_port_ref_regular_serde_uses_repr() {
1136        let value = OncePortRefEnvelope {
1137            port: test_once_port_ref(),
1138            seq: 1011,
1139        };
1140
1141        let encoded = bincode::serde::encode_to_vec(&value, bincode::config::standard()).unwrap();
1142        let (deserialized, len): (OncePortRefEnvelope, usize) =
1143            bincode::serde::decode_from_slice(&encoded, bincode::config::standard()).unwrap();
1144
1145        assert_eq!(len, encoded.len());
1146        assert_eq!(deserialized.seq, value.seq);
1147        assert_same_once_port_ref(&deserialized.port, &value.port);
1148    }
1149}
hyperactor/ref_.rs

hyperactor/
ref_.rs
