4293824813

Build Type

push

github

Committed by GitHub

Commit Message

feat: Read data from postgres snapshot in parallel (#1094)

Run Details

70 of 70 new or added lines in 1 file covered. (100.0%)

28883 of 39260 relevant lines covered (73.57%)

53914.72 hits per line

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

99.06

/dozer-core/src/executor.rs

use crate::builder_dag::{BuilderDag, NodeKind};
use crate::dag_metadata::DagMetadata;
use crate::dag_schemas::DagSchemas;
use crate::errors::ExecutionError;
use crate::Dag;

use daggy::petgraph::visit::IntoNodeIdentifiers;
use dozer_types::node::NodeHandle;
use dozer_types::types::Operation;

use crate::epoch::Epoch;
use std::collections::hash_map::Entry;
use std::collections::HashMap;
use std::fmt::Debug;
use std::panic::panic_any;
use std::path::PathBuf;
use std::sync::atomic::AtomicBool;
use std::sync::Arc;
use std::thread::JoinHandle;
use std::thread::{self, Builder};
use std::time::Duration;

#[derive(Clone)]
pub struct ExecutorOptions {
    pub commit_sz: u32,
    pub channel_buffer_sz: usize,
    pub commit_time_threshold: Duration,

    pub max_map_size: usize,
}

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

#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) enum InputPortState {
    Open,
    Terminated,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ExecutorOperation {
    Op { op: Operation },
    Commit { epoch: Epoch },
    Terminate,
}

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: HashMap<NodeHandle, JoinHandle<()>>,
}

impl DagExecutor {
    pub fn new<T: Clone + Debug>(
        dag: &Dag<T>,
        path: PathBuf,
        options: ExecutorOptions,
    ) -> Result<Self, ExecutionError> {
        let dag_schemas = DagSchemas::new(dag)?;
        let builder_dag = BuilderDag::new(&dag_schemas, path, options.max_map_size)?;

        Ok(Self {
            builder_dag,
            options,
        })
    }

    pub fn validate<T: Clone + Debug>(dag: &Dag<T>, path: PathBuf) -> Result<(), ExecutionError> {
        let dag_schemas = DagSchemas::new(dag)?;
        DagMetadata::new(&dag_schemas, path)?;
        Ok(())
    }

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

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

        Ok(DagExecutorJoinHandle { join_handles })
    }
}

impl DagExecutorJoinHandle {
    pub fn join(mut self) -> Result<(), ExecutionError> {
        let handles: Vec<NodeHandle> = self.join_handles.iter().map(|e| e.0.clone()).collect();

        loop {
            for handle in &handles {
                if let Entry::Occupied(entry) = self.join_handles.entry(handle.clone()) {
                    if entry.get().is_finished() {
                        if let Err(e) = entry.remove().join() {
                            panic_any(e);
                        }
                    }
                }
            }

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

            thread::sleep(Duration::from_millis(250));
        }
    }
}

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

    let _st_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(ExecutionError::CannotSendToChannel) => {}
            // Other errors result in panic.
            Err(e) => std::panic::panic_any(e),
        })?;

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

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

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

