6105410942

Committed 07 Sep 2023 04:28AM UTC coverage: 77.562% (-0.1%) from 77.686%

Build # 6105410942

Build Type

push

github

Committed by

chloeminkyung

Commit Message

feat: onnx image

Run Details

1141 of 1141 new or added lines in 66 files covered. (100.0%)

49957 of 64409 relevant lines covered (77.56%)

50900.25 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.62

/dozer-core/src/executor/mod.rs

use crate::builder_dag::{BuilderDag, NodeKind};
use crate::checkpoint::{CheckpointFactory, OptionCheckpoint};
use crate::dag_schemas::DagSchemas;
use crate::errors::ExecutionError;
use crate::Dag;

use daggy::petgraph::visit::IntoNodeIdentifiers;

use dozer_tracing::LabelsAndProgress;
use dozer_types::serde::{self, Deserialize, Serialize};
use std::fmt::Debug;
use std::panic::panic_any;
use std::sync::atomic::AtomicBool;
use std::sync::Arc;
use std::thread::JoinHandle;
use std::thread::{self, Builder};
use std::time::Duration;

#[derive(Debug, Clone)]
pub struct ExecutorOptions {
    pub commit_sz: u32,
    pub channel_buffer_sz: usize,
    pub commit_time_threshold: Duration,
    pub error_threshold: Option<u32>,
}

