zenoh_flow/model/descriptor/node/
operator.rs

1//
2// Copyright (c) 2021 - 2023 ZettaScale Technology
3//
4// This program and the accompanying materials are made available under the
5// terms of the Eclipse Public License 2.0 which is available at
6// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
7// which is available at https://www.apache.org/licenses/LICENSE-2.0.
8//
9// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
10//
11// Contributors:
12//   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
13//
14
15use crate::model::descriptor::link::{CompositeInputDescriptor, CompositeOutputDescriptor};
16use crate::model::descriptor::node::{try_load_descriptor_from_file, NodeDescriptor};
17use crate::model::descriptor::LinkDescriptor;
18use crate::prelude::PortId;
19use crate::types::configuration::Merge;
20use crate::types::{Configuration, NodeId};
21use crate::utils::parse_uri;
22use crate::zfresult::{ErrorKind, ZFResult as Result};
23use crate::{bail, zferror};
24use async_recursion::async_recursion;
25use itertools::Itertools;
26use serde::{Deserialize, Serialize};
27use std::collections::HashMap;
28
29/// Describes a simple operator.
30///
31/// Example:
32///
33///
34/// ```yaml
35/// id : PrintSink
36/// uri: file://./target/release/libmy_op.so
37/// configuration:
38///   by: 10
39/// inputs: [Number]
40/// outputs: [Multiplied]
41/// ```
42///
43#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
44pub struct OperatorDescriptor {
45    pub id: NodeId,
46    pub inputs: Vec<PortId>,
47    pub outputs: Vec<PortId>,
48    pub uri: Option<String>,
49    pub configuration: Option<Configuration>,
50}
51
52impl std::fmt::Display for OperatorDescriptor {
53    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
54        write!(f, "{} - Kind: Operator (Simple)", self.id)
55    }
56}
57
58impl OperatorDescriptor {
59    /// Creates a new `OperatorDescriptor` from its YAML representation.
60    ///
61    ///  # Errors
62    /// A variant error is returned if deserialization fails.
63    pub fn from_yaml(data: &str) -> Result<Self> {
64        let dataflow_descriptor = serde_yaml::from_str::<OperatorDescriptor>(data)
65            .map_err(|e| zferror!(ErrorKind::ParsingError, e))?;
66        Ok(dataflow_descriptor)
67    }
68
69    /// Creates a new `OperatorDescriptor` from its JSON representation.
70    ///
71    ///  # Errors
72    /// A variant error is returned if deserialization fails.
73    pub fn from_json(data: &str) -> Result<Self> {
74        let dataflow_descriptor = serde_json::from_str::<OperatorDescriptor>(data)
75            .map_err(|e| zferror!(ErrorKind::ParsingError, e))?;
76        Ok(dataflow_descriptor)
77    }
78
79    /// Returns the JSON representation of the `OperatorDescriptor`.
80    ///
81    ///  # Errors
82    /// A variant error is returned if serialization fails.
83    pub fn to_json(&self) -> Result<String> {
84        serde_json::to_string(&self).map_err(|e| zferror!(ErrorKind::SerializationError, e).into())
85    }
86
87    /// Returns the YAML representation of the `OperatorDescriptor`.
88    ///
89    ///  # Errors
90    /// A variant error is returned if serialization fails.
91    pub fn to_yaml(&self) -> Result<String> {
92        serde_yaml::to_string(&self).map_err(|e| zferror!(ErrorKind::SerializationError, e).into())
93    }
94}
95
96/// Describes a composite operator.
97///
98/// Example:
99///
100///
101/// ```yaml
102/// id: AMagicAIBasedOperator
103/// configuration:
104///   parameter1: value1
105///   parameter2: value2
106/// operators: # Operators composing the flow
107///   - id: ComposedAIDownsampling
108///     descritptor: [file://|...]/some/path/to/its/descritptor/file.yaml
109///     configuration:
110///       paramter3: value3
111///   - id: MagicAI
112///         descriptor: [file://|...]/some/path/to/its/descritptor/file.yaml
113///   - id: MagicPostProcessing
114///         descriptor: [file://|...]/some/path/to/its/descritptor/file.yaml
115/// links:
116/// - from:
117///     node : ComposedAIDownsampling
118///     output : Data
119///   to:
120///     node : MagicAI
121///     input : In
122/// - from:
123///     node : MagicAI
124///     output : Out
125///   to:
126///     node : MagicPostProcessing
127///     input : Data
128///
129/// inputs:
130///     - node: ComposedAIDownsampling
131///       input: Data
132/// outputs:
133///   - node: MagicPostProcessing
134///     output: Data
135/// ```
136///
137#[derive(Serialize, Deserialize, Debug, Clone)]
138pub struct CompositeOperatorDescriptor {
139    pub id: NodeId,
140    pub inputs: Vec<CompositeInputDescriptor>,
141    pub outputs: Vec<CompositeOutputDescriptor>,
142    pub operators: Vec<NodeDescriptor>,
143    pub links: Vec<LinkDescriptor>,
144    pub configuration: Option<Configuration>,
145}
146
147impl std::fmt::Display for CompositeOperatorDescriptor {
148    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
149        write!(f, "{} - Kind: Operator (Composite)", self.id)
150    }
151}
152
153impl CompositeOperatorDescriptor {
154    /// Creates a new `CompositeOperatorDescriptor` from its YAML representation.
155    ///
156    ///  # Errors
157    /// A variant error is returned if deserialization fails.
158    pub fn from_yaml(data: &str) -> Result<Self> {
159        let dataflow_descriptor = serde_yaml::from_str::<CompositeOperatorDescriptor>(data)
160            .map_err(|e| zferror!(ErrorKind::ParsingError, e))?;
161        Ok(dataflow_descriptor)
162    }
163
164    /// Creates a new `CompositeOperatorDescriptor` from its JSON representation.
165    ///
166    ///  # Errors
167    /// A variant error is returned if deserialization fails.
168    pub fn from_json(data: &str) -> Result<Self> {
169        let dataflow_descriptor = serde_json::from_str::<CompositeOperatorDescriptor>(data)
170            .map_err(|e| zferror!(ErrorKind::ParsingError, e))?;
171        Ok(dataflow_descriptor)
172    }
173
174    /// Returns the JSON representation of the `CompositeOperatorDescriptor`.
175    ///
176    ///  # Errors
177    /// A variant error is returned if serialization fails.
178    pub fn to_json(&self) -> Result<String> {
179        serde_json::to_string(&self).map_err(|e| zferror!(ErrorKind::SerializationError, e).into())
180    }
181
182    /// Returns the YAML representation of the `CompositeOperatorDescriptor`.
183    ///
184    ///  # Errors
185    /// A variant error is returned if serialization fails.
186    pub fn to_yaml(&self) -> Result<String> {
187        serde_yaml::to_string(&self).map_err(|e| zferror!(ErrorKind::SerializationError, e).into())
188    }
189
190    /// Flattens the `CompositeOperatorDescriptor` by loading all the composite operators
191    ///
192    ///  # Errors
193    /// A variant error is returned if loading operators fails. Or if the node does not contains an
194    /// operator.
195    #[async_recursion]
196    pub(crate) async fn flatten(
197        mut self,
198        composite_id: NodeId,
199        links: &mut Vec<LinkDescriptor>,
200        global_configuration: Option<Configuration>,
201        ancestors: &mut Vec<String>,
202    ) -> Result<Vec<OperatorDescriptor>> {
203        log::trace!("[Descriptor] Flattening {}", self.id);
204        let mut simple_operators = vec![];
205        self.configuration = global_configuration.merge_overwrite(self.configuration);
206
207        for o in self.operators {
208            let description = match parse_uri(&o.descriptor)? {
209                crate::model::ZFUri::File(path) => try_load_descriptor_from_file(path).await,
210                crate::model::ZFUri::Builtin(_) => bail!(
211                    ErrorKind::ConfigurationError,
212                    "Builtin operators are not yet supported!"
213                ),
214            }?;
215
216            let NodeDescriptor {
217                id: operator_id,
218                descriptor,
219                configuration,
220            } = o;
221
222            let configuration = self.configuration.clone().merge_overwrite(configuration);
223
224            let res_simple = OperatorDescriptor::from_yaml(&description);
225            if let Ok(mut simple_operator) = res_simple {
226                log::trace!(
227                    "[Descriptor] Flattening {} - {} is simple",
228                    self.id,
229                    simple_operator.id
230                );
231                let new_id: NodeId = format!("{composite_id}/{operator_id}").into();
232
233                let output_ids: HashMap<_, _> = self
234                    .outputs
235                    .iter()
236                    .filter(|&output| output.node == operator_id)
237                    .map(|output| (&output.id, &output.output))
238                    .collect();
239
240                let input_ids: HashMap<_, _> = self
241                    .inputs
242                    .iter()
243                    .filter(|&input| input.node == operator_id)
244                    .map(|input| (&input.id, &input.input))
245                    .collect();
246
247                // Updating all the links with the old id to the new ID
248                for l in &mut self.links {
249                    if l.from.node == operator_id {
250                        log::trace!("Updating {} to {}", l.from.node, new_id);
251                        l.from.node = new_id.clone();
252                    }
253                    if l.to.node == operator_id {
254                        log::trace!("Updating {} to {}", l.to.node, new_id);
255                        l.to.node = new_id.clone();
256                    }
257                }
258
259                links
260                    .iter_mut()
261                    .filter(|link| {
262                        link.from.node == composite_id
263                            && output_ids.keys().contains(&&link.from.output)
264                    })
265                    .for_each(|link| {
266                        link.from.node = new_id.clone();
267                        link.from.output = (*output_ids.get(&&link.from.output).unwrap()).clone();
268                    });
269
270                links
271                    .iter_mut()
272                    .filter(|link| {
273                        link.to.node == composite_id && input_ids.keys().contains(&&link.to.input)
274                    })
275                    .for_each(|link| {
276                        link.to.node = new_id.clone();
277                        link.to.input = (*input_ids.get(&&link.to.input).unwrap()).clone();
278                    });
279
280                // Updating the new id
281                simple_operator.id = new_id;
282
283                simple_operator.configuration = configuration
284                    .clone()
285                    .merge_overwrite(simple_operator.configuration);
286
287                log::trace!(
288                    "[Descriptor] Flattening {} - Pushing simple {}",
289                    self.id,
290                    simple_operator.id
291                );
292                // Adding in the list of operators
293                simple_operators.push(simple_operator);
294
295                continue;
296            }
297
298            let res_composite = CompositeOperatorDescriptor::from_yaml(&description);
299            if let Ok(composite_operator) = res_composite {
300                log::trace!(
301                    "[Descriptor] Flattening {} - {} is composite",
302                    self.id,
303                    composite_operator.id
304                );
305                if let Ok(index) = ancestors.binary_search(&descriptor) {
306                    bail!(
307                        ErrorKind::GenericError, // FIXME Dedicated error?
308                        "Possible recursion detected, < {} > would be included again after: {:?}",
309                        descriptor,
310                        &ancestors[index..]
311                    );
312                }
313                ancestors.push(descriptor.clone());
314
315                let mut operators = composite_operator
316                    .flatten(operator_id, &mut self.links, configuration, ancestors)
317                    .await?;
318
319                for operator in operators.iter_mut() {
320                    let new_id: NodeId = format!("{}/{}", composite_id, operator.id).into();
321                    self.links
322                        .iter_mut()
323                        .filter(|link| link.from.node == operator.id || link.to.node == operator.id)
324                        .for_each(|link| {
325                            if link.from.node == operator.id {
326                                link.from.node = new_id.clone();
327                            }
328                            if link.to.node == operator.id {
329                                link.to.node = new_id.clone();
330                            }
331                        });
332
333                    operator.id = new_id;
334                }
335
336                simple_operators.append(&mut operators);
337
338                ancestors.pop();
339                continue;
340            }
341
342            // If we arrive at that code it means that both attempts to parse the descriptor failed.
343            log::error!(
344                "Could not parse < {} > as either a Simple or a Composite Operator:",
345                operator_id
346            );
347            log::error!("Simple: {:?}", res_simple.err().unwrap());
348            log::error!("Composite: {:?}", res_composite.err().unwrap());
349
350            bail!(
351                ErrorKind::ParsingError,
352                "Could not parse < {} >",
353                operator_id
354            );
355        }
356
357        links.append(&mut self.links);
358
359        Ok(simple_operators)
360    }
361}
362
363#[cfg(test)]
364#[path = "../tests/flatten-composite.rs"]
365mod tests;
zenoh_flow/model/descriptor/node/operator.rs

zenoh_flow/model/descriptor/node/
operator.rs
