gmt_dos_actors/actor/
task.rs

1use std::any::type_name;
2
3use async_trait::async_trait;
4use interface::TryUpdate;
5
6use crate::framework::model::{Task, TaskError};
7
8use super::{Actor, PlainActor};
9
10type Result<T> = std::result::Result<T, TaskError>;
11
12#[async_trait]
13impl<C, const NI: usize, const NO: usize> Task for Actor<C, NI, NO>
14where
15    C: 'static + TryUpdate,
16{
17    /// Run the actor loop
18    async fn task(mut self: Box<Self>) -> Result<()> {
19        /*         match self.bootstrap().await {
20            Err(e) => crate::print_info(
21                format!("{} bootstrapping failed", Who::highlight(self)),
22                Some(&e),
23            ),
24            Ok(_) => {
25                crate::print_info(
26                    format!("{} loop started", Who::highlight(self)),
27                    None::<&dyn std::error::Error>,
28                );
29                if let Err(e) = self.async_run().await {
30                    println!(
31                        "{}{:?}",
32                        format!("{} loop ended", Who::highlight(self)),
33                        Some(&e)
34                    );
35                }
36            }
37        } */
38        self.async_run().await
39    }
40
41    /// Starts the actor infinite loop
42    async fn async_run(&mut self) -> Result<()> {
43        log::debug!("ACTOR LOOP ({NI}/{NO}): {}", type_name::<C>());
44        let bootstrap = self.bootstrap().await?;
45        match (self.inputs.as_ref(), self.outputs.as_ref()) {
46            (Some(_), Some(_)) => {
47                if NO >= NI {
48                    // Decimation
49                    if !bootstrap {
50                        // bootstrap is applied when decimation is used
51                        // in conjunction with averaging
52                        // When averaging there is a delay of `NO` samples
53                        // to account for the time to iterate and a default
54                        // values is used for the 1st output
55                        // For decimation of the input signal there is no delay
56                        // and the 1st sample goes through unimpeded
57                        self.collect()
58                            .await?
59                            .client
60                            .lock()
61                            .await
62                            .boxed_try_update()?;
63                        self.distribute().await?;
64                    }
65                    loop {
66                        for _ in 0..NO / NI {
67                            self.collect()
68                                .await?
69                                .client
70                                .lock()
71                                .await
72                                .boxed_try_update()?;
73                        }
74                        self.distribute().await?;
75                    }
76                } else {
77                    // Upsampling
78                    loop {
79                        self.collect()
80                            .await?
81                            .client
82                            .lock()
83                            .await
84                            .boxed_try_update()?;
85                        for _ in 0..NI / NO {
86                            self.distribute().await?;
87                        }
88                    }
89                }
90            }
91            (None, Some(_)) => {
92                // Initiator
93                tokio::task::yield_now().await; // at least cooperates with other tasks
94                loop {
95                    self.client.lock().await.boxed_try_update()?;
96                    self.distribute().await?;
97                }
98            }
99            (Some(_), None) => loop {
100                // Terminator
101                self.collect()
102                    .await?
103                    .client
104                    .lock()
105                    .await
106                    .boxed_try_update()?;
107            },
108            (None, None) => Ok(()),
109        }
110    }
111
112    fn as_plain(&self) -> PlainActor {
113        self.into()
114    }
115    fn name(&self) -> &'static str {
116        type_name::<C>()
117    }
118}
gmt_dos_actors/actor/task.rs

gmt_dos_actors/actor/
task.rs
