1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
use uuid;
use tokio;
use tokio::sync::RwLock;
use std::{fmt, sync::Arc};
use log::{info, error, warn};
use crate as pipeless;
#[derive(Debug)]
pub struct PipelineError {
msg: String
}
impl std::error::Error for PipelineError {}
impl std::fmt::Display for PipelineError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.msg.to_string())
}
}
impl From<pipeless::input::pipeline::InputPipelineError> for PipelineError {
fn from(error: pipeless::input::pipeline::InputPipelineError) -> Self {
Self {
msg: error.to_string(),
}
}
}
impl From<pipeless::output::pipeline::OutputPipelineError> for PipelineError {
fn from(error: pipeless::output::pipeline::OutputPipelineError) -> Self {
Self {
msg: error.to_string(),
}
}
}
#[derive(Debug,Eq,PartialEq)]
pub enum PipelineEndReason {
Completed, // End of stream
Error,
Updated, // When the user changes the stream definition
}
impl fmt::Display for PipelineEndReason {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
PipelineEndReason::Completed => write!(f, "completed"),
PipelineEndReason::Error => write!(f, "error"),
PipelineEndReason::Updated => write!(f, "updated"),
}
}
}
/// A Pipeless pipeline is an association of an input pipeline and an
/// output pipeline, plus the stages the frames must pass through
/// The input and output pipelines are handled via independent Gstreamer pipelines
/// We do it on this way for simplicty, because our gst pipelines are open.
/// Also, the input and output pipelines will always run on the same node
/// avoiding inconsistences when a node fails in a cloud setup.
struct Pipeline {
id: uuid::Uuid,
_input_stream_def: pipeless::input::pipeline::StreamDef,
output_stream_def: Option<pipeless::output::pipeline::StreamDef>,
input_pipeline: pipeless::input::pipeline::Pipeline,
output_pipeline: Option<pipeless::output::pipeline::Pipeline>,
frames_path: pipeless::stages::path::FramePath,
}
impl Pipeline {
fn new(
pipeless_bus_sender: &tokio::sync::mpsc::Sender<pipeless::events::Event>,
input_uri: String,
output_uri: Option<String>,
frames_path: pipeless::stages::path::FramePath,
) -> Result<Self, PipelineError> {
let pipeline_id = uuid::Uuid::new_v4();
let input_stream_def =
pipeless::input::pipeline::StreamDef::new(input_uri.clone())?;
let input_pipeline = pipeless::input::pipeline::Pipeline::new(
pipeline_id,
input_stream_def.clone(),
pipeless_bus_sender,
)?;
let mut output_stream_def = None;
if let Some(uri) = output_uri {
if !uri.is_empty() { // Prevent segfault when the output_uri is provided as empty string
output_stream_def = Some(pipeless::output::pipeline::StreamDef::new(uri)?);
}
}
Ok(Pipeline {
id: pipeline_id,
_input_stream_def: input_stream_def,
output_stream_def,
input_pipeline,
// The output pipeline can't be created until we have the input caps
output_pipeline: None,
frames_path,
})
}
/// The output stream of a pipeline is created once we got the input capabilities
/// because the output replicates the input capabilitites by default.
pub fn create_and_start_output_pipeline(
&mut self,
input_caps: String,
pipeless_bus_sender: &tokio::sync::mpsc::Sender<pipeless::events::Event>,
) -> Result<(), pipeless::output::pipeline::OutputPipelineError> {
if let Some(stream_def) = &self.output_stream_def {
// TODO: build streamdefs within pipelines and pass the uri only
let output_pipeline =
pipeless::output::pipeline::Pipeline::new(
self.id,
stream_def.clone(),
&input_caps,
pipeless_bus_sender
)?;
self.output_pipeline = Some(output_pipeline);
}
Ok(())
}
/// Close only stops the gst pipelines. It does not send EOS.
pub fn close(&self) {
if let Err(err) = self.input_pipeline.close() {
warn!(
"Error closing input stream. Pipeline id: {}. Error: {}",
self.id, err.to_string()
);
}
if let Some(out_pipe) = &self.output_pipeline {
if let Err(err) = out_pipe.close() {
warn!(
"Error closing output stream. Pipeline id: {}. Error: {}",
self.id, err.to_string()
);
}
}
}
pub fn get_frames_path(&self) -> pipeless::stages::path::FramePath {
self.frames_path.clone()
}
}
// TODO: the pipeline manager should distribute the workload
// when using a cloud setup among all nodes registered
pub struct Manager {
// The pipeine manager owns the pipeline that will be used
// in the manager thread
pipeline: Arc<RwLock<pipeless::pipeline::Pipeline>>,
// TODO: we could change this by a callback and avoid using references to the dispatcher here
dispatcher_sender: tokio::sync::mpsc::UnboundedSender<pipeless::dispatcher::DispatcherEvent>,
}
impl Manager {
pub fn new(
input_video_uri: String,
output_video_uri: Option<String>,
frames_path: pipeless::stages::path::FramePath,
// The bus needs to be created before the pipeline
pipeless_bus_sender: &tokio::sync::mpsc::Sender<pipeless::events::Event>,
dispatcher_sender: tokio::sync::mpsc::UnboundedSender<pipeless::dispatcher::DispatcherEvent>,
) -> Result<Self, PipelineError> {
let pipeline = Arc::new(RwLock::new(pipeless::pipeline::Pipeline::new(
&pipeless_bus_sender,
input_video_uri,
output_video_uri,
frames_path,
)?));
Ok(Self { pipeline, dispatcher_sender })
}
// Start takes ownership of self because we have to access the bus,
// which cannot be copied nor cloned due to having a receiver field,
// that does not implement Copy nor Clone traits.
pub fn start(
&self,
event_bus: pipeless::events::Bus,
frame_path_executor_arc: Arc<RwLock<pipeless::stages::path::FramePathExecutor>>
) {
let rw_pipeline = self.pipeline.clone();
let dispatcher_sender = self.dispatcher_sender.clone();
let frame_path_executor_arc = frame_path_executor_arc.clone();
// Leave the pipeline manager running as a tokio task
tokio::spawn(async move {
// Process events on the pipeline concurrently. So frames are processed even
// before the previous one has finished its processing.
// The for_each_concurrent will end once the event bus receiver is closed.
let rw_pipeline = rw_pipeline.clone();
let dispatcher_sender = dispatcher_sender.clone();
let pipeless_bus_sender = event_bus.get_sender();
let concurrent_limit = num_cpus::get() * 2; // NOTE: Making benchmarks we found this is a good value
let frame_path_executor_arc = frame_path_executor_arc.clone();
event_bus.process_events(concurrent_limit,
move |event, end_signal| {
let rw_pipeline = rw_pipeline.clone();
let dispatcher_sender = dispatcher_sender.clone();
let pipeless_bus_sender = pipeless_bus_sender.clone();
let frame_path_executor_arc = frame_path_executor_arc.clone();
async move {
match event {
pipeless::events::Event::FrameChangeEvent(e) => {
let frame = e.into_frame();
let frame_path;
{
let read_guard = rw_pipeline.read().await;
frame_path = read_guard.get_frames_path();
}
let out_frame_opt;
{
let frame_path_executor = frame_path_executor_arc.read().await;
out_frame_opt = frame_path_executor.execute_path(frame, frame_path).await;
}
if let Some(out_frame) = out_frame_opt {
let read_guard = rw_pipeline.read().await;
match &read_guard.output_pipeline {
Some(pipe) => {
if let Err(err) = pipe.on_new_frame(out_frame, &pipeless_bus_sender) {
error!("{}", err);
}
}
None => {}
}
} else {
warn!("No frame returned from path execution, skipping frame forwarding to the output (if any).");
}
}
pipeless::events::Event::NewInputCapsEvent(e) => {
let caps = e.get_caps();
info!("New input caps. Creating output pipeline for caps: {}", caps);
let mut write_guard = rw_pipeline.write().await;
if let Err(err) = write_guard.create_and_start_output_pipeline(
caps.to_string(),
&pipeless_bus_sender
) {
error!("Error creating output: {}. The stream will be processed without the output", err);
};
}
pipeless::events::Event::TagsChangeEvent(e) => {
let tags = e.get_tags();
info!("Tags updated to: {}", tags);
let mut write_guard = rw_pipeline.write().await;
match &write_guard.output_pipeline {
Some(pipe) => pipe.on_new_tags(tags.clone()),
None => {
if let Some(out_stream_def) = &mut write_guard.output_stream_def {
out_stream_def.set_initial_tags(tags.clone());
}
}
}
}
pipeless::events::Event::EndOfInputStreamEvent(_e) => {
// TODO: we must take into account that when reading from files the input eos appears sooner than expected
// meaning we could loose frames
{
let write_guard = rw_pipeline.write().await;
if let Some(out_pipe) = &write_guard.output_pipeline {
info!("End of input stream reached. Pipeline id: {}", out_pipe.get_pipeline_id());
if let Err(err) = out_pipe.on_eos() {
error!("Error sending end of stream signal to output: {}", err);
}
} else {
// When there is no output, stop the stream as fast as the input EOS is reached
info!("End of stream reached for pipeline: {}", write_guard.input_pipeline.get_pipeline_id());
let message = pipeless::dispatcher::DispatcherEvent::PipelineFinished(
write_guard.input_pipeline.get_pipeline_id(), PipelineEndReason::Completed
);
if let Err(err) = dispatcher_sender.send(message) {
warn!("Unable to send pipeline finished event to dispatcher. Error: {}", err);
};
// End the processing loop
if let Err(err) = end_signal.send(()).await {
error!("Error signaling stream event loop end: {}", err);
}
}
}
}
pipeless::events::Event::EndOfOutputStreamEvent(_e) => {
let pipeline_id;
{
let read_guard = rw_pipeline.read().await;
pipeline_id = read_guard.id;
}
info!("End of output stream reached for pipeline: {}", pipeline_id);
let message = pipeless::dispatcher::DispatcherEvent::PipelineFinished(
pipeline_id, PipelineEndReason::Completed
);
if let Err(err) = dispatcher_sender.send(message) {
warn!("Unable to send pipeline finished event to dispatcher. Error: {}", err);
};
// End the processing loop
if let Err(err) = end_signal.send(()).await {
error!("Error signaling stream event loop end: {}", err);
}
}
pipeless::events::Event::InputStreamErrorEvent(e) => {
let pipeline_id;
{
let read_guard = rw_pipeline.read().await;
pipeline_id = read_guard.id;
}
error!(
"Stopping stream for pipeline: {} due to input stream error: {}",
pipeline_id, e.get_msg()
);
let message = pipeless::dispatcher::DispatcherEvent::PipelineFinished(
pipeline_id, PipelineEndReason::Error
);
if let Err(err) = dispatcher_sender.send(message) {
warn!("Unable to send pipeline finished event to dispatcher. Error: {}", err);
};
// End the processing loop
if let Err(err) = end_signal.send(()).await {
error!("Error signaling stream event loop end: {}", err)
};
}
pipeless::events::Event::OutputStreamErrorEvent(e) => {
let pipeline_id;
{
let read_guard = rw_pipeline.read().await;
pipeline_id = read_guard.id;
}
error!(
"Stopping stream for pipeline: {} due to output stream error: {}",
pipeline_id, e.get_msg()
);
if let Err(err) = dispatcher_sender
.send(pipeless::dispatcher::DispatcherEvent::PipelineFinished(pipeline_id, PipelineEndReason::Error)) {
warn!("Unable to send pipeline finished event to dispatcher. Error: {}", err);
}
// End the processing loop
if let Err(err) = end_signal.send(()).await {
error!("Error signaling stream event loop end: {}", err)
};
}
}
}
}
).await;
});
}
// Calling stop in the manager does not notify send EOS to the gst pipelines, simply stops them.
// Otherwise, it would produce several undesired EOS events.
pub async fn stop(&self) -> uuid::Uuid {
let read_guard = self.pipeline.read().await;
let pipeline_id = read_guard.id;
read_guard.close();
pipeline_id
}
pub async fn get_pipeline_id(&self) -> uuid::Uuid {
let read_guard = self.pipeline.read().await;
let pipeline_id = read_guard.id;
pipeline_id
}
}