impl Default for ExecutorOptions {
    fn default() -> Self {
        Self {
            commit_sz: 10_000,
            channel_buffer_sz: 20_000,
            commit_time_threshold: Duration::from_millis(50),
            error_threshold: Some(0),
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(crate = "self::serde")]
pub(crate) enum InputPortState {
    Open,
    Terminated,
}

mod execution_dag;
mod name;
mod node;
mod processor_node;
mod receiver_loop;
mod sink_node;
mod source_node;

use node::Node;
use processor_node::ProcessorNode;
use sink_node::SinkNode;

use self::execution_dag::ExecutionDag;
use self::source_node::{create_source_nodes, SourceListenerNode, SourceSenderNode};

pub struct DagExecutor {
    builder_dag: BuilderDag,
    options: ExecutorOptions,
}

pub struct DagExecutorJoinHandle {
    join_handles: Vec<JoinHandle<()>>,
}

impl DagExecutor {
    pub async fn new(
        dag: Dag,
        checkpoint_factory: Arc<CheckpointFactory>,
        checkpoint: OptionCheckpoint,
        options: ExecutorOptions,
    ) -> Result<Self, ExecutionError> {
        let dag_schemas = DagSchemas::new(dag)?;

        let builder_dag = BuilderDag::new(checkpoint_factory, checkpoint, dag_schemas).await?;

        Ok(Self {
            builder_dag,
            options,
        })
    }

    pub fn validate<T: Clone + Debug>(dag: Dag) -> Result<(), ExecutionError> {
        DagSchemas::new(dag)?;
        Ok(())
    }

    pub fn start(
        self,
        running: Arc<AtomicBool>,
        labels: LabelsAndProgress,
    ) -> Result<DagExecutorJoinHandle, ExecutionError> {
        // Construct execution dag.
        let initial_epoch_id = self.builder_dag.initial_epoch_id();
        let mut execution_dag = ExecutionDag::new(
            self.builder_dag,
            labels,
            self.options.channel_buffer_sz,
            self.options.error_threshold,
        )?;
        let node_indexes = execution_dag.graph().node_identifiers().collect::<Vec<_>>();

        // Start the threads.
        let mut join_handles = Vec::new();
        for node_index in node_indexes {
            let node = execution_dag.graph()[node_index]
                .as_ref()
                .expect("We created all nodes");
            match &node.kind {
                NodeKind::Source(_, _) => {
                    let (source_sender_node, source_listener_node) = create_source_nodes(
                        &mut execution_dag,
                        node_index,
                        &self.options,
                        running.clone(),
                    );
                    let (sender, receiver) =
                        start_source(source_sender_node, source_listener_node)?;
                    join_handles.extend([sender, receiver]);
                }
                NodeKind::Processor(_) => {
                    let processor_node =
                        ProcessorNode::new(&mut execution_dag, node_index, initial_epoch_id);
                    join_handles.push(start_processor(processor_node)?);
                }
                NodeKind::Sink(_) => {
                    let sink_node = SinkNode::new(&mut execution_dag, node_index, initial_epoch_id);
                    join_handles.push(start_sink(sink_node)?);
                }
            }
        }

        Ok(DagExecutorJoinHandle { join_handles })
    }
}

impl DagExecutorJoinHandle {
    pub fn join(mut self) -> Result<(), ExecutionError> {
        loop {
            let Some(finished) = self
                .join_handles
                .iter()
                .enumerate()
                .find_map(|(i, handle)| handle.is_finished().then_some(i))
            else {
                thread::sleep(Duration::from_millis(250));

                continue;
            };
            let handle = self.join_handles.swap_remove(finished);
            if let Err(e) = handle.join() {
                panic_any(e)
            }

            if self.join_handles.is_empty() {
                return Ok(());
            }
        }
    }
}

fn start_source(
    source_sender: SourceSenderNode,
    source_listener: SourceListenerNode,
) -> Result<(JoinHandle<()>, JoinHandle<()>), ExecutionError> {
    let handle = source_sender.handle().clone();

    let sender_handle = Builder::new()
        .name(format!("{handle}-sender"))
        .spawn(move || match source_sender.run() {
            Ok(_) => {}
            // Channel disconnection means the source listener has quit.
            // Maybe it quit gracefully so we don't need to panic.
            Err(e) => {
                if let ExecutionError::Source(e) = &e {
                    if let Some(ExecutionError::CannotSendToChannel) = e.downcast_ref() {
                        return;
                    }
                }
                std::panic::panic_any(e);
            }
        })
        .map_err(ExecutionError::CannotSpawnWorkerThread)?;

    let listener_handle = Builder::new()
        .name(format!("{handle}-listener"))
        .spawn(move || {
            if let Err(e) = source_listener.run() {
                std::panic::panic_any(e);
            }
        })
        .map_err(ExecutionError::CannotSpawnWorkerThread)?;

    Ok((sender_handle, listener_handle))
}

fn start_processor(processor: ProcessorNode) -> Result<JoinHandle<()>, ExecutionError> {
    Builder::new()
        .name(processor.handle().to_string())
        .spawn(move || {
            if let Err(e) = processor.run() {
                std::panic::panic_any(e);
            }
        })
        .map_err(ExecutionError::CannotSpawnWorkerThread)
}

fn start_sink(sink: SinkNode) -> Result<JoinHandle<()>, ExecutionError> {
    Builder::new()
        .name(sink.handle().to_string())
        .spawn(|| {
            if let Err(e) = sink.run() {
                std::panic::panic_any(e);
            }
        })
        .map_err(ExecutionError::CannotSpawnWorkerThread)
}

1	use crate::builder_dag::{BuilderDag, NodeKind};
2	use crate::checkpoint::{CheckpointFactory, OptionCheckpoint};
3	use crate::dag_schemas::DagSchemas;
4	use crate::errors::ExecutionError;
5	use crate::Dag;
6
7	use daggy::petgraph::visit::IntoNodeIdentifiers;
8
9	use dozer_tracing::LabelsAndProgress;
10	use dozer_types::serde::{self, Deserialize, Serialize};
11	use std::fmt::Debug;
12	use std::panic::panic_any;
13	use std::sync::atomic::AtomicBool;
14	use std::sync::Arc;
15	use std::thread::JoinHandle;
16	use std::thread::{self, Builder};
17	use std::time::Duration;
18
19	#[derive(Debug, Clone)]	×
20	pub struct ExecutorOptions {
21	pub commit_sz: u32,
22	pub channel_buffer_sz: usize,
23	pub commit_time_threshold: Duration,
24	pub error_threshold: Option<u32>,
25	}
26
27	impl Default for ExecutorOptions {
28	fn default() -> Self {	585✔
29	Self {	585✔
30	commit_sz: 10_000,	585✔
31	channel_buffer_sz: 20_000,	585✔
32	commit_time_threshold: Duration::from_millis(50),	585✔
33	error_threshold: Some(0),	585✔
34	}	585✔
35	}	585✔
36	}
37
38	#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]	3,215✔
39	#[serde(crate = "self::serde")]
40	pub(crate) enum InputPortState {
41	Open,
42	Terminated,
43	}
44
45	mod execution_dag;
46	mod name;
47	mod node;
48	mod processor_node;
49	mod receiver_loop;
50	mod sink_node;
51	mod source_node;
52
53	use node::Node;
54	use processor_node::ProcessorNode;
55	use sink_node::SinkNode;
56
57	use self::execution_dag::ExecutionDag;
58	use self::source_node::{create_source_nodes, SourceListenerNode, SourceSenderNode};
59
60	pub struct DagExecutor {
61	builder_dag: BuilderDag,
62	options: ExecutorOptions,
63	}
64
65	pub struct DagExecutorJoinHandle {
66	join_handles: Vec<JoinHandle<()>>,
67	}
68
69	impl DagExecutor {
70	pub async fn new(	633✔
71	dag: Dag,	633✔
72	checkpoint_factory: Arc<CheckpointFactory>,	633✔
73	checkpoint: OptionCheckpoint,	633✔
74	options: ExecutorOptions,	633✔
75	) -> Result<Self, ExecutionError> {	633✔
76	let dag_schemas = DagSchemas::new(dag)?;	118✔
77
78	let builder_dag = BuilderDag::new(checkpoint_factory, checkpoint, dag_schemas).await?;	118✔
79
80	Ok(Self {	114✔
81	builder_dag,	114✔
82	options,	114✔
83	})	114✔
84	}	116✔
85
86	pub fn validate<T: Clone + Debug>(dag: Dag) -> Result<(), ExecutionError> {
87	DagSchemas::new(dag)?;	×
88	Ok(())	×
89	}	×
90
91	pub fn start(	629✔
92	self,	629✔
93	running: Arc<AtomicBool>,	629✔
94	labels: LabelsAndProgress,	629✔
95	) -> Result<DagExecutorJoinHandle, ExecutionError> {	629✔
96	// Construct execution dag.	629✔
97	let initial_epoch_id = self.builder_dag.initial_epoch_id();	629✔
98	let mut execution_dag = ExecutionDag::new(	629✔
99	self.builder_dag,	629✔
100	labels,	629✔
101	self.options.channel_buffer_sz,	629✔
102	self.options.error_threshold,	629✔
103	)?;	629✔
104	let node_indexes = execution_dag.graph().node_identifiers().collect::<Vec<_>>();	629✔
105		629✔
106	// Start the threads.	629✔
107	let mut join_handles = Vec::new();	629✔
108	for node_index in node_indexes {	3,661✔
109	let node = execution_dag.graph()[node_index]	3,032✔
110	.as_ref()	3,032✔
111	.expect("We created all nodes");	3,032✔
112	match &node.kind {	3,032✔
113	NodeKind::Source(_, _) => {
114	let (source_sender_node, source_listener_node) = create_source_nodes(	636✔
115	&mut execution_dag,	636✔
116	node_index,	636✔
117	&self.options,	636✔
118	running.clone(),	636✔
119	);	636✔
120	let (sender, receiver) =	636✔
121	start_source(source_sender_node, source_listener_node)?;	636✔
122	join_handles.extend([sender, receiver]);	636✔
123	}
124	NodeKind::Processor(_) => {
125	let processor_node =	1,763✔
126	ProcessorNode::new(&mut execution_dag, node_index, initial_epoch_id);	1,763✔
127	join_handles.push(start_processor(processor_node)?);	1,763✔
128	}
129	NodeKind::Sink(_) => {
130	let sink_node = SinkNode::new(&mut execution_dag, node_index, initial_epoch_id);	633✔
131	join_handles.push(start_sink(sink_node)?);	633✔
132	}
133	}
134	}
135
136	Ok(DagExecutorJoinHandle { join_handles })	629✔
137	}	629✔
138	}
139
140	impl DagExecutorJoinHandle {
141	pub fn join(mut self) -> Result<(), ExecutionError> {	628✔
142	loop {
143	let Some(finished) = self	4,605✔
144	.join_handles	4,605✔
145	.iter()	4,605✔
146	.enumerate()	4,605✔
147	.find_map(\|(i, handle)\| handle.is_finished().then_some(i))	8,738✔
148	else {
149	thread::sleep(Duration::from_millis(250));	953✔
150		953✔
151	continue;	953✔
152	};
153	let handle = self.join_handles.swap_remove(finished);	3,652✔
154	if let Err(e) = handle.join() {	3,652✔
155	panic_any(e)	5✔
156	}	3,647✔
157		3,647✔
158	if self.join_handles.is_empty() {	3,647✔
159	return Ok(());	623✔
160	}	3,024✔
161	}
162	}	623✔
163	}
164
165	fn start_source(	636✔
166	source_sender: SourceSenderNode,	636✔
167	source_listener: SourceListenerNode,	636✔
168	) -> Result<(JoinHandle<()>, JoinHandle<()>), ExecutionError> {	636✔
169	let handle = source_sender.handle().clone();	636✔
170
171	let sender_handle = Builder::new()	636✔
172	.name(format!("{handle}-sender"))	636✔
173	.spawn(move \|\| match source_sender.run() {	636✔
174	Ok(_) => {}	629✔
175	// Channel disconnection means the source listener has quit.
176	// Maybe it quit gracefully so we don't need to panic.
177	Err(e) => {	7✔
178	if let ExecutionError::Source(e) = &e {	7✔
179	if let Some(ExecutionError::CannotSendToChannel) = e.downcast_ref() {	7✔
180	return;	7✔
181	}	×
182	}	×
183	std::panic::panic_any(e);	×
184	}
185	})	636✔
186	.map_err(ExecutionError::CannotSpawnWorkerThread)?;	636✔
187
188	let listener_handle = Builder::new()	636✔
189	.name(format!("{handle}-listener"))	636✔
190	.spawn(move \|\| {	636✔
191	if let Err(e) = source_listener.run() {	636✔
192	std::panic::panic_any(e);	6✔
193	}	630✔
194	})	636✔
195	.map_err(ExecutionError::CannotSpawnWorkerThread)?;	636✔
196
197	Ok((sender_handle, listener_handle))	636✔
198	}	636✔
199
200	fn start_processor(processor: ProcessorNode) -> Result<JoinHandle<()>, ExecutionError> {	1,763✔
201	Builder::new()	1,763✔
202	.name(processor.handle().to_string())	1,763✔
203	.spawn(move \|\| {	1,763✔
204	if let Err(e) = processor.run() {	1,763✔
205	std::panic::panic_any(e);	8✔
206	}	1,755✔
207	})	1,763✔
208	.map_err(ExecutionError::CannotSpawnWorkerThread)	1,763✔
209	}	1,763✔
210
211	fn start_sink(sink: SinkNode) -> Result<JoinHandle<()>, ExecutionError> {	633✔
212	Builder::new()	633✔
213	.name(sink.handle().to_string())	633✔
214	.spawn(\|\| {	633✔
215	if let Err(e) = sink.run() {	633✔
216	std::panic::panic_any(e);	6✔
217	}	627✔
218	})	633✔
219	.map_err(ExecutionError::CannotSpawnWorkerThread)	633✔
220	}	633✔

getdozer / dozer / 6105410942

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous