acktor_ipc/
remote_address.rs

1use std::fmt::{self, Debug};
2use std::future;
3use std::hash::Hash;
4use std::sync::Arc;
5use std::time::Duration;
6
7use bytes::Bytes;
8use futures_util::{FutureExt, TryFutureExt};
9use tracing::Instrument;
10
11use acktor::{
12    Address, Message, Recipient, SendError, Sender, SenderId,
13    address::{ClosedResultFuture, DoSendResult, DoSendResultFuture, SendResult, SendResultFuture},
14    channel::oneshot,
15};
16
17use crate::codec::{Decode, DecodeContext, Encode, EncodeContext};
18use crate::remote_actor::RemoteActorRegistry;
19use crate::remote_message::RemoteMessage;
20use crate::session::Session;
21
22/// A type which is used to send messages to a remote actor.
23///
24/// [`Sender`] trait is implemented for this type, so it can be converted into a [`Recipient`]
25/// type easily. This will allow us to store a remote address in the same place as a locals
26/// address, and send messages with it without caring about the underlying transport details.
27///
28/// **NOTE**: user should not send a received [`RemoteAddress`] to another remote actor,
29/// it is considered as a meaningless operation.
30pub struct RemoteAddress {
31    /// Computed once at construction, see [`RemoteAddress::new`].
32    index: u64,
33    /// Index of the corresponding actor in the remote node.
34    remote_actor_id: u64,
35    /// Address of the IPC connection session actor, used to send encoded messages to the remote
36    /// actor.
37    session: Address<Session>,
38    /// Pre-built encode context reused for every outbound message through this address.
39    encode_context: EncodeContext,
40}
41
42impl Debug for RemoteAddress {
43    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
44        f.debug_tuple("RemoteAddress")
45            .field(&self.session.index())
46            .field(&self.remote_actor_id)
47            .finish()
48    }
49}
50
51impl Clone for RemoteAddress {
52    #[inline]
53    fn clone(&self) -> Self {
54        Self {
55            index: self.index,
56            remote_actor_id: self.remote_actor_id,
57            session: self.session.clone(),
58            encode_context: self.encode_context.clone(),
59        }
60    }
61}
62
63impl PartialEq for RemoteAddress {
64    #[inline]
65    fn eq(&self, other: &Self) -> bool {
66        self.remote_actor_id == other.remote_actor_id
67            && self.session.index() == other.session.index()
68    }
69}
70
71impl Eq for RemoteAddress {}
72
73impl Hash for RemoteAddress {
74    #[inline]
75    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
76        self.index.hash(state)
77    }
78}
79
80impl RemoteAddress {
81    /// High bit of the 64-bit id space, reserved to tag a [`SenderId::index`] value as
82    /// remote address.
83    pub const REMOTE_FLAG: u64 = 1 << 63;
84
85    /// Constructs a new [`RemoteAddress`] with the specified remote actor index and the address
86    /// of the IPC connection session actor. The `registry` is used to construct the encode
87    /// context for this address.
88    ///
89    /// The `index` is computed by reversing the bits of `session.index()` into bits 0..62 (small
90    /// session ids occupy the high bits, growing downward) and XORing with `remote_actor_id`
91    /// (small ids occupy the low bits, growing upward). Bit 63 is reserved for
92    /// [`REMOTE_FLAG`][Self::REMOTE_FLAG].
93    ///
94    /// The collision of the `index` is structurally possible but extremely unlikely in practice
95    /// since it requires `remote_actor_id * session.index() > 2^63`.
96    pub fn new(
97        remote_actor_id: u64,
98        session: Address<Session>,
99        registry: RemoteActorRegistry,
100    ) -> Self {
101        let index = Self::REMOTE_FLAG | ((session.index().reverse_bits() >> 1) ^ remote_actor_id);
102        Self {
103            index,
104            remote_actor_id,
105            session,
106            encode_context: EncodeContext::new(registry),
107        }
108    }
109
110    fn decode_context(&self) -> DecodeContext {
111        self.encode_context
112            .create_decode_context(self.session.clone())
113    }
114
115    /// Completes when the session has been closed.
116    pub fn closed(&self) -> impl Future<Output = ()> + Send {
117        self.session.closed()
118    }
119
120    /// Checks if the session has been closed.
121    pub fn is_closed(&self) -> bool {
122        self.session.is_closed()
123    }
124
125    /// Returns the capacity of the session.
126    pub fn capacity(&self) -> usize {
127        self.session.capacity()
128    }
129}
130
131async fn decode_and_forward<M>(
132    raw_rx: oneshot::Receiver<Bytes>,
133    tx: oneshot::Sender<M::Result>,
134    decode_context: DecodeContext,
135) where
136    M: Message + Encode,
137    M::Result: Decode,
138{
139    let decode_result = match raw_rx.await {
140        Ok(bytes) => M::Result::decode(bytes, Some(&decode_context)),
141        Err(e) => {
142            let _ = tx.send_err(e);
143            return;
144        }
145    };
146
147    match decode_result {
148        Ok(result) => {
149            let _ = tx.send(result);
150        }
151        Err(e) => {
152            let _ = tx.send_err(e);
153        }
154    }
155}
156
157impl SenderId for RemoteAddress {
158    #[inline]
159    fn index(&self) -> u64 {
160        self.index
161    }
162}
163
164impl<M> Sender<M> for RemoteAddress
165where
166    M: Message + Encode,
167    M::Result: Decode,
168{
169    fn closed(&self) -> ClosedResultFuture<'_> {
170        self.closed().boxed()
171    }
172
173    fn is_closed(&self) -> bool {
174        self.is_closed()
175    }
176
177    fn capacity(&self) -> usize {
178        self.capacity()
179    }
180
181    fn send(&self, msg: M) -> SendResultFuture<'_, M> {
182        let message_bytes = match msg.encode_to_bytes(Some(&self.encode_context)) {
183            Ok(bytes) => bytes,
184            Err(e) => return future::ready(Err(SendError::other(e, msg))).boxed(),
185        };
186
187        let (raw_tx, raw_rx) = oneshot::channel::<Bytes>();
188        let (tx, rx) = oneshot::channel::<M::Result>();
189
190        let decode_context = self.decode_context();
191
192        tokio::spawn(
193            async move {
194                decode_and_forward::<M>(raw_rx, tx, decode_context).await;
195            }
196            .in_current_span(),
197        );
198
199        // this future will be resolved to Ok once the message is in the session actor's mailbox,
200        // it does not guarantee the message arrives at the remote peer actor, the IPC
201        // communication may fail
202        self.session
203            .do_send(RemoteMessage::send(
204                self.remote_actor_id,
205                <M as Encode>::ID,
206                message_bytes,
207                raw_tx,
208            ))
209            // error return by the sender, which means the session actor's mailbox is closed
210            // before receiving the message
211            .map(|result| match result {
212                Ok(_) => Ok(rx),
213                Err(_) => Err(SendError::Closed(msg)),
214            })
215            .boxed()
216    }
217
218    fn do_send(&self, msg: M) -> DoSendResultFuture<'_, M> {
219        let message_bytes = match msg.encode_to_bytes(Some(&self.encode_context)) {
220            Ok(bytes) => bytes,
221            Err(e) => return future::ready(Err(SendError::other(e, msg))).boxed(),
222        };
223
224        // this future will be resolved to Ok once the message is in the session actor's mailbox,
225        // it does not guarantee the message arrives at the remote peer actor, the IPC
226        // communication may fail
227        self.session
228            .do_send(RemoteMessage::do_send(
229                self.remote_actor_id,
230                <M as Encode>::ID,
231                message_bytes,
232            ))
233            // error return by the sender, which means the session actor's mailbox is closed
234            // before receiving the message
235            .map_err(|_| SendError::Closed(msg))
236            .boxed()
237    }
238
239    fn try_send(&self, msg: M) -> SendResult<M> {
240        let message_bytes = match msg.encode_to_bytes(Some(&self.encode_context)) {
241            Ok(bytes) => bytes,
242            Err(e) => return Err(SendError::other(e, msg)),
243        };
244
245        let (raw_tx, raw_rx) = oneshot::channel::<Bytes>();
246        let (tx, rx) = oneshot::channel::<M::Result>();
247
248        let decode_context = self.decode_context();
249
250        tokio::spawn(
251            async move {
252                decode_and_forward::<M>(raw_rx, tx, decode_context).await;
253            }
254            .in_current_span(),
255        );
256
257        match self.session.try_do_send(RemoteMessage::send(
258            self.remote_actor_id,
259            <M as Encode>::ID,
260            message_bytes,
261            raw_tx,
262        )) {
263            Ok(_) => Ok(rx),
264            Err(e) => match e {
265                SendError::Full(_) => Err(SendError::Full(msg)),
266                _ => Err(SendError::Closed(msg)),
267            },
268        }
269    }
270
271    fn try_do_send(&self, msg: M) -> DoSendResult<M> {
272        let message_bytes = match msg.encode_to_bytes(Some(&self.encode_context)) {
273            Ok(bytes) => bytes,
274            Err(e) => return Err(SendError::other(e, msg)),
275        };
276
277        self.session
278            .try_do_send(RemoteMessage::do_send(
279                self.remote_actor_id,
280                <M as Encode>::ID,
281                message_bytes,
282            ))
283            .map_err(|e| match e {
284                SendError::Full(_) => SendError::Full(msg),
285                _ => SendError::Closed(msg),
286            })
287    }
288
289    fn send_timeout(&self, msg: M, timeout: Duration) -> SendResultFuture<'_, M> {
290        let message_bytes = match msg.encode_to_bytes(Some(&self.encode_context)) {
291            Ok(bytes) => bytes,
292            Err(e) => return future::ready(Err(SendError::other(e, msg))).boxed(),
293        };
294
295        let (raw_tx, raw_rx) = oneshot::channel::<Bytes>();
296        let (tx, rx) = oneshot::channel::<M::Result>();
297
298        let decode_context = self.decode_context();
299
300        tokio::spawn(
301            async move {
302                decode_and_forward::<M>(raw_rx, tx, decode_context).await;
303            }
304            .in_current_span(),
305        );
306
307        // this future will be resolved to Ok once the message is in the session actor's mailbox,
308        // it does not guarantee the message arrives at the remote peer actor, the IPC
309        // communication may fail
310        self.session
311            .do_send_timeout(
312                RemoteMessage::send(
313                    self.remote_actor_id,
314                    <M as Encode>::ID,
315                    message_bytes,
316                    raw_tx,
317                ),
318                timeout,
319            )
320            // error return by the sender, which means the session actor's mailbox is closed
321            // before receiving the message
322            .map(|result| match result {
323                Ok(_) => Ok(rx),
324                Err(_) => Err(SendError::Closed(msg)),
325            })
326            .boxed()
327    }
328
329    fn do_send_timeout(&self, msg: M, timeout: Duration) -> DoSendResultFuture<'_, M> {
330        let message_bytes = match msg.encode_to_bytes(Some(&self.encode_context)) {
331            Ok(bytes) => bytes,
332            Err(e) => return future::ready(Err(SendError::other(e, msg))).boxed(),
333        };
334
335        // this future will be resolved to Ok once the message is in the session actor's mailbox,
336        // it does not guarantee the message arrives at the remote peer actor, the IPC
337        // communication may fail
338        self.session
339            .do_send_timeout(
340                RemoteMessage::do_send(self.remote_actor_id, <M as Encode>::ID, message_bytes),
341                timeout,
342            )
343            // error return by the sender, which means the session actor's mailbox is closed
344            // before receiving the message
345            .map_err(|_| SendError::Closed(msg))
346            .boxed()
347    }
348
349    fn blocking_send(&self, msg: M) -> SendResult<M> {
350        let message_bytes = match msg.encode_to_bytes(Some(&self.encode_context)) {
351            Ok(bytes) => bytes,
352            Err(e) => return Err(SendError::other(e, msg)),
353        };
354
355        let (raw_tx, raw_rx) = oneshot::channel::<Bytes>();
356        let (tx, rx) = oneshot::channel::<M::Result>();
357
358        let decode_context = self.decode_context();
359
360        tokio::spawn(
361            async move {
362                decode_and_forward::<M>(raw_rx, tx, decode_context).await;
363            }
364            .in_current_span(),
365        );
366
367        match self.session.blocking_do_send(RemoteMessage::send(
368            self.remote_actor_id,
369            <M as Encode>::ID,
370            message_bytes,
371            raw_tx,
372        )) {
373            Ok(_) => Ok(rx),
374            Err(_) => Err(SendError::Closed(msg)),
375        }
376    }
377
378    fn blocking_do_send(&self, msg: M) -> DoSendResult<M> {
379        let message_bytes = match msg.encode_to_bytes(Some(&self.encode_context)) {
380            Ok(bytes) => bytes,
381            Err(e) => return Err(SendError::other(e, msg)),
382        };
383
384        self.session
385            .blocking_do_send(RemoteMessage::do_send(
386                self.remote_actor_id,
387                <M as Encode>::ID,
388                message_bytes,
389            ))
390            .map_err(|_| SendError::Closed(msg))
391    }
392}
393
394impl<M> From<RemoteAddress> for Recipient<M>
395where
396    M: Message + Encode,
397    M::Result: Decode,
398{
399    fn from(address: RemoteAddress) -> Self {
400        Recipient::new(Arc::new(address))
401    }
402}
acktor_ipc/remote_address.rs

acktor_ipc/
remote_address.rs
