3975593040

Build Type

push

github

Committed by GitHub

Commit Message

feature: Atomatically trim record history in `RecordWriter` (#699)

Run Details

255 of 255 new or added lines in 7 files covered. (100.0%)

22731 of 34140 relevant lines covered (66.58%)

72049.25 hits per line

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

89.05

/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.
    /// - `retention_queue_size`: Size of retention queue (used by RecordWriter)
    #[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),
        retention_queue_size: usize,
    ) -> 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,
                retention_queue_size,
            )?,
            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 {	188✔
39	Self { sender }	188✔
40	}	188✔
41	}
42
43	impl SourceChannelForwarder for InternalChannelSourceForwarder {
44	fn send(	4,773,442✔
45	&mut self,	4,773,442✔
46	txid: u64,	4,773,442✔
47	seq_in_tx: u64,	4,773,442✔
48	op: Operation,	4,773,442✔
49	port: PortHandle,	4,773,442✔
50	) -> Result<(), ExecutionError> {	4,773,442✔
51	internal_err!(self.sender.send((port, txid, seq_in_tx, op)))	9✔
52	}	4,773,442✔
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(	190✔
80	node_handle: NodeHandle,	190✔
81	source_factory: &dyn SourceFactory,	190✔
82	output_schemas: HashMap<PortHandle, Schema>,	190✔
83	last_checkpoint: (u64, u64),	190✔
84	sender: Sender<(PortHandle, u64, u64, Operation)>,	190✔
85	running: Arc<AtomicBool>,	190✔
86	) -> Result<Self, ExecutionError> {	190✔
87	let source = source_factory.build(output_schemas)?;	190✔
88	let forwarder = InternalChannelSourceForwarder::new(sender);	189✔
89	Ok(Self {	189✔
90	node_handle,	189✔
91	source,	189✔
92	last_checkpoint,	189✔
93	forwarder,	189✔
94	running,	189✔
95	})	189✔
96	}	190✔
97	}
98
99	impl Node for SourceSenderNode {
100	fn run(mut self) -> Result<(), ExecutionError> {	188✔
101	let result = self	188✔
102	.source	188✔
103	.start(&mut self.forwarder, Some(self.last_checkpoint));	188✔
104	self.running.store(false, Ordering::SeqCst);	188✔
105	debug!("[{}-sender] Quit", self.node_handle);	188✔
106	result	188✔
107	}	188✔
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	/// - `retention_queue_size`: Size of retention queue (used by RecordWriter)
140	#[allow(clippy::too_many_arguments)]	×
141	pub(crate) fn new(	138✔
142	node_handle: NodeHandle,	138✔
143	receiver: Receiver<(PortHandle, u64, u64, Operation)>,	138✔
144	timeout: Duration,	138✔
145	base_path: &Path,	138✔
146	output_ports: &[OutputPortDef],	138✔
147	record_readers: Arc<	138✔
148	RwLock<HashMap<NodeHandle, HashMap<PortHandle, Box<dyn RecordReader>>>>,	138✔
149	>,	138✔
150	senders: HashMap<PortHandle, Vec<Sender<ExecutorOperation>>>,	138✔
151	edges: &[Edge],	138✔
152	running: Arc<AtomicBool>,	138✔
153	commit_sz: u32,	138✔
154	max_duration_between_commits: Duration,	138✔
155	epoch_manager: Arc<EpochManager>,	138✔
156	output_schemas: HashMap<PortHandle, Schema>,	138✔
157	start_seq: (u64, u64),	138✔
158	retention_queue_size: usize,	138✔
159	) -> Result<Self, ExecutionError> {	138✔
160	let state_meta = init_component(&node_handle, base_path, \|_\| Ok(()))?;	138✔
161	let (master_tx, port_databases) =	136✔
162	create_ports_databases_and_fill_downstream_record_readers(	136✔
163	&node_handle,	136✔
164	edges,	136✔
165	state_meta.env,	136✔
166	output_ports,	136✔
167	&mut record_readers.write(),	136✔
168	)?;	136✔
169	let channel_manager = SourceChannelManager::new(	136✔
170	node_handle.clone(),	136✔
171	senders,	136✔
172	StateWriter::new(	136✔
173	state_meta.meta_db,	136✔
174	port_databases,	136✔
175	master_tx,	136✔
176	output_schemas,	136✔
177	retention_queue_size,	136✔
178	)?,	136✔
179	true,	×
180	commit_sz,	136✔
181	max_duration_between_commits,	136✔
182	epoch_manager,	136✔
183	start_seq,	136✔
184	);	136✔
185	Ok(Self {	136✔
186	node_handle,	136✔
187	receiver,	136✔
188	timeout,	136✔
189	running,	136✔
190	channel_manager,	136✔
191	})	136✔
192	}	138✔
193	}	×
194		×
195	impl SourceListenerNode {	×
196	/// Returns if the node should terminate.	×
197	fn send_and_trigger_commit_if_needed(	4,568,244✔
198	&mut self,	4,568,244✔
199	data: Option<(PortHandle, u64, u64, Operation)>,	4,568,244✔
200	) -> Result<bool, ExecutionError> {	4,568,244✔
201	// First check if termination was requested.	4,568,244✔
202	let terminating = !self.running.load(Ordering::SeqCst);	4,568,244✔
203	// If this commit was not requested with termination at the start, we shouldn't terminate either.	×
204	let terminating = match data {	4,568,244✔
205	Some((port, txid, seq_in_tx, op)) => self	4,566,064✔
206	.channel_manager	4,566,064✔
207	.send_and_trigger_commit_if_needed(txid, seq_in_tx, op, port, terminating)?,	4,566,064✔
208	None => self.channel_manager.trigger_commit_if_needed(terminating)?,	2,180✔
209	};	×
210	if terminating {	4,568,238✔
211	self.channel_manager.terminate()?;	3,240✔
212	debug!("[{}-listener] Quitting", &self.node_handle);	3,240✔
213	}	4,564,998✔
214	Ok(terminating)	4,565,126✔
215	}	4,565,132✔
216	}	×
217		×
218	impl Node for SourceListenerNode {	×
219	fn run(mut self) -> Result<(), ExecutionError> {	136✔
220	loop {	4,567,200✔
221	match self.receiver.recv_timeout(self.timeout) {	4,567,200✔
222	Ok(data) => {	4,564,959✔
223	if self.send_and_trigger_commit_if_needed(Some(data))? {	4,564,959✔
224	return Ok(());	7✔
225	}	4,564,946✔
226	}	×
227	Err(e) => {	2,241✔
228	if self.send_and_trigger_commit_if_needed(None)? {	2,241✔
229	return Ok(());	121✔
230	}	2,120✔
231	// Channel disconnected but running flag not set to false, the source sender must have panicked.	2,120✔
232	if self.running.load(Ordering::SeqCst) && e == RecvTimeoutError::Disconnected {	2,120✔
233	return Err(ExecutionError::ChannelDisconnected);	2✔
234	}	2,118✔
235	}
236	}
237	}
238	}	136✔
239	}

getdozer / dozer / 3975593040

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