dynamo_runtime/
pipeline.rs

1// SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4/// In a Pipeline, the [`AsyncEngine`] is constrained to take a [`Context`] as input and return
5/// a [`super::engine::ResponseStream`] as output.
6use serde::{Deserialize, Serialize};
7
8mod nodes;
9pub use nodes::{
10    Operator, PipelineNode, PipelineOperator, SegmentSink, SegmentSource, Service, ServiceBackend,
11    ServiceFrontend, Sink, Source,
12};
13
14pub mod context;
15pub mod error;
16pub mod network;
17pub use network::egress::addressed_router::{AddressedPushRouter, AddressedRequest};
18pub use network::egress::push_router::{PushRouter, RouterMode, WorkerLoadMonitor};
19pub mod registry;
20
21pub use crate::engine::{
22    self as engine, AsyncEngine, AsyncEngineContext, AsyncEngineContextProvider, Data, DataStream,
23    Engine, EngineStream, EngineUnary, RequestStream, ResponseStream, async_trait,
24};
25pub use anyhow::Error;
26pub use context::Context;
27pub use error::{PipelineError, PipelineErrorExt, TwoPartCodecError};
28
29/// Pipeline inputs carry a [`Context`] which can be used to carry metadata or additional information
30/// about the request. This information propagates through the stages, both local and distributed.
31pub type SingleIn<T> = Context<T>;
32
33/// Pipeline input for streaming-request engines: a trait-object alias around a
34/// stream that carries its own cancellation context (via the
35/// [`AsyncEngineContextProvider`] half of [`AsyncEngineStream`]).
36///
37/// This is the input-side mirror of [`ManyOut`]. Both aliases resolve to the
38/// same underlying [`EngineStream<T>`] type; the directional names are
39/// documentary. Earlier definitions wrapped the stream in a `Context<…>`
40/// (`Context<DataStream<T>>`); that shape was uninstantiable because
41/// `DataStream<T>` is `!Sync` while `Context<T: Data>` requires `Sync`. The
42/// trait-object form solves that cleanly: the cancellation surface is part of
43/// the trait contract rather than an outer wrapper.
44pub type ManyIn<T> = EngineStream<T>;
45
46/// Type alias for the output of pipeline that returns a single value
47pub type SingleOut<T> = EngineUnary<T>;
48
49/// Type alias for the output of pipeline that returns multiple values
50pub type ManyOut<T> = EngineStream<T>;
51
52pub type ServiceEngine<T, U> = Engine<T, U, Error>;
53
54/// Unary Engine is a pipeline that takes a single input and returns a single output
55pub type UnaryEngine<T, U> = ServiceEngine<SingleIn<T>, SingleOut<U>>;
56
57/// `ClientStreaming` Engine is a pipeline that takes multiple inputs and returns a single output
58/// Typically the engine will consume the entire input stream; however, it can also decided to exit
59/// early and emit a response without consuming the entire input stream.
60pub type ClientStreamingEngine<T, U> = ServiceEngine<ManyIn<T>, SingleOut<U>>;
61
62/// `ServerStreaming` takes a single input and returns multiple outputs.
63pub type ServerStreamingEngine<T, U> = ServiceEngine<SingleIn<T>, ManyOut<U>>;
64
65/// `BidirectionalStreaming` takes multiple inputs and returns multiple outputs. Input and output values
66/// are considered independent of each other; however, they could be constrained to be related.
67pub type BidirectionalStreamingEngine<T, U> = ServiceEngine<ManyIn<T>, ManyOut<U>>;
68
69pub trait AsyncTransportEngine<T: Data + PipelineIO, U: Data + PipelineIO>:
70    AsyncEngine<T, U, Error> + Send + Sync + 'static
71{
72}
73
74// pub type TransportEngine<T, U> = Arc<dyn AsyncTransportEngine<T, U>>;
75
76mod sealed {
77    use super::*;
78
79    #[allow(dead_code)]
80    pub struct Token;
81
82    pub trait Connectable {
83        type DataType: Data;
84    }
85
86    impl<T: Data> Connectable for Context<T> {
87        type DataType = T;
88    }
89    impl<T: Data> Connectable for EngineUnary<T> {
90        type DataType = T;
91    }
92    impl<T: Data> Connectable for EngineStream<T> {
93        type DataType = T;
94    }
95}
96
97pub trait PipelineIO: sealed::Connectable + AsyncEngineContextProvider + 'static {
98    fn id(&self) -> String;
99}
100
101impl<T: Data> PipelineIO for Context<T> {
102    fn id(&self) -> String {
103        self.id().to_string()
104    }
105}
106impl<T: Data> PipelineIO for EngineUnary<T> {
107    fn id(&self) -> String {
108        self.context().id().to_string()
109    }
110}
111impl<T: Data> PipelineIO for EngineStream<T> {
112    fn id(&self) -> String {
113        self.context().id().to_string()
114    }
115}
116
117#[derive(Serialize, Deserialize, Debug, Clone)]
118pub struct Event {
119    pub id: String,
120}
dynamo_runtime/pipeline.rs

dynamo_runtime/
pipeline.rs
