reactive_mutiny/
stream_executor.rs

1//! Contains the logic for executing [Stream] pipelines on their own Tokio tasks with zero-cost [Instruments] options for gathering stats,
2//! logging and so on.\
3//! Four executors are provided to attend to different Stream output item types:
4//!   1. Items that are non-futures & non-fallible. For instance, `Stream::Item = String`
5//!   2. Items that are futures, but are non-fallible: `Stream::Item = Future<Output=DataType>` -- DataType may be, for instance, String
6//!   3. Items that are non-futures, but are fallible: `Stream::Item = Result<DataType, Box<dyn std::error::Error>>`
7//!   4. Items that are fallible futures: `Stream::Item = Future<Output=Result<DataType, Box<dyn std::error::Error>>>` -- allowing futures to time out
8//!
9//! Apart from handling the specific return types, logging & filling in the available metrics, all executors do the same things:
10//!   1. Register start & finish metrics
11//!   2. Log start (info), items (trace) and finish (warn)
12//!
13//! Specific executors do additional work:
14//!   1. Streams that returns a `Result`: logs & counts untreated-errors (errors that made it all the way through the pipeline and
15//!      ended up in the executor), calling the provided `on_err_callback()`;
16//!   2. Streams that returns a `Future`: they will, optionally, compute the time for each future to complete and, also optionally,
17//!      register a time out for each future to be completed -- cancelling the `Future` if it exceeds the time budget.
18
19
20use super::{
21    instruments::Instruments,
22    incremental_averages::AtomicIncrementalAverage64,
23};
24use std::{
25    sync::{
26        Arc,
27        atomic::{
28            AtomicU64,
29            Ordering::Relaxed,
30        },
31    },
32    future::Future,
33    fmt::Debug,
34    time::Duration,
35    error::Error,
36    future,
37};
38use atomic_enum::atomic_enum;
39use futures::stream::{Stream,StreamExt};
40use tokio::time::timeout;
41// using this instead of Tokio's Instant -- or even std's Instant -- saves a system call (and a context switch), avoiding a huge performance prejudice
42use minstant::Instant;
43use log::{trace,info,warn,error};
44
45
46/// The trait to be passed along after the executor has been started
47pub trait StreamExecutorStats: Debug {
48    fn executor_name(&self) -> &String;
49    fn futures_timeout(&self) -> &Duration;
50    fn creation_time(&self) -> &Instant;
51    fn executor_status(&self) -> &AtomicExecutorStatus;
52    fn execution_start_delta_nanos(&self) -> u64;
53    fn execution_finish_delta_nanos(&self) -> u64;
54    fn ok_events_avg_future_duration(&self) -> &AtomicIncrementalAverage64;
55    fn timed_out_events_avg_future_duration(&self) -> &AtomicIncrementalAverage64;
56    fn failed_events_avg_future_duration(&self) -> &AtomicIncrementalAverage64;
57    /// Tells this executor that its Stream will be artificially ended -- so to cause it to cease its execution
58    fn report_scheduled_to_finish(&self);
59}
60
61
62/// See [Instruments]
63#[derive(Debug)]
64pub struct StreamExecutor<const INSTRUMENTS_USIZE:  usize> {
65    executor_name:                 String,
66    futures_timeout:               Duration,
67    creation_time:                 Instant,
68    executor_status:               AtomicExecutorStatus,
69    execution_start_delta_nanos:   AtomicU64,
70    execution_finish_delta_nanos:  AtomicU64,
71
72    // data to the fields bellow will depend on computations being enabled when instantiating this struct
73
74    /// computes: counter of successful events & average times (seconds) for the future resolution
75    pub ok_events_avg_future_duration: AtomicIncrementalAverage64,
76
77    /// computes: counter of timed out events & average times (seconds) for the failed future resolution (even if it should be ~ constant)
78    /// -- events computed here are NOT computed at [failed_events_avg_future_duration]
79    pub timed_out_events_avg_future_duration: AtomicIncrementalAverage64,
80
81    /// computes: counter of all *other* failed events & average times (seconds) for the failed future resolution
82    /// (timed out events, computed by [timed_out_events_avg_future_duration], are NOT computed here)
83    pub failed_events_avg_future_duration: AtomicIncrementalAverage64,
84
85    // currently, we're only able to measure how long it took to execute the future returned by the stream.
86    // to automatically measure the time it took for the pipeline to process the event (time taken to build & return the future),
87    // a new type has to be introduced -- which would wrap around the payload before passing it to the pipeline... and it should
88    // be available at the end, even if the actual "return" type changes along the way...
89    // this may be done transparently if I have my own Stream implementation doing this... but it seems too much...
90    // if needed, applications should do it by their own for now.
91}
92
93// /// A snapshot taken from a (running or not) [StreamExecutor] that may be easily shared around
94// pub struct StreamExecutorStatsSnapshot {
95//     pub executor
96// }
97
98
99/// registers metrics & logs (if opted in) that an executor with the given capabilities has started.\
100///   - `self`: this executor's instance. Example: `self_ref`
101///   - `future`: true if items yielded by this stream are of type `Future<Output=ItemType>`
102///   - `fallible`: true if items yielded by this stream are of type `Result<DataType, ErrType>`
103///   - `futures_timeout`: if different than `Duration::ZERO` means we enforce a timeout for every item (`Future<Outut=ItemType>`) when resolving them
104macro_rules! on_executor_start {
105    ($self: expr, $future: expr, $fallible: expr, $futures_timeout: expr, $INSTRUMENTS: expr) => {
106        $self.register_execution_start();
107        if $INSTRUMENTS.logging() {
108            info!("✓✓✓✓ Stream Executor '{}' started: {}Futures{} / {}Fallible Items & {}Metrics",
109                  $self.executor_name,
110                  if $future {""} else {"Non-"},
111                  if $future {
112                      if $futures_timeout != Duration::ZERO {format!(" (with timeouts of {:?})", $futures_timeout)} else {" (NO timeouts)".to_string()}
113                  } else {
114                      "".to_string()
115                  },
116                  if $fallible {""} else {"Non-"},
117                  if !$INSTRUMENTS.metrics() {"NO "} else {""});
118        }
119    }
120}
121
122/// logs & registers metrics (if opted in) for an `Ok` item yielded by this stream for which we don't have timings for -- most likely a non-future or future with metrics disabled\
123///   - `self`: this executor's instance. Example: `self_ref`
124///   - `item`: the just yielded `Ok` item
125macro_rules! on_non_timed_ok_item {
126    ($self: expr, $item: expr, $INSTRUMENTS: expr) => {
127        {
128            if $INSTRUMENTS.cheap_profiling() {
129                $self.ok_events_avg_future_duration.inc(-1.0);  // since there is no time measurement (item is non-future or METRICS=false), the convention is to use -1.0
130            }
131            if $INSTRUMENTS.tracing() {
132                trace!("✓✓✓✓ Executor '{}' yielded '{:?}'", $self.executor_name, $item);
133            }
134        }
135    }
136}
137
138/// logs & registers metrics (if opted in) for an `Ok` item yielded by this stream for which we DO have timings for -- most certainly, a future
139///   - `self`: this executor's instance. Example: `self_ref`
140///   - `item`: the just yielded `Ok` item
141///   - `elapsed`: the `Duration` it took to resolve the this item's `Future`
142macro_rules! on_timed_ok_item {
143    ($self: expr, $item: expr, $elapsed: expr, $INSTRUMENTS: expr) => {
144        {
145            if $INSTRUMENTS.cheap_profiling() {
146                $self.ok_events_avg_future_duration.inc($elapsed.as_secs_f32());
147            } else {
148                panic!("\nThis macro can only be used if at least one of the Instruments' PROFILING are enabled -- otherwise you should use `on_non_timed_ok_item!(...)` instead");
149            }
150            if $INSTRUMENTS.tracing() {
151                trace!("✓✓✓✓ Executor '{}' yielded '{:?}' in {:?}", $self.executor_name, $item, $elapsed);
152            }
153        }
154    }
155}
156
157/// logs & registers metrics (if opted in) for an `Err` item yielded by this stream for which we don't have timings for -- most likely a non-future or future with metrics disabled\
158///   - `self`: this executor's instance. Example: `self_ref`
159///   - `err`: the just yielded `Err` item
160macro_rules! on_non_timed_err_item {
161    ($self: expr, $err: expr, $INSTRUMENTS: expr) => {
162        {
163            if $INSTRUMENTS.cheap_profiling() {
164                $self.failed_events_avg_future_duration.inc(-1.0);  // since there is no time measurement (item is non-future or METRICS=false), the convention is to use -1.0
165            }
166            if $INSTRUMENTS.logging() {
167                error!("✗✗✗✗ Executor '{}' yielded ERROR '{:?}'", $self.executor_name, $err);
168            }
169        }
170    }
171}
172
173/// logs & registers metrics (if opted in) for an `Err` item yielded by this stream for which we DO have timings for -- most certainly, a future
174///   - `self`: this executor's instance. Example: `self_ref`
175///   - `err`: the just yielded `Err` item
176///   - `elapsed`: the `Duration` it took to resolve the this item's `Future`
177macro_rules! on_timed_err_item {
178    ($self: expr, $err: expr, $elapsed: expr, $INSTRUMENTS: expr) => {
179        {
180            if $INSTRUMENTS.cheap_profiling() {
181                $self.failed_events_avg_future_duration.inc($elapsed.as_secs_f32());
182            } else {
183                panic!("This macro can only be used if at least one of the Instruments' PROFILING are enabled -- otherwise you should use `on_non_timed_err_item!(...)` instead");
184            }
185            if $INSTRUMENTS.logging() {
186                error!("✗✗✗✗ Executor '{}' yielded ERROR '{:?}' in {:?}", $self.executor_name, $err, $elapsed);
187            }
188        }
189    }
190}
191
192/// registers metrics & logs (if opted in) that the either the stream or the executor ended.\
193///   - `self`: this executor's instance. Example: `self_ref`
194///   - `future`: true if items yielded by this stream were of type `Future<Output=ItemType>`
195///   - `fallible`: true if items yielded by this stream were of type `Result<DataType, ErrType>`
196///   - `futures_timeout`: if different than `Duration::ZERO`, means time out stats could have beem collected and may be logged
197macro_rules! on_executor_end {
198    ($self: expr, $future: expr, $fallible: expr, $futures_timeout: expr, $INSTRUMENTS: expr) => {
199        $self.register_execution_finish();
200        let stream_ended = $self.executor_status.load(Relaxed) == ExecutorStatus::StreamEnded;
201        let execution_nanos = $self.execution_finish_delta_nanos.load(Relaxed) - $self.execution_start_delta_nanos.load(Relaxed);
202        if $INSTRUMENTS.logging() && $INSTRUMENTS.cheap_profiling() {
203            let (ok_counter, ok_avg_seconds) = $self.ok_events_avg_future_duration.probe();
204            let (timed_out_counter, timed_out_avg_seconds) = $self.timed_out_events_avg_future_duration.probe();
205            let (failed_counter, failed_avg_seconds) = $self.failed_events_avg_future_duration.probe();
206            let execution_secs: f64 = Duration::from_nanos(execution_nanos).as_secs_f64() + f64::MIN_POSITIVE /* dirty way to avoid silly /0 divisions */;
207            let ok_stats = if $future {
208                               format!("ok: {} events; avg {:?} - {:.5}/sec", ok_counter, Duration::from_secs_f32(ok_avg_seconds), ok_counter as f64 / execution_secs)
209                           } else {
210                               format!("ok: {} events", ok_counter)
211                           };
212            let timed_out_stats = if $future && $futures_timeout != Duration::ZERO {
213                                      format!(" | time out: {} events; avg {:?} - {:.5}/sec", timed_out_counter, Duration::from_secs_f32(timed_out_avg_seconds), timed_out_counter as f64 / execution_secs)
214                                  } else {
215                                      format!("")
216                                  };
217            let failed_stats = if $future && $fallible {
218                                   format!(" | failed: {} events; avg {:?} - {:.5}/sec", failed_counter, Duration::from_secs_f32(failed_avg_seconds), failed_counter as f64 / execution_secs)
219                               } else if $fallible {
220                                   format!(" | failed: {} events", failed_counter)
221                               } else {
222                                   format!("")
223                               };
224            warn!("✓✓✓✓ {} '{}' ended after running for {:?} -- stats: | {}{}{}",
225                  if stream_ended {"Stream"} else {"Executor"},
226                  $self.executor_name,
227                  Duration::from_nanos(execution_nanos),
228                  ok_stats,
229                  timed_out_stats,
230                  failed_stats);
231        } else if $INSTRUMENTS.logging() {
232            warn!("✓✓✓✓ {} '{}' ended after running for {:?} -- metrics were disabled",
233                  if stream_ended {"Stream"} else {"Executor"},
234                  $self.executor_name,
235                  Duration::from_nanos(execution_nanos));
236        }
237    }
238}
239
240
241impl<const INSTRUMENTS_USIZE:  usize>
242StreamExecutor<INSTRUMENTS_USIZE> {
243
244    const INSTRUMENTS: Instruments = {Instruments::from(INSTRUMENTS_USIZE)};
245
246    /// Initializes an executor that should not timeout any futures returned by the Stream
247    pub fn new<IntoString: Into<String>>(executor_name: IntoString) -> Arc<Self> {
248        Self::with_futures_timeout(executor_name, Duration::ZERO)
249    }
250
251    /// Initializes an executor that should be able to `timeout` Futures returned by the Stream
252    pub fn with_futures_timeout<IntoString: Into<String>>(executor_name: IntoString, futures_timeout: Duration) -> Arc<Self> {
253        Arc::new(Self {
254            executor_name: executor_name.into(),
255            futures_timeout,
256            creation_time:                        Instant::now(),
257            executor_status:                      AtomicExecutorStatus::new(ExecutorStatus::NotStarted),
258            execution_start_delta_nanos:          AtomicU64::new(u64::MAX),
259            execution_finish_delta_nanos:         AtomicU64::new(u64::MAX),
260            ok_events_avg_future_duration:        AtomicIncrementalAverage64::new(),
261            failed_events_avg_future_duration:    AtomicIncrementalAverage64::new(),
262            timed_out_events_avg_future_duration: AtomicIncrementalAverage64::new(),
263        })
264    }
265
266
267    /// Sets the executor state & computes some always-enabled metrics
268    fn register_execution_start(&self) {
269        self.executor_status.store(ExecutorStatus::Running, Relaxed);
270        self.execution_start_delta_nanos.store(self.creation_time.elapsed().as_nanos() as u64, Relaxed);
271    }
272
273    /// Sets the executor state & computes some always-enabled metrics
274    fn register_execution_finish(&self) {
275        loop {
276            if self.executor_status.compare_exchange(ExecutorStatus::Running,           ExecutorStatus::StreamEnded,           Relaxed, Relaxed).is_ok() ||
277               self.executor_status.compare_exchange(ExecutorStatus::ScheduledToFinish, ExecutorStatus::ProgrammaticallyEnded, Relaxed, Relaxed).is_ok() {
278                break
279            }
280        }
281        self.execution_finish_delta_nanos.store(self.creation_time.elapsed().as_nanos() as u64, Relaxed);
282    }
283
284    /// Spawns an optimized executor for a Stream of `ItemType`s which are:
285    ///   * Futures  -- `ItemType := Future<Output=InnerFallibleType>`
286    ///   * Fallible -- `InnerFallibleType := Result<InnerType, Box<dyn std::error::Error>>.`
287    /// 
288    /// NOTE: special (optimized) versions are spawned depending if we should or not enforce each item's `Future` a resolution timeout.\
289    /// `stream_ended_callback()` should be either a closure or a generic function declared with `<StreamExecutorType: StreamExecutorStats>(executor_stats: Arc<StreamExecutorStats>)`
290    pub fn spawn_executor<OutItemType:        Send + Debug,
291                          FutureItemType:     Future<Output=Result<OutItemType, Box<dyn std::error::Error + Send + Sync>>> + Send,
292                          CloseVoidAsyncType: Future<Output=()>   + Send + 'static,
293                          ErrVoidAsyncType:   Future<Output=()>   + Send + 'static>
294                         (self:                  Arc<Self>,
295                          concurrency_limit:     u32,
296                          on_err_callback:       impl Fn(Box<dyn Error + Send + Sync>)                   -> ErrVoidAsyncType   + Send + Sync + 'static,
297                          stream_ended_callback: impl FnOnce(Arc<dyn StreamExecutorStats + Send + Sync>) -> CloseVoidAsyncType + Send + Sync + 'static,
298                          stream:                impl Stream<Item=FutureItemType>                                              + Send        + 'static) {
299
300        match self.futures_timeout {
301
302            // spawns an optimized executor that do not track `Future`s timeouts
303            Duration::ZERO => {
304                tokio::spawn(async move {
305                    let self_ref: &Self = &self;
306                    let on_err_callback_ref = &on_err_callback;
307                    on_executor_start!(self_ref, true, true, self_ref.futures_timeout, Self::INSTRUMENTS);
308                    let mut start = Instant::now();     // item's Future resolution time -- declared here to allow optimizations when METRICS=false
309                    let item_processor = |future_element| {
310                        async move {
311                            if Self::INSTRUMENTS.cheap_profiling() {
312                                start = Instant::now();
313                            }
314                            match future_element.await {
315                                Ok(yielded_item) => {
316                                    if Self::INSTRUMENTS.cheap_profiling() {
317                                        let elapsed = start.elapsed();
318                                        on_timed_ok_item!(self_ref, yielded_item, elapsed, Self::INSTRUMENTS);
319                                    } else {
320                                        on_non_timed_ok_item!(self_ref, yielded_item, Self::INSTRUMENTS);
321                                    }
322                                },
323                                Err(err) => {
324                                    if Self::INSTRUMENTS.cheap_profiling() {
325                                        let elapsed = start.elapsed();
326                                        on_timed_err_item!(self_ref, err, elapsed, Self::INSTRUMENTS);
327                                    } else {
328                                        on_non_timed_err_item!(self_ref, err, Self::INSTRUMENTS);
329                                    }
330                                    on_err_callback_ref(err).await;
331                                },
332                            }
333                        }
334                    };
335                    match concurrency_limit {
336                        1 => stream.for_each(item_processor).await,     // faster in `futures 0.3` -- may be useless in the future
337                        _ => stream.for_each_concurrent(concurrency_limit as usize, item_processor).await,
338                    }
339                    on_executor_end!(self_ref, true, true, Duration::ZERO, Self::INSTRUMENTS);
340                    stream_ended_callback(self).await;
341                });
342            },
343
344            // spawns an optimized executor that tracks `Future`s timeouts
345            _ => {
346                tokio::spawn(async move {
347                    let self_ref: &Self = &self;
348                    let on_err_callback_ref = &on_err_callback;
349                    on_executor_start!(self_ref, true, true, self_ref.futures_timeout, Self::INSTRUMENTS);
350                    let mut start = Instant::now();     // item's Future resolution time -- declared here to allow optimizations when METRICS=false
351                    let item_processor = |future_element| {
352                        async move {
353                            if Self::INSTRUMENTS.cheap_profiling() {
354                                start = Instant::now();
355                            }
356                            match timeout(self_ref.futures_timeout, future_element).await {
357                                Ok(non_timed_out_result) => match non_timed_out_result {
358                                                                                          Ok(yielded_item) => {
359                                                                                              if Self::INSTRUMENTS.cheap_profiling() {
360                                                                                                  let elapsed = start.elapsed();
361                                                                                                  on_timed_ok_item!(self_ref, yielded_item, elapsed, Self::INSTRUMENTS);
362                                                                                              } else {
363                                                                                                  on_non_timed_ok_item!(self_ref, yielded_item, Self::INSTRUMENTS);
364                                                                                              }
365                                                                                          },
366                                                                                          Err(err) => {
367                                                                                              if Self::INSTRUMENTS.cheap_profiling() {
368                                                                                                  let elapsed = start.elapsed();
369                                                                                                  on_timed_err_item!(self_ref, err, elapsed, Self::INSTRUMENTS);
370                                                                                              } else {
371                                                                                                  on_non_timed_err_item!(self_ref, err, Self::INSTRUMENTS);
372                                                                                              }
373                                                                                              on_err_callback_ref(err).await;
374                                                                                          },
375                                                                                      },
376                                Err(_time_out_err) => {
377                                    if Self::INSTRUMENTS.cheap_profiling() {
378                                        let elapsed = start.elapsed();
379                                        self_ref.timed_out_events_avg_future_duration.inc(elapsed.as_secs_f32());
380                                        if Self::INSTRUMENTS.logging() {
381                                            error!("🕝🕝🕝🕝 Executor '{}' TIMED OUT after {:?}", self_ref.executor_name, elapsed);
382                                        }
383                                    } else if Self::INSTRUMENTS.logging() {
384                                        error!("🕝🕝🕝🕝 Executor '{}' TIMED OUT", self_ref.executor_name);
385                                    }
386                                }
387                            }
388                        }
389                    };
390                    match concurrency_limit {
391                        1 => stream.for_each(item_processor).await,     // faster in `futures 0.3` -- may be useless in other versions
392                        _ => stream.for_each_concurrent(concurrency_limit as usize, item_processor).await,
393                    }
394                    on_executor_end!(self_ref, true, true, self_ref.futures_timeout, Self::INSTRUMENTS);
395                    stream_ended_callback(self).await;
396                });
397            },
398
399        }
400    }
401
402    /// Spawns an optimized executor for a `Stream` of `FutureItemType`s (non fallible Futures).\
403    /// NOTE: Since this is non-fallible, timeouts are enforced but not detectable. If that is not acceptable, use [Self::spawn_executor()] instead.\
404    /// `stream_ended_callback()` should be either a closure or a generic function declared with `<StreamExecutorType: StreamExecutorStats>(executor_stats: Arc<StreamExecutorStats>)`
405    pub fn spawn_futures_executor<OutItemType:        Send + Debug,
406                                  FutureItemType:     Future<Output=OutItemType> + Send,
407                                  CloseVoidAsyncType: Future<Output=()> + Send + 'static>
408                                 (self:                  Arc<Self>,
409                                  concurrency_limit:     u32,
410                                  stream_ended_callback: impl FnOnce(Arc<dyn StreamExecutorStats + Send + Sync>) -> CloseVoidAsyncType + Send + Sync + 'static,
411                                  stream:                impl Stream<Item=FutureItemType>                                              + Send        + 'static) {
412
413        match self.futures_timeout {
414
415            // spawns an optimized executor that do not track `Future`s timeouts
416            Duration::ZERO => {
417                tokio::spawn(async move {
418                    let self_ref: &Self = &self;
419                    on_executor_start!(self_ref, true, true, self_ref.futures_timeout, Self::INSTRUMENTS);
420                    let mut start = Instant::now();     // item's Future resolution time -- declared here to allow optimizations when METRICS=false
421                    let item_processor = |future_element| {
422                        async move {
423                            if Self::INSTRUMENTS.cheap_profiling() {
424                                start = Instant::now();
425                            }
426                            let yielded_item = future_element.await;
427                            if Self::INSTRUMENTS.cheap_profiling() {
428                                let elapsed = start.elapsed();
429                                on_timed_ok_item!(self_ref, yielded_item, elapsed, Self::INSTRUMENTS);
430                            } else {
431                                on_non_timed_ok_item!(self_ref, yielded_item, Self::INSTRUMENTS);
432                            }
433                        }
434                    };
435                    match concurrency_limit {
436                        1 => stream.for_each(item_processor).await,     // faster in `futures 0.3` -- may be useless in the future
437                        _ => stream.for_each_concurrent(concurrency_limit as usize, item_processor).await,
438                    }
439                    on_executor_end!(self_ref, true, true, Duration::ZERO, Self::INSTRUMENTS);
440                    stream_ended_callback(self).await;
441                });
442            },
443
444            // spawns an optimized executor that tracks `Future`s timeouts
445            _ => {
446                tokio::spawn(async move {
447                    let self_ref: &Self = &self;
448                    on_executor_start!(self_ref, true, true, self_ref.futures_timeout, Self::INSTRUMENTS);
449                    let mut start = Instant::now();     // item's Future resolution time -- declared here to allow optimizations when METRICS=false
450                    let item_processor = |future_element| {
451                        async move {
452                            if Self::INSTRUMENTS.cheap_profiling() {
453                                start = Instant::now();
454                            }
455                            match timeout(self_ref.futures_timeout, future_element).await {
456                                Ok(non_timed_out_result) => {
457                                    if Self::INSTRUMENTS.cheap_profiling() {
458                                        let elapsed = start.elapsed();
459                                        on_timed_ok_item!(self_ref, non_timed_out_result, elapsed, Self::INSTRUMENTS);
460                                    } else {
461                                        on_non_timed_ok_item!(self_ref, non_timed_out_result, Self::INSTRUMENTS);
462                                    }
463                                }
464                                Err(_time_out_err) => {
465                                    if Self::INSTRUMENTS.cheap_profiling() {
466                                        let elapsed = start.elapsed();
467                                        self_ref.timed_out_events_avg_future_duration.inc(elapsed.as_secs_f32());
468                                        if Self::INSTRUMENTS.logging() {
469                                            error!("🕝🕝🕝🕝 Executor '{}' TIMED OUT after {:?}", self_ref.executor_name, elapsed);
470                                        }
471                                    } else if Self::INSTRUMENTS.logging() {
472                                        error!("🕝🕝🕝🕝 Executor '{}' TIMED OUT", self_ref.executor_name);
473                                    }
474                                }
475                            }
476                        }
477                    };
478                    match concurrency_limit {
479                        1 => stream.for_each(item_processor).await,     // faster in `futures 0.3` -- may be useless in other versions
480                        _ => stream.for_each_concurrent(concurrency_limit as usize, item_processor).await,
481                    }
482                    on_executor_end!(self_ref, true, true, self_ref.futures_timeout, Self::INSTRUMENTS);     // notice the `fallible = true` here -- this is due to the timeouts, that shows as errors
483                    stream_ended_callback(self).await;
484                });
485            },
486
487        }
488    }
489
490    /// Spawns an optimized executor for a Stream of `FallibleItemType`s, where:\
491    ///   * `FallibleItemType := Result<OutItemType, Box<dyn std::error::Error + Send + Sync>>`
492    /// 
493    /// `stream_ended_callback()` should be either a closure or a generic function declared with `<StreamExecutorType: StreamExecutorStats>(executor_stats: Arc<StreamExecutorStats>)`
494    pub fn spawn_fallibles_executor<OutItemType:        Send + Debug,
495                                    CloseVoidAsyncType: Future<Output=()> + Send + 'static>
496                                   (self:                  Arc<Self>,
497                                    concurrency_limit:     u32,
498                                    on_err_callback:       impl Fn(Box<dyn Error + Send + Sync>)                                           + Send + Sync + 'static,
499                                    stream_ended_callback: impl FnOnce(Arc<dyn StreamExecutorStats + Send + Sync>) -> CloseVoidAsyncType   + Send + Sync + 'static,
500                                    stream:                impl Stream<Item=Result<OutItemType, Box<dyn std::error::Error + Send + Sync>>> + Send + 'static) {
501
502        tokio::spawn(async move {
503            on_executor_start!(self, true, true, Duration::ZERO, Self::INSTRUMENTS);
504            let item_processor = |element| {
505                match element {
506                    Ok(yielded_item) => {
507                        on_non_timed_ok_item!(self, yielded_item, Self::INSTRUMENTS);
508                    },
509                    Err(err) => {
510                        on_non_timed_err_item!(self, err, Self::INSTRUMENTS);
511                        on_err_callback(err);
512                    }
513                }
514            };
515            match concurrency_limit {
516                // faster in `futures 0.3` -- may be useless in the future
517                1 => stream.for_each(|item| {
518                    item_processor(item);
519                    future::ready(())
520                }).await,
521                _ => stream.for_each_concurrent(concurrency_limit as usize, |item| {
522                    item_processor(item);
523                    future::ready(())
524                }).await,
525            }
526            on_executor_end!(self, false, true, Duration::ZERO, Self::INSTRUMENTS);
527            stream_ended_callback(self).await;
528        });
529    }
530
531    /// Spawns an executor for a Stream of `ItemType`s which are not Futures but are fallible:
532    ///   * `InnerFallibleType := Result<ItemType, Box<dyn std::error::Error>>`
533    /// 
534    /// `stream_ended_callback()` should be either a closure or a generic function declared with `<StreamExecutorType: StreamExecutorStats>(executor_stats: Arc<StreamExecutorStats>)`
535    pub fn spawn_non_futures_executor<ItemType:          Send + Debug,
536                                      VoidAsyncType:     Future<Output=()> + Send + 'static>
537                                     (self:                      Arc<Self>,
538                                      concurrency_limit:         u32,
539                                      stream_ended_callback:     impl FnOnce(Arc<dyn StreamExecutorStats + Send + Sync>) -> VoidAsyncType     + Send + Sync + 'static,
540                                      stream:                    impl Stream<Item=Result<ItemType, Box<dyn std::error::Error + Send + Sync>>> + Send        + 'static) {
541        tokio::spawn(async move {
542            on_executor_start!(self, false, true, Duration::ZERO, Self::INSTRUMENTS);
543            let item_processor = |fallible_element| {
544                match fallible_element {
545                    Ok(yielded_item) => on_non_timed_ok_item!(self, yielded_item, Self::INSTRUMENTS),
546                    Err(err) => on_non_timed_err_item!(self, err, Self::INSTRUMENTS),
547                }
548            };
549            match concurrency_limit {
550                // faster in `futures 0.3` -- may be useless in other versions
551                1 => stream.for_each(|fallible_item| {
552                    item_processor(fallible_item);
553                    future::ready(())
554                }).await,
555                _ => stream.for_each_concurrent(concurrency_limit as usize, |fallible_item| {
556                    item_processor(fallible_item);
557                    future::ready(())
558                }).await,
559            }
560            on_executor_end!(self, false, true, Duration::ZERO, Self::INSTRUMENTS);
561            stream_ended_callback(self).await;
562        });
563    }
564
565    /// Spawns an optimized executor for a Stream of `ItemType`s which are not Futures and, also, are not fallible.\
566    /// `stream_ended_callback()` should be either a closure or a generic function declared with `<StreamExecutorType: StreamExecutorStats>(executor_stats: Arc<StreamExecutorStats>)`
567    pub fn spawn_non_futures_non_fallibles_executor<OutItemType:       Send + Debug,
568                                                    VoidAsyncType:     Future<Output=()> + Send + 'static>
569                                                  (self:                  Arc<Self>,
570                                                   concurrency_limit:     u32,
571                                                   stream_ended_callback: impl FnOnce(Arc<dyn StreamExecutorStats + Send + Sync>) -> VoidAsyncType + Send + Sync + 'static,
572                                                   stream:                impl Stream<Item=OutItemType>                                            + Send        + 'static) {
573
574        tokio::spawn(async move {
575            on_executor_start!(self, false, false, Duration::ZERO, Self::INSTRUMENTS);
576            let item_processor = |yielded_item| on_non_timed_ok_item!(self, yielded_item, Self::INSTRUMENTS);
577            match concurrency_limit {
578                // faster in `futures 0.3` -- may be useless in the near future?
579                1 => stream.for_each(|item| {
580                    item_processor(item);
581                    future::ready(())
582                }).await,
583                _ => stream.for_each_concurrent(concurrency_limit as usize, |item| {
584                    item_processor(item);
585                    future::ready(())
586                }).await,
587            }
588            on_executor_end!(self, false, false, Duration::ZERO, Self::INSTRUMENTS);
589            stream_ended_callback(self).await;
590        });
591    }
592
593}
594
595
596impl<const INSTRUMENTS_USIZE: usize> StreamExecutorStats for
597StreamExecutor<INSTRUMENTS_USIZE> {
598    //const INSTRUMENTS: Instruments = Instruments::from(INSTRUMENTS_USIZE);
599    fn executor_name(&self) -> &String {
600        &self.executor_name
601    }
602    fn futures_timeout(&self) -> &Duration {
603        &self.futures_timeout
604    }
605    fn creation_time(&self) -> &Instant {
606        &self.creation_time
607    }
608    fn executor_status(&self) -> &AtomicExecutorStatus {
609        &self.executor_status
610    }
611    fn execution_start_delta_nanos(&self) -> u64 {
612        self.execution_start_delta_nanos.load(Relaxed)
613    }
614    fn execution_finish_delta_nanos(&self) -> u64 {
615        self.execution_finish_delta_nanos.load(Relaxed)
616    }
617    fn ok_events_avg_future_duration(&self) -> &AtomicIncrementalAverage64 {
618        &self.ok_events_avg_future_duration
619    }
620    fn timed_out_events_avg_future_duration(&self) -> &AtomicIncrementalAverage64 {
621        &self.timed_out_events_avg_future_duration
622    }
623    fn failed_events_avg_future_duration(&self) -> &AtomicIncrementalAverage64 {
624        &self.failed_events_avg_future_duration
625    }
626    fn report_scheduled_to_finish(&self) {
627        self.executor_status.store(ExecutorStatus::ScheduledToFinish, Relaxed);
628    }    
629}
630
631/// Unit tests & enforces the requisites of the [stream_executor](self) module.\
632/// Tests here mixes manual & automated assertions -- you should manually inspect the output of each one and check if the log outputs make sense
633#[cfg(any(test,doc))]
634mod tests {
635    use super::*;
636    use std::sync::atomic::AtomicU32;
637    use futures::{
638        stream::{self, StreamExt},
639        channel::mpsc,
640        SinkExt,
641    };
642
643
644    #[ctor::ctor]
645    fn suite_setup() {
646        simple_logger::SimpleLogger::new().with_utc_timestamps().init().unwrap_or_else(|_| eprintln!("--> LOGGER WAS ALREADY STARTED"));
647        info!("minstant: is TSC / RDTSC instruction available for time measurement? {}", minstant::is_tsc_available());
648    }
649
650    // fallible futures
651    ///////////////////
652
653    #[cfg_attr(not(doc),tokio::test)]
654    async fn spawn_non_timeout_futures_fallible_executor_with_logs_and_metrics() {
655        assert_spawn_futures_fallible_executor::<{Instruments::LogsWithMetrics.into()}>(StreamExecutor::new("executor with logs & metrics")).await;
656    }
657
658    #[cfg_attr(not(doc),tokio::test)]
659    async fn spawn_non_timeout_futures_fallible_executor_with_metrics_and_no_logs() {
660        assert_spawn_futures_fallible_executor::<{Instruments::MetricsWithoutLogs.into()}>(StreamExecutor::new("executor with metrics & NO logs")).await;
661    }
662
663    #[cfg_attr(not(doc),tokio::test)]
664    async fn spawn_non_timeout_futures_fallible_executor_with_logs_and_no_metrics() {
665        assert_spawn_futures_fallible_executor::<{Instruments::LogsWithoutMetrics.into()}>(StreamExecutor::new("executor with logs & NO metrics")).await;
666    }
667
668    #[cfg_attr(not(doc),tokio::test)]
669    async fn spawn_non_timeout_futures_fallible_executor_with_no_logs_and_no_metrics() {
670        assert_spawn_futures_fallible_executor::<{Instruments::NoInstruments.into()}>(StreamExecutor::new("executor with NO logs & NO metrics")).await;
671    }
672
673    #[cfg_attr(not(doc),tokio::test)]
674    async fn spawn_timeout_futures_fallible_executor_with_logs_and_metrics() {
675        assert_spawn_futures_fallible_executor::<{Instruments::LogsWithMetrics.into()}>(StreamExecutor::with_futures_timeout("executor with logs & metrics", Duration::from_millis(100))).await;
676    }
677
678    #[cfg_attr(not(doc),tokio::test)]
679    async fn spawn_timeout_futures_fallible_executor_with_metrics_and_no_logs() {
680        assert_spawn_futures_fallible_executor::<{Instruments::MetricsWithoutLogs.into()}>(StreamExecutor::with_futures_timeout("executor with metrics & NO logs", Duration::from_millis(100))).await;
681    }
682
683    #[cfg_attr(not(doc),tokio::test)]
684    async fn spawn_timeout_futures_fallible_executor_with_logs_and_no_metrics() {
685        assert_spawn_futures_fallible_executor::<{Instruments::LogsWithoutMetrics.into()}>(StreamExecutor::with_futures_timeout("executor with logs & NO metrics", Duration::from_millis(100))).await;
686    }
687
688    #[cfg_attr(not(doc),tokio::test)]
689    async fn spawn_timeout_futures_fallible_executor_with_no_logs_and_no_metrics() {
690        assert_spawn_futures_fallible_executor::<{Instruments::NoInstruments.into()}>(StreamExecutor::with_futures_timeout("executor with NO logs & NO metrics", Duration::from_millis(100))).await;
691    }
692
693    /// executes assertions on the given `executor` by spawning the executor to operate on streams yielding `Future<Result>` elements
694    async fn assert_spawn_futures_fallible_executor<const INSTRUMENTS:  usize>
695                                                   (executor: Arc<StreamExecutor<INSTRUMENTS>>) {
696
697        async fn to_future(item: Result<u32, Box<dyn Error + Send + Sync>>) -> Result<u32, Box<dyn Error + Send + Sync>> {
698            // OK items > 100 will sleep for a while -- intended to cause a timeout, provided the given executor is appropriately configured
699            if item.is_ok() && item.as_ref().unwrap() > &100 {
700                tokio::time::sleep(Duration::from_millis(150)).await;
701            }
702            item
703        }
704
705        let (tx, mut rx) = mpsc::channel::<bool>(10);
706        let error_counter = Arc::new(AtomicU32::new(0));
707        let error_counter_ref = Arc::clone(&error_counter);
708        let cloned_executor = Arc::clone(&executor);
709        let timeout_enabled = *executor.futures_timeout() > Duration::ZERO;
710        let expected_timeout_count = if timeout_enabled {2} else {0};
711        executor.spawn_executor::<_, _, _, _>
712                                 (1,
713                                  move |_| { let error_counter = Arc::clone(&error_counter_ref); async move {error_counter.fetch_add(1, Relaxed);} },
714                                  move |_| { let mut tx = tx.clone(); async move {tx.send(true).await.unwrap()} },
715                                  stream::iter(vec![to_future(Ok(17)),
716                                                              to_future(Err(Box::from("17"))),
717                                                              to_future(Ok(170)),     // times out
718                                                              to_future(Ok(19)),
719                                                              to_future(Err(Box::from("19"))),
720                                                              to_future(Ok(190))]     // times out
721                                  ));
722        assert!(rx.next().await.expect("consumption_done_reporter() wasn't called"), "consumption_done_reporter yielded the wrong value");
723        assert_eq!(error_counter.load(Relaxed), 2, "Error callback wasn't called the right number of times");
724        assert_metrics::<INSTRUMENTS>
725                        (cloned_executor, 4 - expected_timeout_count, expected_timeout_count, 2);
726
727    }
728
729    // non-fallible futures
730    ///////////////////////
731
732    #[cfg_attr(not(doc),tokio::test)]
733    async fn spawn_non_timeout_futures_executor_with_logs_and_metrics() {
734        assert_spawn_futures_executor::<{Instruments::LogsWithMetrics.into()}>(StreamExecutor::new("executor with logs & metrics")).await;
735    }
736
737    #[cfg_attr(not(doc),tokio::test)]
738    async fn spawn_non_timeout_futures_executor_with_metrics_and_no_logs() {
739        assert_spawn_futures_executor::<{Instruments::MetricsWithoutLogs.into()}>(StreamExecutor::new("executor with metrics & NO logs")).await;
740    }
741
742    #[cfg_attr(not(doc),tokio::test)]
743    async fn spawn_non_timeout_futures_executor_with_logs_and_no_metrics() {
744        assert_spawn_futures_executor::<{Instruments::LogsWithoutMetrics.into()}>(StreamExecutor::new("executor with logs & NO metrics")).await;
745    }
746
747    #[cfg_attr(not(doc),tokio::test)]
748    async fn spawn_non_timeout_futures_executor_with_no_logs_and_no_metrics() {
749        assert_spawn_futures_executor::<{Instruments::NoInstruments.into()}>(StreamExecutor::new("executor with NO logs & NO metrics")).await;
750    }
751
752    #[cfg_attr(not(doc),tokio::test)]
753    async fn spawn_timeout_futures_executor_with_logs_and_metrics() {
754        assert_spawn_futures_executor::<{Instruments::LogsWithMetrics.into()}>(StreamExecutor::with_futures_timeout("executor with logs & metrics", Duration::from_millis(100))).await;
755    }
756
757    #[cfg_attr(not(doc),tokio::test)]
758    async fn spawn_timeout_futures_executor_with_metrics_and_no_logs() {
759        assert_spawn_futures_executor::<{Instruments::MetricsWithoutLogs.into()}>(StreamExecutor::with_futures_timeout("executor with metrics & NO logs", Duration::from_millis(100))).await;
760    }
761
762    #[cfg_attr(not(doc),tokio::test)]
763    async fn spawn_timeout_futures_executor_with_logs_and_no_metrics() {
764        assert_spawn_futures_executor::<{Instruments::LogsWithoutMetrics.into()}>(StreamExecutor::with_futures_timeout("executor with logs & NO metrics", Duration::from_millis(100))).await;
765    }
766
767    #[cfg_attr(not(doc),tokio::test)]
768    async fn spawn_timeout_futures_executor_with_no_logs_and_no_metrics() {
769        assert_spawn_futures_executor::<{Instruments::NoInstruments.into()}>(StreamExecutor::with_futures_timeout("executor with NO logs & NO metrics", Duration::from_millis(100))).await;
770    }
771
772    /// executes assertions on the given `executor` by spawning the executor to operate on streams yielding `Future` elements
773    async fn assert_spawn_futures_executor<const INSTRUMENTS: usize>
774                                          (executor: Arc<StreamExecutor<INSTRUMENTS>>) {
775
776        async fn to_future(item: u32) -> u32 {
777            // OK items > 100 will sleep for a while -- intended to cause a timeout, provided the given executor is appropriately configured
778            if item > 100 {
779                tokio::time::sleep(Duration::from_millis(150)).await;
780            }
781            item
782        }
783
784        let (tx, mut rx) = mpsc::channel::<bool>(10);
785        let cloned_executor = Arc::clone(&executor);
786        let timeout_enabled = executor.futures_timeout > Duration::ZERO;
787        let expected_timeout_count = if timeout_enabled {2} else {0};
788        executor.spawn_futures_executor::<_, _, _>
789                                         (1,
790                                          move |_| { let mut tx = tx.clone(); async move {tx.send(true).await.unwrap()} },
791                                          stream::iter(vec![to_future(17),
792                                                                      to_future(170),     // times out
793                                                                      to_future(19),
794                                                                      to_future(190)]     // times out
795                                          ));
796        assert!(rx.next().await.expect("consumption_done_reporter() wasn't called"), "consumption_done_reporter yielded the wrong value");
797        assert_metrics::<INSTRUMENTS>
798                        (cloned_executor, 4 - expected_timeout_count, expected_timeout_count, 0);
799
800    }
801
802    // fallible (non-futures)
803    /////////////////////////
804
805    #[cfg_attr(not(doc),tokio::test)]
806    async fn spawn_fallibles_executor_with_logs_and_metrics() {
807        assert_spawn_fallibles_executor::<{Instruments::LogsWithMetrics.into()}>(StreamExecutor::new("executor with logs & metrics")).await;
808    }
809
810    #[cfg_attr(not(doc),tokio::test)]
811    async fn spawn_fallibles_executor_with_metrics_and_no_logs() {
812        assert_spawn_fallibles_executor::<{Instruments::MetricsWithoutLogs.into()}>(StreamExecutor::new("executor with metrics & NO logs")).await;
813    }
814
815    #[cfg_attr(not(doc),tokio::test)]
816    async fn spawn_fallibles_executor_with_logs_and_no_metrics() {
817        assert_spawn_fallibles_executor::<{Instruments::LogsWithoutMetrics.into()}>(StreamExecutor::new("executor with logs & NO metrics")).await;
818    }
819
820    #[cfg_attr(not(doc),tokio::test)]
821    async fn spawn_fallibles_executor_with_no_logs_and_no_metrics() {
822        assert_spawn_fallibles_executor::<{Instruments::NoInstruments.into()}>(StreamExecutor::new("executor with NO logs & NO metrics")).await;
823    }
824
825    /// executes assertions on the given `executor` by spawning the executor to operate on streams yielding fallible elements
826    async fn assert_spawn_fallibles_executor<const INSTRUMENTS: usize>
827                                            (executor: Arc<StreamExecutor<INSTRUMENTS>>) {
828        let error_count = Arc::new(AtomicU32::new(0));
829        let error_count_ref = Arc::clone(&error_count);
830        let (mut tx, mut rx) = mpsc::channel::<bool>(10);
831        let cloned_executor = Arc::clone(&executor);
832        executor.spawn_fallibles_executor::<_, _>
833                                           (1,
834                                            move |_err| {
835                                                error_count_ref.fetch_add(1, Relaxed);
836                                            },
837                                            move |_| async move {
838                                                tx.send(true).await.unwrap()
839                                            },
840                                            stream::iter(vec![Ok(17), Ok(19), Err(Box::from(String::from("Error on 20th")))]) );
841        assert!(rx.next().await.expect("consumption_done_reporter() wasn't called"), "consumption_done_reporter yielded the wrong value");
842        assert_eq!(error_count.load(Relaxed), 1, "Error count is wrong, as computed by the error callback");
843        assert_metrics::<INSTRUMENTS>
844                        (cloned_executor, 2, 0, 1);
845
846    }
847
848    // non-fallible & non-futures
849    /////////////////////////////
850
851    #[cfg_attr(not(doc),tokio::test)]
852    async fn spawn_non_futures_non_fallibles_executor_with_logs_and_metrics() {
853        assert_spawn_non_futures_non_fallibles_executor::<{Instruments::LogsWithMetrics.into()}>(StreamExecutor::new("executor with logs & metrics")).await;
854    }
855
856    #[cfg_attr(not(doc),tokio::test)]
857    async fn spawn_non_futures_non_fallibles_executor_with_metrics_and_no_logs() {
858        assert_spawn_non_futures_non_fallibles_executor::<{Instruments::MetricsWithoutLogs.into()}>(StreamExecutor::new("executor with metrics & NO logs")).await;
859    }
860
861    #[cfg_attr(not(doc),tokio::test)]
862    async fn spawn_non_futures_non_fallibles_executor_with_logs_and_no_metrics() {
863        assert_spawn_non_futures_non_fallibles_executor::<{Instruments::LogsWithoutMetrics.into()}>(StreamExecutor::new("executor with logs & NO metrics")).await;
864    }
865
866    #[cfg_attr(not(doc),tokio::test)]
867    async fn spawn_non_futures_non_fallibles_executor_with_no_logs_and_no_metrics() {
868        assert_spawn_non_futures_non_fallibles_executor::<{Instruments::NoInstruments.into()}>(StreamExecutor::new("executor with NO logs & NO metrics")).await;
869    }
870
871    /// executes assertions on the given `executor` by spawning the executor to operate on streams yielding non-futures / non-fallible elements
872    async fn assert_spawn_non_futures_non_fallibles_executor<const INSTRUMENTS:  usize>
873                                                            (executor: Arc<StreamExecutor<INSTRUMENTS>>) {
874        let (mut tx, mut rx) = mpsc::channel::<bool>(10);
875        let cloned_executor = Arc::clone(&executor);
876        executor.spawn_non_futures_non_fallibles_executor::<_, _>
877                                                         (1,
878                                                          move |_| async move {tx.send(true).await.unwrap()},
879                                                          stream::iter(vec![17, 19]) );
880        assert!(rx.next().await.expect("consumption_done_reporter() wasn't called"), "consumption_done_reporter yielded the wrong value");
881        assert_metrics::<INSTRUMENTS>
882                        (cloned_executor, 2, 0, 0);
883
884    }
885
886    // auxiliary functions
887    //////////////////////
888
889    /// apply assertions on metrics for the given `executor`
890    fn assert_metrics<const INSTRUMENTS: usize>
891                     (executor: Arc<StreamExecutor<INSTRUMENTS>>,
892                      expected_ok_counter:         u32,
893                      expected_timed_out_counter:  u32,
894                      expected_failed_counter:     u32) {
895
896        println!("### Stats assertions for Stream pipeline executor named '{}' (Logs? {}; Metrics? {}) ####",
897                 executor.executor_name, Instruments::from(INSTRUMENTS).logging(), Instruments::from(INSTRUMENTS).metrics());
898        let creation_duration = executor.creation_time.elapsed();
899        let execution_start_delta_nanos = executor.execution_start_delta_nanos.load(Relaxed);
900        let execution_finish_delta_nanos = executor.execution_finish_delta_nanos.load(Relaxed);
901        let (ok_counter, ok_average) = executor.ok_events_avg_future_duration.lightweight_probe();
902        let (timed_out_counter, timed_out_average) = executor.timed_out_events_avg_future_duration.lightweight_probe();
903        let (failed_counter, failed_average) = executor.failed_events_avg_future_duration.lightweight_probe();
904        println!("Creation time:    {:?} ago", creation_duration);
905        println!("Execution Start:  {:?} after creation", Duration::from_nanos(execution_start_delta_nanos));
906        println!("Execution Finish: {:?} after creation", Duration::from_nanos(execution_finish_delta_nanos));
907        println!("OK elements count: {ok_counter}; OK elements average Future resolution time: {ok_average}s{}{}",
908                 if Instruments::from(INSTRUMENTS).metrics() {""} else {" -- metrics are DISABLED"},
909                 if Instruments::from(INSTRUMENTS).logging() {" -- verify these values against the \"executor closed\" message"} else {" -- logs are DISABLED"});
910        println!("TIMED OUT elements count: {timed_out_counter}; TIMED OUT elements average Future resolution time: {timed_out_average}s{}{}",
911                 if Instruments::from(INSTRUMENTS).metrics() {""} else {" -- metrics are DISABLED"},
912                 if Instruments::from(INSTRUMENTS).logging() {" -- verify these values against the \"executor closed\" message"} else {" -- logs are DISABLED"});
913        println!("FAILED elements count: {failed_counter}; FAILED elements average Future resolution time: {failed_average}s{}{}",
914                 if Instruments::from(INSTRUMENTS).metrics() {""} else {" -- metrics are DISABLED"},
915                 if Instruments::from(INSTRUMENTS).logging() {" -- verify these values against the \"executor closed\" message"} else {" -- logs are DISABLED"});
916
917        assert_ne!(execution_start_delta_nanos,  u64::MAX, "'execution_start_delta_nanos' wasn't set");
918        assert_ne!(execution_finish_delta_nanos, u64::MAX, "'execution_finish_delta_nanos' wasn't set");
919        assert!(execution_finish_delta_nanos >= execution_start_delta_nanos, "INSTRUMENTATION ERROR: 'execution_start_delta_nanos' was set after 'execution_finish_delta_nanos'");
920
921        if Instruments::from(INSTRUMENTS).metrics() {
922            assert_eq!(ok_counter,        expected_ok_counter,        "OK elements counter doesn't match -- Metrics are ENABLED");
923            assert_eq!(timed_out_counter, expected_timed_out_counter, "TIMED OUT elements counter doesn't match -- Metrics are ENABLED");
924            assert_eq!(failed_counter,    expected_failed_counter,    "FAILED elements counter doesn't match -- Metrics are ENABLED");
925        } else {
926            assert_eq!(ok_counter,        0, "Metrics are DISABLED, so the reported OK elements should be ZERO");
927            assert_eq!(timed_out_counter, 0, "Metrics are DISABLED, so the reported TIMED OUT elements should be ZERO");
928            assert_eq!(failed_counter,    0, "Metrics are DISABLED, so the reported FAILED elements should be ZERO");
929        }
930
931
932        // if executor.instruments.measure_time is set
933        // ...
934    }
935}
936
937/// will derive `AtomicExecutorStatus`
938#[atomic_enum]
939#[derive(PartialEq)]
940pub enum ExecutorStatus {
941    NotStarted,
942    Running,
943    ScheduledToFinish,
944    ProgrammaticallyEnded,
945    StreamEnded,
946}
reactive_mutiny/stream_executor.rs

reactive_mutiny/
stream_executor.rs
