3974044832

Build Type

github

Committed by GitHub

Commit Message

Merge 6f30ce3b1 into 9fa836726

Pull Request Pull Request #703: fix: forbid duplicated cte names

Run Details

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

22298 of 33673 relevant lines covered (66.22%)

35756.78 hits per line

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

90.91

/dozer-core/src/dag/executor/source_node.rs

use std::{
    collections::HashMap,
    path::Path,
    sync::{
        atomic::{AtomicBool, Ordering},
        Arc,
    },
    time::Duration,
};

use crossbeam::channel::{Receiver, RecvTimeoutError, Sender};
use dozer_types::log::debug;
use dozer_types::{
    internal_err,
    parking_lot::RwLock,
    types::{Operation, Schema},
};

use crate::dag::{
    channels::SourceChannelForwarder,
    dag::Edge,
    epoch::EpochManager,
    errors::ExecutionError::{self, InternalError},
    executor_utils::{create_ports_databases_and_fill_downstream_record_readers, init_component},
    forwarder::{SourceChannelManager, StateWriter},
    node::{NodeHandle, OutputPortDef, PortHandle, Source, SourceFactory},
    record_store::RecordReader,
};

use super::{node::Node, ExecutorOperation};

#[derive(Debug)]
struct InternalChannelSourceForwarder {
    sender: Sender<(PortHandle, u64, u64, Operation)>,
}

impl InternalChannelSourceForwarder {
    pub fn new(sender: Sender<(PortHandle, u64, u64, Operation)>) -> Self {
        Self { sender }
    }
}

impl SourceChannelForwarder for InternalChannelSourceForwarder {
    fn send(
        &mut self,
        txid: u64,
        seq_in_tx: u64,
        op: Operation,
        port: PortHandle,
    ) -> Result<(), ExecutionError> {
        internal_err!(self.sender.send((port, txid, seq_in_tx, op)))
    }
}

/// The sender half of a source in the execution DAG.
#[derive(Debug)]
pub struct SourceSenderNode {
    /// Node handle in description DAG.
    node_handle: NodeHandle,
    /// The source.
    source: Box<dyn Source>,
    /// Last checkpointed output data sequence number.
    last_checkpoint: (u64, u64),
    /// The forwarder that will be passed to the source for outputig data.
    forwarder: InternalChannelSourceForwarder,
    /// If the execution DAG should be running. Used for terminating the execution DAG.
    running: Arc<AtomicBool>,
}

impl SourceSenderNode {
    /// # Arguments
    ///
    /// - `node_handle`: Node handle in description DAG.
    /// - `source_factory`: Source factory in description DAG.
    /// - `output_schemas`: Output data schemas.
    /// - `last_checkpoint`: Last checkpointed output of this source.
    /// - `sender`: Channel to send data to.
    /// - `running`: If the execution DAG should still be running.
    pub fn new(
        node_handle: NodeHandle,
        source_factory: &dyn SourceFactory,
        output_schemas: HashMap<PortHandle, Schema>,
        last_checkpoint: (u64, u64),
        sender: Sender<(PortHandle, u64, u64, Operation)>,
        running: Arc<AtomicBool>,
    ) -> Result<Self, ExecutionError> {
        let source = source_factory.build(output_schemas)?;
        let forwarder = InternalChannelSourceForwarder::new(sender);
        Ok(Self {
            node_handle,
            source,
            last_checkpoint,
            forwarder,
            running,
        })
    }
}

impl Node for SourceSenderNode {
    fn run(mut self) -> Result<(), ExecutionError> {
        let result = self
            .source
            .start(&mut self.forwarder, Some(self.last_checkpoint));
        self.running.store(false, Ordering::SeqCst);
        debug!("[{}-sender] Quit", self.node_handle);
        result
    }
}

/// The listener part of a source in the execution DAG.
#[derive(Debug)]
pub struct SourceListenerNode {
    /// Node handle in description DAG.
    node_handle: NodeHandle,
    /// Output from corresponding source sender.
    receiver: Receiver<(PortHandle, u64, u64, Operation)>,
    /// Receiving timeout.
    timeout: Duration,
    /// If the execution DAG should be running. Used for determining if a `terminate` message should be sent.
    running: Arc<AtomicBool>,
    /// This node's output channel manager, for communicating to other sources to coordinate terminate and commit, forwarding data, writing metadata and writing port state.
    channel_manager: SourceChannelManager,
}

