hyperactor/
message.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//! This module provides a framework for mutating serialized multipart messages
10//! without deserializing the full message. This capability is useful when
11//! sending messages to a remote destination through intermediate nodes, where
12//! the intermediate nodes do not contain the message's type information.
13//!
14//! Briefly, it works by following these steps:
15//!
16//! 1. On the sender side, the typed message is serialized with multipart
17//!    encoding and bundled in a `wirevalue::Any<wirevalue::encoding::Multipart>`.
18//! 2. On intermediate nodes, the serialized message is relayed and selected
19//!    typed parts are mutated in place.
20//! 3. On the receiver side, the serialized message is delivered to the ordinary
21//!    typed handler port and deserialized as the final message type.
22//!
23//! One main use case of this framework is to mutate the reply ports of a
24//! multicast message, so the replies can be relayed through intermediate nodes,
25//! rather than directly sent to the original sender.
26
27#[cfg(test)]
28mod tests {
29    use serde::Deserialize;
30    use serde::Serialize;
31
32    use crate::PortRef;
33    use crate::PortRefRepr;
34    use crate::accum::ReducerSpec;
35    use crate::accum::StreamingReducerOpts;
36    use crate::testing::ids::test_port_id;
37
38    // Used to demonstrate a user defined reply type.
39    #[derive(Debug, PartialEq, Serialize, Deserialize, typeuri::Named)]
40    struct MyReply(String);
41
42    // Used to demonstrate a two-way message type.
43    #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, typeuri::Named)]
44    struct MyMessage {
45        arg0: bool,
46        arg1: u32,
47        reply0: PortRef<String>,
48        reply1: PortRef<MyReply>,
49    }
50
51    #[test]
52    fn test_multipart_part_mutation() {
53        let original_port0 = PortRef::attest(test_port_id("world_0", "actor", 123));
54        let original_port1 = PortRef::attest_reducible(
55            test_port_id("world_1", "actor1", 456),
56            Some(ReducerSpec {
57                typehash: 123,
58                builder_params: None,
59            }),
60            StreamingReducerOpts::default(),
61        );
62        let my_message = MyMessage {
63            arg0: true,
64            arg1: 42,
65            reply0: original_port0.clone(),
66            reply1: original_port1.clone(),
67        };
68
69        let serialized_multipart_my_message =
70            wirevalue::Any::<wirevalue::encoding::Multipart>::serialize(&my_message).unwrap();
71
72        let mut message =
73            wirevalue::Any::<wirevalue::encoding::Multipart>::serialize(&my_message).unwrap();
74        assert_eq!(message, serialized_multipart_my_message);
75
76        let new_port_id0 = test_port_id("world_0", "comm", 680);
77        assert_ne!(&new_port_id0, original_port0.port_addr());
78        let new_port_id1 = test_port_id("world_1", "comm", 257);
79        assert_ne!(&new_port_id1, original_port1.port_addr());
80
81        let mut new_ports = vec![&new_port_id0, &new_port_id1].into_iter();
82        message
83            .visit_multipart_parts_mut::<PortRefRepr, anyhow::Error>(|b| {
84                let port = new_ports.next().unwrap();
85                b.update_port_addr(port.clone());
86                Ok(())
87            })
88            .unwrap();
89
90        let new_port0 = PortRef::<String>::attest(new_port_id0);
91        let new_port1 = PortRef::<MyReply>::attest_reducible(
92            new_port_id1,
93            Some(ReducerSpec {
94                typehash: 123,
95                builder_params: None,
96            }),
97            StreamingReducerOpts::default(),
98        );
99        let new_my_message = message.deserialized_unchecked::<MyMessage>().unwrap();
100        assert_eq!(
101            new_my_message,
102            MyMessage {
103                arg0: true,
104                arg1: 42,
105                reply0: new_port0,
106                reply1: new_port1,
107            }
108        );
109    }
110}
hyperactor/message.rs

hyperactor/
message.rs