1	use crate::builder_dag::{BuilderDag, NodeKind};
2	use crate::dag_metadata::DagMetadata;
3	use crate::dag_schemas::DagSchemas;
4	use crate::errors::ExecutionError;
5	use crate::Dag;
6
7	use daggy::petgraph::visit::IntoNodeIdentifiers;
8	use dozer_types::node::NodeHandle;
9	use dozer_types::types::Operation;
10
11	use crate::epoch::Epoch;
12	use std::collections::hash_map::Entry;
13	use std::collections::HashMap;
14	use std::fmt::Debug;
15	use std::panic::panic_any;
16	use std::path::PathBuf;
17	use std::sync::atomic::AtomicBool;
18	use std::sync::Arc;
19	use std::thread::JoinHandle;
20	use std::thread::{self, Builder};
21	use std::time::Duration;
22
23	#[derive(Clone)]	×
24	pub struct ExecutorOptions {
25	pub commit_sz: u32,
26	pub channel_buffer_sz: usize,
27	pub commit_time_threshold: Duration,
28
29	pub max_map_size: usize,
30	}
31
32	impl Default for ExecutorOptions {
33	fn default() -> Self {	330✔
34	Self {	330✔
35	commit_sz: 10_000,	330✔
36	channel_buffer_sz: 20_000,	330✔
37	commit_time_threshold: Duration::from_millis(50),	330✔
38	max_map_size: 1024 * 1024 * 1024 * 1024,	330✔
39	}	330✔
40	}	330✔
41	}
42
43	#[derive(Clone, Debug, PartialEq, Eq)]	1,658✔
44	pub(crate) enum InputPortState {
45	Open,
46	Terminated,
47	}
48
49	#[derive(Clone, Debug, PartialEq, Eq)]	84,996✔
50	pub enum ExecutorOperation {
51	Op { op: Operation },
52	Commit { epoch: Epoch },
53	Terminate,
54	}
55
56	mod execution_dag;
57	mod name;
58	mod node;
59	mod processor_node;
60	mod receiver_loop;
61	mod sink_node;
62	mod source_node;
63
64	use node::Node;
65	use processor_node::ProcessorNode;
66	use sink_node::SinkNode;
67
68	use self::execution_dag::ExecutionDag;
69	use self::source_node::{create_source_nodes, SourceListenerNode, SourceSenderNode};
70
71	pub struct DagExecutor {
72	builder_dag: BuilderDag,
73	options: ExecutorOptions,
74	}
75
76	pub struct DagExecutorJoinHandle {
77	join_handles: HashMap<NodeHandle, JoinHandle<()>>,
78	}
79
80	impl DagExecutor {
81	pub fn new<T: Clone + Debug>(	223✔
82	dag: &Dag<T>,	223✔
83	path: PathBuf,	223✔
84	options: ExecutorOptions,	223✔
85	) -> Result<Self, ExecutionError> {	223✔
86	let dag_schemas = DagSchemas::new(dag)?;	223✔
87	let builder_dag = BuilderDag::new(&dag_schemas, path, options.max_map_size)?;	223✔
88
89	Ok(Self {	219✔
90	builder_dag,	219✔
91	options,	219✔
92	})	219✔
93	}	221✔
94
95	pub fn validate<T: Clone + Debug>(dag: &Dag<T>, path: PathBuf) -> Result<(), ExecutionError> {	8✔
96	let dag_schemas = DagSchemas::new(dag)?;	8✔
97	DagMetadata::new(&dag_schemas, path)?;	8✔
98	Ok(())	8✔
99	}	8✔
100
101	pub fn start(self, running: Arc<AtomicBool>) -> Result<DagExecutorJoinHandle, ExecutionError> {	332✔
102	// Construct execution dag.
103	let mut execution_dag =	332✔
104	ExecutionDag::new(self.builder_dag, self.options.channel_buffer_sz)?;	332✔
105	let node_indexes = execution_dag.graph().node_identifiers().collect::<Vec<_>>();	332✔
106		332✔
107	// Start the threads.	332✔
108	let mut join_handles = HashMap::new();	332✔
109	for node_index in node_indexes {	2,027✔
110	let node = &execution_dag.graph()[node_index];	1,695✔
111	let node_handle = node.handle.clone();	1,695✔
112	match &node.kind.as_ref().expect("We created all nodes") {	1,695✔
113	NodeKind::Source(_, _) => {
114	let (source_sender_node, source_listener_node) = create_source_nodes(	342✔
115	&mut execution_dag,	342✔
116	node_index,	342✔
117	&self.options,	342✔
118	running.clone(),	342✔
119	);	342✔
120	join_handles.insert(	342✔
121	node_handle,	342✔
122	start_source(source_sender_node, source_listener_node)?,	342✔
123	);
124	}
125	NodeKind::Processor(_) => {
126	let processor_node = ProcessorNode::new(&mut execution_dag, node_index);	1,017✔
127	join_handles.insert(node_handle, start_processor(processor_node)?);	1,017✔
128	}
129	NodeKind::Sink(_) => {
130	let sink_node = SinkNode::new(&mut execution_dag, node_index);	336✔
131	join_handles.insert(node_handle, start_sink(sink_node)?);	336✔
132	}
133	}
134	}
135
136	Ok(DagExecutorJoinHandle { join_handles })	332✔
137	}	332✔
138	}
139
140	impl DagExecutorJoinHandle {
141	pub fn join(mut self) -> Result<(), ExecutionError> {	331✔
142	let handles: Vec<NodeHandle> = self.join_handles.iter().map(\|e\| e.0.clone()).collect();	1,692✔
143
144	loop {
145	for handle in &handles {	8,638✔
146	if let Entry::Occupied(entry) = self.join_handles.entry(handle.clone()) {	7,190✔
147	if entry.get().is_finished() {	7,046✔
148	if let Err(e) = entry.remove().join() {	1,680✔
149	panic_any(e);	5✔
150	}	1,675✔
151	}	5,366✔
152	}	144✔
153	}
154
155	if self.join_handles.is_empty() {	1,448✔
156	return Ok(());	326✔
157	}	1,122✔
158		1,122✔
159	thread::sleep(Duration::from_millis(250));	1,122✔
160	}
161	}	326✔
162	}
163
164	fn start_source(	342✔
165	source_sender: SourceSenderNode,	342✔
166	source_listener: SourceListenerNode,	342✔
167	) -> Result<JoinHandle<()>, ExecutionError> {	342✔
168	let handle = source_sender.handle().clone();	342✔
169
170	let _st_handle = Builder::new()	342✔
171	.name(format!("{handle}-sender"))	342✔
172	.spawn(move \|\| match source_sender.run() {	342✔
173	Ok(_) => {}	335✔
174	// Channel disconnection means the source listener has quit.
175	// Maybe it quit gracefully so we don't need to panic.
176	Err(ExecutionError::CannotSendToChannel) => {}	6✔
177	// Other errors result in panic.
178	Err(e) => std::panic::panic_any(e),	1✔
179	})?;	342✔
180
181	Ok(Builder::new()	342✔
182	.name(format!("{handle}-listener"))	342✔
183	.spawn(move \|\| {	342✔
184	if let Err(e) = source_listener.run() {	342✔
185	std::panic::panic_any(e);	5✔
186	}	337✔
187	})?)	342✔
188	}	342✔
189
190	fn start_processor(processor: ProcessorNode) -> Result<JoinHandle<()>, ExecutionError> {	1,017✔
191	Ok(Builder::new()	1,017✔
192	.name(processor.handle().to_string())	1,017✔
193	.spawn(move \|\| {	1,017✔
194	if let Err(e) = processor.run() {	1,017✔
195	std::panic::panic_any(e);	7✔
196	}	1,010✔
197	})?)	1,017✔
198	}	1,017✔
199
200	fn start_sink(sink: SinkNode) -> Result<JoinHandle<()>, ExecutionError> {	336✔
201	Ok(Builder::new().name(sink.handle().to_string()).spawn(\|\| {	336✔
202	if let Err(e) = sink.run() {	336✔
203	std::panic::panic_any(e);	5✔
204	}	331✔
205	})?)	336✔
206	}	336✔

getdozer / dozer / 4293824813

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