impl SourceListenerNode {
    /// # Arguments
    ///
    /// - `node_handle`: Node handle in description DAG.
    /// - `receiver`: Channel that the data comes in.
    /// - `timeout`: `Listener timeout. After this timeout, listener will check if commit or terminate need to happen.
    /// - `base_path`: Base path of persisted data for the last execution of the description DAG.
    /// - `output_ports`: Output port definition of the source in description DAG.
    /// - `record_readers`: Record readers of all stateful ports.
    /// - `senders`: Output channels from this processor.
    /// - `edges`: All edges in the description DAG, used for creating record readers for input ports which is connected to this processor's stateful output ports.
    /// - `running`: If the execution DAG should still be running.
    /// - `epoch_manager`: Used for coordinating commit and terminate between sources. Shared by all sources.
    /// - `output_schemas`: Output data schemas.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new(
        node_handle: NodeHandle,
        receiver: Receiver<(PortHandle, u64, u64, Operation)>,
        timeout: Duration,
        base_path: &Path,
        output_ports: &[OutputPortDef],
        record_readers: Arc<
            RwLock<HashMap<NodeHandle, HashMap<PortHandle, Box<dyn RecordReader>>>>,
        >,
        senders: HashMap<PortHandle, Vec<Sender<ExecutorOperation>>>,
        edges: &[Edge],
        running: Arc<AtomicBool>,
        commit_sz: u32,
        max_duration_between_commits: Duration,
        epoch_manager: Arc<EpochManager>,
        output_schemas: HashMap<PortHandle, Schema>,
        start_seq: (u64, u64),
    ) -> Result<Self, ExecutionError> {
        let state_meta = init_component(&node_handle, base_path, |_| Ok(()))?;
        let (master_tx, port_databases) =
            create_ports_databases_and_fill_downstream_record_readers(
                &node_handle,
                edges,
                state_meta.env,
                output_ports,
                &mut record_readers.write(),
            )?;
        let channel_manager = SourceChannelManager::new(
            node_handle.clone(),
            senders,
            StateWriter::new(
                state_meta.meta_db,
                port_databases,
                master_tx,
                output_schemas,
            )?,
            true,
            commit_sz,
            max_duration_between_commits,
            epoch_manager,
            start_seq,
        );
        Ok(Self {
            node_handle,
            receiver,
            timeout,
            running,
            channel_manager,
        })
    }
}

impl SourceListenerNode {
    /// Returns if the node should terminate.
    fn send_and_trigger_commit_if_needed(
        &mut self,
        data: Option<(PortHandle, u64, u64, Operation)>,
    ) -> Result<bool, ExecutionError> {
        // First check if termination was requested.
        let terminating = !self.running.load(Ordering::SeqCst);
        // If this commit was not requested with termination at the start, we shouldn't terminate either.
        let terminating = match data {
            Some((port, txid, seq_in_tx, op)) => self
                .channel_manager
                .send_and_trigger_commit_if_needed(txid, seq_in_tx, op, port, terminating)?,
            None => self.channel_manager.trigger_commit_if_needed(terminating)?,
        };
        if terminating {
            self.channel_manager.terminate()?;
            debug!("[{}-listener] Quitting", &self.node_handle);
        }
        Ok(terminating)
    }
}

impl Node for SourceListenerNode {
    fn run(mut self) -> Result<(), ExecutionError> {
        loop {
            match self.receiver.recv_timeout(self.timeout) {
                Ok(data) => {
                    if self.send_and_trigger_commit_if_needed(Some(data))? {
                        return Ok(());
                    }
                }
                Err(e) => {
                    if self.send_and_trigger_commit_if_needed(None)? {
                        return Ok(());
                    }
                    // Channel disconnected but running flag not set to false, the source sender must have panicked.
                    if self.running.load(Ordering::SeqCst) && e == RecvTimeoutError::Disconnected {
                        return Err(ExecutionError::ChannelDisconnected);
                    }
                }
            }
        }
    }
}

