paladin/task/
mod.rs

1//! Task and TaskResult types.
2//!
3//! Tasks encode an [`Operation`] paired with arguments and additional metadata.
4//! They represent the payloads used to communicate between
5//! [`Runtime`](crate::runtime::Runtime)s and
6//! [`WorkerRuntime`](crate::runtime::WorkerRuntime)s.
7use std::fmt::Debug;
8
9use anyhow::Result;
10use bytes::Bytes;
11use serde::{Deserialize, Serialize};
12
13use crate::{
14    __private::OPERATIONS,
15    operation::Operation,
16    serializer::{Serializable, Serializer},
17};
18
19/// A [`Task`] encodes an [`Operation`] paired with arguments.
20///
21/// In addition to the [`Operation`] and its arguments, a [`Task`] also contains
22/// metadata and routing information. The routing information is used to
23/// identify the [`Channel`](crate::channel::Channel) to which execution results
24/// should be sent.
25///
26/// Metadata can be any arbitrary [`Serializable`] type.
27/// It's typically used by [`Directive`](crate::directive::Directive)s to encode
28/// additional information about the computation.
29#[derive(Debug)]
30pub struct Task<'a, Op: Operation, Metadata: Serializable> {
31    /// The routing key used to identify the
32    /// [`Channel`](crate::channel::Channel) to which execution results should
33    /// be sent.
34    pub routing_key: String,
35    /// Metadata associated with the [`Task`].
36    pub metadata: Metadata,
37    /// The [`Operation`] to be executed.
38    pub op: &'a Op,
39    /// The arguments to the [`Operation`].
40    pub input: Op::Input,
41}
42
43/// A [`TaskResult`] encodes the result of executing a [`Task`].
44///
45/// The [`TaskResult`] passes back whatever metadata was associated with the
46/// [`Task`] that produced it.
47#[derive(Serialize, Deserialize, Debug)]
48#[serde(bound = "Op: Operation")]
49pub struct TaskOutput<Op: Operation, Metadata: Serializable> {
50    /// Metadata associated with the [`Task`] that produced this result.
51    pub metadata: Metadata,
52    /// The output of the [`Operation`] execution.
53    pub output: Op::Output,
54}
55
56pub type TaskResult<Op, Metadata> = Result<TaskOutput<Op, Metadata>>;
57
58/// A [`Task`] that has been serialized for remote execution.
59///
60/// This type is used to facilitate opaque execution of [`Operation`]s, such
61/// that executors can execute arbitrary [`Operation`]s.
62#[derive(Serialize, Deserialize, Debug, Clone)]
63pub struct AnyTask {
64    /// The routing key used to identify the
65    /// [`Channel`](crate::channel::Channel) to which execution results should
66    /// be sent.
67    pub routing_key: String,
68    /// Serialized metadata associated with the [`Task`].
69    pub metadata: Bytes,
70    /// The serialized [`Operation`] to be executed.
71    pub op: Bytes,
72    /// The unique identifier of the [`Operation`] to be executed.
73    pub operation_id: u8,
74    /// Serialized arguments to the [`Operation`].
75    pub input: Bytes,
76    /// The [`Serializer`] used to serialize and deserialize the [`Operation`]
77    /// arguments.
78    pub serializer: Serializer,
79}
80
81/// Serialized output of a [`Task`].
82#[derive(Serialize, Deserialize, Debug)]
83pub struct AnyTaskOutput {
84    /// Serialized metadata associated with the [`Task`].
85    pub metadata: Bytes,
86    /// Serialized output of the [`Operation`] execution.
87    pub output: Bytes,
88    /// The [`Serializer`] used to serialize and deserialize the [`Operation`].
89    pub serializer: Serializer,
90}
91
92impl<Op: Operation, Metadata: Serializable> TryFrom<AnyTaskOutput> for TaskOutput<Op, Metadata> {
93    type Error = anyhow::Error;
94
95    fn try_from(
96        AnyTaskOutput {
97            metadata,
98            output,
99            serializer,
100        }: AnyTaskOutput,
101    ) -> Result<Self> {
102        let metadata = serializer.from_bytes(&metadata)?;
103        let output = serializer.from_bytes(&output)?;
104
105        Ok(TaskOutput { metadata, output })
106    }
107}
108
109/// A serializable `Result` type for [`AnyTaskOutput`].
110///
111/// `Result` isn't serializable, so we need to wrap it in a type that is.
112#[derive(Serialize, Deserialize, Debug)]
113pub enum AnyTaskResult {
114    Ok(AnyTaskOutput),
115    Err(String),
116}
117
118impl<'a, Op: Operation, Metadata: Serializable> Task<'a, Op, Metadata> {
119    /// Convert a [`Task`] into an opaque [`AnyTask`].
120    pub fn as_any_task(&self, serializer: Serializer) -> Result<AnyTask> {
121        let routing_key = self.routing_key.clone();
122        let metadata = serializer.to_bytes(&self.metadata)?;
123        let input = serializer.to_bytes(&self.input)?;
124        let op = serializer.to_bytes(self.op)?;
125
126        Ok(AnyTask {
127            routing_key,
128            metadata,
129            operation_id: Op::ID,
130            op,
131            input,
132            serializer,
133        })
134    }
135}
136
137impl AnyTaskResult {
138    /// Convert an opaque [`AnyTaskResult`] into a typed [`TaskResult`].
139    pub fn into_task_result<Op: Operation, Metadata: Serializable>(
140        self,
141    ) -> TaskResult<Op, Metadata> {
142        match self {
143            Self::Ok(any_task_output) => Ok(any_task_output.try_into()?),
144            Self::Err(msg) => Err(anyhow::anyhow!(msg)),
145        }
146    }
147}
148
149impl<Op: Operation, Metadata: Serializable> From<AnyTaskResult>
150    for Result<TaskOutput<Op, Metadata>>
151{
152    fn from(value: AnyTaskResult) -> Self {
153        value.into_task_result()
154    }
155}
156
157impl AnyTask {
158    /// Opaque execution of a [`Task`].
159    ///
160    /// This function is used to execute arbitrary [`Operation`]s. It uses the
161    /// [`RemoteExecute::ID`](crate::operation::RemoteExecute::ID) field to
162    /// acquire the correct execution pointer from the [`static@OPERATIONS`]
163    /// slice.
164    pub async fn remote_execute(self) -> crate::operation::Result<AnyTaskOutput> {
165        OPERATIONS[self.operation_id as usize](self).await
166    }
167}
paladin/task/mod.rs

paladin/task/
mod.rs
