4361707165

Build Type

push

github

Committed by GitHub

Commit Message

refactor: comparison errors (#1163)

Run Details

286 of 286 new or added lines in 3 files covered. (100.0%)

27143 of 35806 relevant lines covered (75.81%)

33477.66 hits per line

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

95.0

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

use std::{
    sync::{
        atomic::{AtomicBool, Ordering},
        Arc,
    },
    time::Duration,
};

use crossbeam::channel::{bounded, Receiver, RecvTimeoutError, Sender};
use dozer_types::ingestion_types::IngestionMessage;
use dozer_types::{
    log::debug,
    node::{NodeHandle, OpIdentifier},
};

use crate::{
    builder_dag::NodeKind,
    channels::SourceChannelForwarder,
    errors::ExecutionError,
    forwarder::{SourceChannelManager, StateWriter},
    node::{PortHandle, Source},
};

use super::{execution_dag::ExecutionDag, node::Node, ExecutorOptions};

impl SourceChannelForwarder for InternalChannelSourceForwarder {
    fn send(&mut self, message: IngestionMessage, port: PortHandle) -> Result<(), ExecutionError> {
        Ok(self.sender.send((port, message))?)
    }
}

/// 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: Option<OpIdentifier>,
    /// The forwarder that will be passed to the source for outputting data.
    forwarder: InternalChannelSourceForwarder,
}

impl SourceSenderNode {
    pub fn handle(&self) -> &NodeHandle {
        &self.node_handle
    }
}

impl Node for SourceSenderNode {
    fn run(mut self) -> Result<(), ExecutionError> {
        let result = self.source.start(
            &mut self.forwarder,
            self.last_checkpoint
                .map(|op_id| (op_id.txid, op_id.seq_in_tx)),
        );
        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, IngestionMessage)>,
    /// 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,
}

#[derive(Debug, Clone, PartialEq)]
enum DataKind {
    Data((PortHandle, IngestionMessage)),
    NoDataBecauseOfTimeout,
    NoDataBecauseOfChannelDisconnection,
}