1	use std::{
2	collections::HashMap,
3	path::Path,
4	sync::{
5	atomic::{AtomicBool, Ordering},
6	Arc,
7	},
8	time::Duration,
9	};
10
11	use crossbeam::channel::{Receiver, RecvTimeoutError, Sender};
12	use dozer_types::log::debug;
13	use dozer_types::{
14	internal_err,
15	parking_lot::RwLock,
16	types::{Operation, Schema},
17	};
18
19	use crate::dag::{
20	channels::SourceChannelForwarder,
21	dag::Edge,
22	epoch::EpochManager,
23	errors::ExecutionError::{self, InternalError},
24	executor_utils::{create_ports_databases_and_fill_downstream_record_readers, init_component},
25	forwarder::{SourceChannelManager, StateWriter},
26	node::{NodeHandle, OutputPortDef, PortHandle, Source, SourceFactory},
27	record_store::RecordReader,
28	};
29
30	use super::{node::Node, ExecutorOperation};
31
32	#[derive(Debug)]	×
33	struct InternalChannelSourceForwarder {
34	sender: Sender<(PortHandle, u64, u64, Operation)>,
35	}
36
37	impl InternalChannelSourceForwarder {
38	pub fn new(sender: Sender<(PortHandle, u64, u64, Operation)>) -> Self {	135✔
39	Self { sender }	135✔
40	}	135✔
41	}
42
43	impl SourceChannelForwarder for InternalChannelSourceForwarder {
44	fn send(	3,496,510✔
45	&mut self,	3,496,510✔
46	txid: u64,	3,496,510✔
47	seq_in_tx: u64,	3,496,510✔
48	op: Operation,	3,496,510✔
49	port: PortHandle,	3,496,510✔
50	) -> Result<(), ExecutionError> {	3,496,510✔
51	internal_err!(self.sender.send((port, txid, seq_in_tx, op)))	10✔
52	}	3,496,510✔
53	}
54
55	/// The sender half of a source in the execution DAG.
56	#[derive(Debug)]	×
57	pub struct SourceSenderNode {
58	/// Node handle in description DAG.
59	node_handle: NodeHandle,
60	/// The source.
61	source: Box<dyn Source>,
62	/// Last checkpointed output data sequence number.
63	last_checkpoint: (u64, u64),
64	/// The forwarder that will be passed to the source for outputig data.
65	forwarder: InternalChannelSourceForwarder,
66	/// If the execution DAG should be running. Used for terminating the execution DAG.
67	running: Arc<AtomicBool>,
68	}
69
70	impl SourceSenderNode {
71	/// # Arguments
72	///
73	/// - `node_handle`: Node handle in description DAG.
74	/// - `source_factory`: Source factory in description DAG.
75	/// - `output_schemas`: Output data schemas.
76	/// - `last_checkpoint`: Last checkpointed output of this source.
77	/// - `sender`: Channel to send data to.
78	/// - `running`: If the execution DAG should still be running.
79	pub fn new(	137✔
80	node_handle: NodeHandle,	137✔
81	source_factory: &dyn SourceFactory,	137✔
82	output_schemas: HashMap<PortHandle, Schema>,	137✔
83	last_checkpoint: (u64, u64),	137✔
84	sender: Sender<(PortHandle, u64, u64, Operation)>,	137✔
85	running: Arc<AtomicBool>,	137✔
86	) -> Result<Self, ExecutionError> {	137✔
87	let source = source_factory.build(output_schemas)?;	137✔
88	let forwarder = InternalChannelSourceForwarder::new(sender);	136✔
89	Ok(Self {	136✔
90	node_handle,	136✔
91	source,	136✔
92	last_checkpoint,	136✔
93	forwarder,	136✔
94	running,	136✔
95	})	136✔
96	}	137✔
97	}
98
99	impl Node for SourceSenderNode {
100	fn run(mut self) -> Result<(), ExecutionError> {	135✔
101	let result = self	135✔
102	.source	135✔
103	.start(&mut self.forwarder, Some(self.last_checkpoint));	135✔
104	self.running.store(false, Ordering::SeqCst);	135✔
105	debug!("[{}-sender] Quit", self.node_handle);	135✔
106	result	135✔
107	}	135✔
108	}
109
110	/// The listener part of a source in the execution DAG.
111	#[derive(Debug)]	×
112	pub struct SourceListenerNode {
113	/// Node handle in description DAG.
114	node_handle: NodeHandle,
115	/// Output from corresponding source sender.
116	receiver: Receiver<(PortHandle, u64, u64, Operation)>,
117	/// Receiving timeout.
118	timeout: Duration,
119	/// If the execution DAG should be running. Used for determining if a `terminate` message should be sent.
120	running: Arc<AtomicBool>,
121	/// This node's output channel manager, for communicating to other sources to coordinate terminate and commit, forwarding data, writing metadata and writing port state.
122	channel_manager: SourceChannelManager,
123	}
124
125	impl SourceListenerNode {
126	/// # Arguments
127	///
128	/// - `node_handle`: Node handle in description DAG.
129	/// - `receiver`: Channel that the data comes in.
130	/// - `timeout`: `Listener timeout. After this timeout, listener will check if commit or terminate need to happen.
131	/// - `base_path`: Base path of persisted data for the last execution of the description DAG.
132	/// - `output_ports`: Output port definition of the source in description DAG.
133	/// - `record_readers`: Record readers of all stateful ports.
134	/// - `senders`: Output channels from this processor.
135	/// - `edges`: All edges in the description DAG, used for creating record readers for input ports which is connected to this processor's stateful output ports.
136	/// - `running`: If the execution DAG should still be running.
137	/// - `epoch_manager`: Used for coordinating commit and terminate between sources. Shared by all sources.
138	/// - `output_schemas`: Output data schemas.
139	#[allow(clippy::too_many_arguments)]
140	pub(crate) fn new(	137✔
141	node_handle: NodeHandle,	137✔
142	receiver: Receiver<(PortHandle, u64, u64, Operation)>,	137✔
143	timeout: Duration,	137✔
144	base_path: &Path,	137✔
145	output_ports: &[OutputPortDef],	137✔
146	record_readers: Arc<	137✔
147	RwLock<HashMap<NodeHandle, HashMap<PortHandle, Box<dyn RecordReader>>>>,	137✔
148	>,	137✔
149	senders: HashMap<PortHandle, Vec<Sender<ExecutorOperation>>>,	137✔
150	edges: &[Edge],	137✔
151	running: Arc<AtomicBool>,	137✔
152	commit_sz: u32,	137✔
153	max_duration_between_commits: Duration,	137✔
154	epoch_manager: Arc<EpochManager>,	137✔
155	output_schemas: HashMap<PortHandle, Schema>,	137✔
156	start_seq: (u64, u64),	137✔
157	) -> Result<Self, ExecutionError> {	137✔
158	let state_meta = init_component(&node_handle, base_path, \|_\| Ok(()))?;	137✔
159	let (master_tx, port_databases) =	136✔
160	create_ports_databases_and_fill_downstream_record_readers(	136✔
161	&node_handle,	136✔
162	edges,	136✔
163	state_meta.env,	136✔
164	output_ports,	136✔
165	&mut record_readers.write(),	136✔
166	)?;	136✔
167	let channel_manager = SourceChannelManager::new(	136✔
168	node_handle.clone(),	136✔
169	senders,	136✔
170	StateWriter::new(	136✔
171	state_meta.meta_db,	136✔
172	port_databases,	136✔
173	master_tx,	136✔
174	output_schemas,	136✔
175	)?,	136✔
176	true,	×
177	commit_sz,	136✔
178	max_duration_between_commits,	136✔
179	epoch_manager,	136✔
180	start_seq,	136✔
181	);	136✔
182	Ok(Self {	136✔
183	node_handle,	136✔
184	receiver,	136✔
185	timeout,	136✔
186	running,	136✔
187	channel_manager,	136✔
188	})	136✔
189	}	137✔
190	}
191
192	impl SourceListenerNode {	×
193	/// Returns if the node should terminate.	×
194	fn send_and_trigger_commit_if_needed(	3,498,283✔
195	&mut self,	3,498,283✔
196	data: Option<(PortHandle, u64, u64, Operation)>,	3,498,283✔
197	) -> Result<bool, ExecutionError> {	3,498,283✔
198	// First check if termination was requested.	3,498,283✔
199	let terminating = !self.running.load(Ordering::SeqCst);	3,498,283✔
200	// If this commit was not requested with termination at the start, we shouldn't terminate either.	×
201	let terminating = match data {	3,498,283✔
202	Some((port, txid, seq_in_tx, op)) => self	3,495,902✔
203	.channel_manager	3,495,902✔
204	.send_and_trigger_commit_if_needed(txid, seq_in_tx, op, port, terminating)?,	3,495,902✔
205	None => self.channel_manager.trigger_commit_if_needed(terminating)?,	2,381✔
206	};	×
207	if terminating {	3,498,276✔
208	self.channel_manager.terminate()?;	129✔
209	debug!("[{}-listener] Quitting", &self.node_handle);	129✔
210	}	3,498,147✔
211	Ok(terminating)	3,498,274✔
212	}	3,498,281✔
213	}
214		×
215	impl Node for SourceListenerNode {	×
216	fn run(mut self) -> Result<(), ExecutionError> {	136✔
217	loop {	3,498,248✔
218	match self.receiver.recv_timeout(self.timeout) {	3,498,248✔
219	Ok(data) => {	3,495,809✔
220	if self.send_and_trigger_commit_if_needed(Some(data))? {	3,495,809✔
221	return Ok(());	6✔
222	}	3,495,796✔
223	}	×
224	Err(e) => {	2,439✔
225	if self.send_and_trigger_commit_if_needed(None)? {	2,439✔
226	return Ok(());	121✔
227	}	2,318✔
228	// Channel disconnected but running flag not set to false, the source sender must have panicked.	2,318✔
229	if self.running.load(Ordering::SeqCst) && e == RecvTimeoutError::Disconnected {	2,318✔
230	return Err(ExecutionError::ChannelDisconnected);	2✔
231	}	2,316✔
232	}
233	}	×
234	}
235	}	136✔
236	}

getdozer / dozer / 3974044832

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