impl SourceListenerNode {
    /// Returns if the node should terminate.
    fn send_and_trigger_commit_if_needed(
        &mut self,
        data: DataKind,
    ) -> Result<bool, ExecutionError> {
        // If termination was requested the or source quit, we try to terminate.
        let terminating = data == DataKind::NoDataBecauseOfChannelDisconnection
            || !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 {
            DataKind::Data((port, message)) => self
                .channel_manager
                .send_and_trigger_commit_if_needed(message, port, terminating)?,
            DataKind::NoDataBecauseOfTimeout | DataKind::NoDataBecauseOfChannelDisconnection => {
                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 {
            let terminating = match self.receiver.recv_timeout(self.timeout) {
                Ok(data) => self.send_and_trigger_commit_if_needed(DataKind::Data(data))?,
                Err(RecvTimeoutError::Timeout) => {
                    self.send_and_trigger_commit_if_needed(DataKind::NoDataBecauseOfTimeout)?
                }
                Err(RecvTimeoutError::Disconnected) => self.send_and_trigger_commit_if_needed(
                    DataKind::NoDataBecauseOfChannelDisconnection,
                )?,
            };
            if terminating {
                return Ok(());
            }
        }
    }
}

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

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

pub fn create_source_nodes(
    dag: &mut ExecutionDag,
    node_index: daggy::NodeIndex,
    options: &ExecutorOptions,
    running: Arc<AtomicBool>,
) -> (SourceSenderNode, SourceListenerNode) {
    // Get the source node.
    let Some(node) = dag.node_weight_mut(node_index).take() else {
        panic!("Must pass in a node")
    };
    let node_handle = node.handle;
    let node_storage = node.storage;
    let NodeKind::Source(source, last_checkpoint) = node.kind else {
        panic!("Must pass in a source node");
    };

    // Create channel between source sender and source listener.
    let (source_sender, source_receiver) = bounded(options.channel_buffer_sz);
    // let (source_sender, source_receiver) = bounded(1);

    // Create source listener.
    let forwarder = InternalChannelSourceForwarder::new(source_sender);
    let source_sender_node = SourceSenderNode {
        node_handle: node_handle.clone(),
        source,
        last_checkpoint,
        forwarder,
    };

    // Create source sender node.
    let (senders, record_writers) = dag.collect_senders_and_record_writers(node_index);
    let state_writer = StateWriter::new(
        node_storage.meta_db,
        record_writers,
        node_storage.master_txn,
    );
    let channel_manager = SourceChannelManager::new(
        node_handle.clone(),
        senders,
        state_writer,
        true,
        options.commit_sz,
        options.commit_time_threshold,
        dag.epoch_manager().clone(),
    );
    let source_listener_node = SourceListenerNode {
        node_handle,
        receiver: source_receiver,
        timeout: options.commit_time_threshold,
        running,
        channel_manager,
    };

    (source_sender_node, source_listener_node)
}

1	use std::{
2	sync::{
3	atomic::{AtomicBool, Ordering},
4	Arc,
5	},
6	time::Duration,
7	};
8
9	use crossbeam::channel::{bounded, Receiver, RecvTimeoutError, Sender};
10	use dozer_types::ingestion_types::IngestionMessage;
11	use dozer_types::{
12	log::debug,
13	node::{NodeHandle, OpIdentifier},
14	};
15
16	use crate::{
17	builder_dag::NodeKind,
18	channels::SourceChannelForwarder,
19	errors::ExecutionError,
20	forwarder::{SourceChannelManager, StateWriter},
21	node::{PortHandle, Source},
22	};
23
24	use super::{execution_dag::ExecutionDag, node::Node, ExecutorOptions};
25
26	impl SourceChannelForwarder for InternalChannelSourceForwarder {
27	fn send(&mut self, message: IngestionMessage, port: PortHandle) -> Result<(), ExecutionError> {	4,082,711✔
28	Ok(self.sender.send((port, message))?)	4,082,711✔
29	}	4,082,711✔
30	}
31
32	/// The sender half of a source in the execution DAG.
33	#[derive(Debug)]	×
34	pub struct SourceSenderNode {
35	/// Node handle in description DAG.
36	node_handle: NodeHandle,
37	/// The source.
38	source: Box<dyn Source>,
39	/// Last checkpointed output data sequence number.
40	last_checkpoint: Option<OpIdentifier>,
41	/// The forwarder that will be passed to the source for outputting data.
42	forwarder: InternalChannelSourceForwarder,
43	}
44
45	impl SourceSenderNode {
46	pub fn handle(&self) -> &NodeHandle {	338✔
47	&self.node_handle	338✔
48	}	338✔
49	}
50
51	impl Node for SourceSenderNode {
52	fn run(mut self) -> Result<(), ExecutionError> {	338✔
53	let result = self.source.start(	338✔
54	&mut self.forwarder,	338✔
55	self.last_checkpoint	338✔
56	.map(\|op_id\| (op_id.txid, op_id.seq_in_tx)),	338✔
57	);	338✔
58	debug!("[{}-sender] Quit", self.node_handle);	338✔
59	result	332✔
60	}	332✔
61	}
62
63	/// The listener part of a source in the execution DAG.
64	#[derive(Debug)]	×
65	pub struct SourceListenerNode {
66	/// Node handle in description DAG.
67	node_handle: NodeHandle,
68	/// Output from corresponding source sender.
69	receiver: Receiver<(PortHandle, IngestionMessage)>,
70	/// Receiving timeout.
71	timeout: Duration,
72	/// If the execution DAG should be running. Used for determining if a `terminate` message should be sent.
73	running: Arc<AtomicBool>,
74	/// This node's output channel manager, for communicating to other sources to coordinate terminate and commit, forwarding data, writing metadata and writing port state.
75	channel_manager: SourceChannelManager,
76	}
77
78	#[derive(Debug, Clone, PartialEq)]	3,910,908✔
79	enum DataKind {
80	Data((PortHandle, IngestionMessage)),
81	NoDataBecauseOfTimeout,
82	NoDataBecauseOfChannelDisconnection,
83	}
84
85	impl SourceListenerNode {
86	/// Returns if the node should terminate.
87	fn send_and_trigger_commit_if_needed(	3,891,002✔
88	&mut self,	3,891,002✔
89	data: DataKind,	3,891,002✔
90	) -> Result<bool, ExecutionError> {	3,891,002✔
91	// If termination was requested the or source quit, we try to terminate.
92	let terminating = data == DataKind::NoDataBecauseOfChannelDisconnection	3,891,002✔
93	\|\| !self.running.load(Ordering::SeqCst);	3,890,654✔
94	// If this commit was not requested with termination at the start, we shouldn't terminate either.
95	let terminating = match data {	3,891,002✔
96	DataKind::Data((port, message)) => self	3,887,899✔
97	.channel_manager	3,887,899✔
98	.send_and_trigger_commit_if_needed(message, port, terminating)?,	3,887,899✔
99	DataKind::NoDataBecauseOfTimeout \| DataKind::NoDataBecauseOfChannelDisconnection => {
100	self.channel_manager.trigger_commit_if_needed(terminating)?	3,103✔
101	}
102	};
103	if terminating {	3,890,997✔
104	self.channel_manager.terminate()?;	27,362✔
105	debug!("[{}-listener] Quitting", &self.node_handle);	27,362✔
106	}	3,863,635✔
107	Ok(terminating)	3,863,968✔
108	}	3,863,973✔
109	}
110
111	impl Node for SourceListenerNode {
112	fn run(mut self) -> Result<(), ExecutionError> {	338✔
113	loop {
114	let terminating = match self.receiver.recv_timeout(self.timeout) {	3,899,342✔
115	Ok(data) => self.send_and_trigger_commit_if_needed(DataKind::Data(data))?,	3,896,239✔
116	Err(RecvTimeoutError::Timeout) => {
117	self.send_and_trigger_commit_if_needed(DataKind::NoDataBecauseOfTimeout)?	2,755✔
118	}
119	Err(RecvTimeoutError::Disconnected) => self.send_and_trigger_commit_if_needed(	348✔
120	DataKind::NoDataBecauseOfChannelDisconnection,	348✔
121	)?,	348✔
122	};
123	if terminating {	3,899,336✔
124	return Ok(());	333✔
125	}	3,899,004✔
126	}
127	}	338✔
128	}
129
130	#[derive(Debug)]	×
131	struct InternalChannelSourceForwarder {
132	sender: Sender<(PortHandle, IngestionMessage)>,
133	}
134
135	impl InternalChannelSourceForwarder {
136	pub fn new(sender: Sender<(PortHandle, IngestionMessage)>) -> Self {	338✔
137	Self { sender }	338✔
138	}	338✔
139	}
140
141	pub fn create_source_nodes(	338✔
142	dag: &mut ExecutionDag,	338✔
143	node_index: daggy::NodeIndex,	338✔
144	options: &ExecutorOptions,	338✔
145	running: Arc<AtomicBool>,	338✔
146	) -> (SourceSenderNode, SourceListenerNode) {	338✔
147	// Get the source node.
148	let Some(node) = dag.node_weight_mut(node_index).take() else {	338✔
149	panic!("Must pass in a node")	×
150	};
151	let node_handle = node.handle;	338✔
152	let node_storage = node.storage;	338✔
153	let NodeKind::Source(source, last_checkpoint) = node.kind else {	338✔
154	panic!("Must pass in a source node");	×
155	};
156
157	// Create channel between source sender and source listener.
158	let (source_sender, source_receiver) = bounded(options.channel_buffer_sz);	338✔
159	// let (source_sender, source_receiver) = bounded(1);	338✔
160		338✔
161	// Create source listener.	338✔
162	let forwarder = InternalChannelSourceForwarder::new(source_sender);	338✔
163	let source_sender_node = SourceSenderNode {	338✔
164	node_handle: node_handle.clone(),	338✔
165	source,	338✔
166	last_checkpoint,	338✔
167	forwarder,	338✔
168	};	338✔
169		338✔
170	// Create source sender node.	338✔
171	let (senders, record_writers) = dag.collect_senders_and_record_writers(node_index);	338✔
172	let state_writer = StateWriter::new(	338✔
173	node_storage.meta_db,	338✔
174	record_writers,	338✔
175	node_storage.master_txn,	338✔
176	);	338✔
177	let channel_manager = SourceChannelManager::new(	338✔
178	node_handle.clone(),	338✔
179	senders,	338✔
180	state_writer,	338✔
181	true,	338✔
182	options.commit_sz,	338✔
183	options.commit_time_threshold,	338✔
184	dag.epoch_manager().clone(),	338✔
185	);	338✔
186	let source_listener_node = SourceListenerNode {	338✔
187	node_handle,	338✔
188	receiver: source_receiver,	338✔
189	timeout: options.commit_time_threshold,	338✔
190	running,	338✔
191	channel_manager,	338✔
192	};	338✔
193		338✔
194	(source_sender_node, source_listener_node)	338✔
195	}	338✔

getdozer / dozer / 4361707165

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