Merge pull request #2416 from AmmarAbouZor/producer-collector

Refactor Producer: Collect records and write them in bulks & Fix tailing for binary files

Author: marcmo

application/apps/indexer/addons/dlt-tools/Cargo.toml

@@ -9,6 +9,7 @@ indexer_base = { path = "../../indexer_base" }
 log.workspace = true
 parsers = { path = "../../parsers" }
 sources = { path = "../../sources" }
+processor = { path = "../../processor"}
 tokio = { workspace = true , features = ["full"] }
 tokio-util = { workspace = true, features = ["codec", "net"] }
 

application/apps/indexer/addons/dlt-tools/src/attachments_collector.rs

@@ -0,0 +1,20 @@
+use parsers::{Attachment, LogMessage};
+use processor::producer::LogRecordsCollector;
+
+/// Logs Collector with interest of attachments only.
+#[derive(Default)]
+pub struct AttachmentsCollector {
+    pub attachments: Vec<Attachment>,
+}
+
+impl<T: LogMessage> LogRecordsCollector<T> for AttachmentsCollector {
+    fn append(&mut self, log_record: parsers::ParseYield<T>) {
+        match log_record {
+            parsers::ParseYield::Message(_) => {}
+            parsers::ParseYield::Attachment(attachment) => self.attachments.push(attachment),
+            parsers::ParseYield::MessageAndAttachment((_msg, attachment)) => {
+                self.attachments.push(attachment)
+            }
+        }
+    }
+}

application/apps/indexer/addons/dlt-tools/src/lib.rs

@@ -16,15 +16,20 @@ extern crate indexer_base;
 extern crate log;
 
 use dlt_core::filtering::DltFilterConfig;
-use parsers::{Attachment, MessageStreamItem, ParseYield, dlt::DltParser};
-use sources::{binary::raw::BinaryByteSource, producer::MessageProducer};
+use parsers::{Attachment, dlt::DltParser};
+use processor::producer::{MessageProducer, ProduceSummary};
+use sources::binary::raw::BinaryByteSource;
 use std::{
     fs::File,
     io::{BufReader, BufWriter, Write},
     path::{Path, PathBuf},
 };
 use tokio_util::sync::CancellationToken;
 
+use crate::attachments_collector::AttachmentsCollector;
+
+mod attachments_collector;
+
 pub async fn scan_dlt_ft(
     input: PathBuf,
     filter: Option<DltFilterConfig>,
@@ -47,7 +52,8 @@ pub async fn scan_dlt_ft(
 
             let mut canceled = false;
 
-            let mut attachments = vec![];
+            let mut collector = AttachmentsCollector::default();
+
             loop {
                 tokio::select! {
                     // Check on events in current order ensuring cancel will be checked at first
@@ -58,20 +64,18 @@ pub async fn scan_dlt_ft(
                         canceled = true;
                         break;
                     }
-                    items = producer.read_next_segment() => {
-                        match items {
-                            Some(items) => {
-                                for (_, item) in items {
-                                    if let MessageStreamItem::Item(ParseYield::MessageAndAttachment((_msg, attachment))) = item {
-                                        attachments.push(attachment.to_owned());
-                                    } else if let MessageStreamItem::Item(ParseYield::Attachment(attachment)) = item {
-                                        attachments.push(attachment.to_owned());
-                                    }
-                                }
-                            }
-                            _ => {
+                    prod_res = producer.produce_next(&mut collector) => {
+                        match prod_res {
+                            Ok(ProduceSummary::Processed {..}) => {
+                                // Attachments are appended.
+                            },
+                            Ok(ProduceSummary::Done {..} | ProduceSummary::NoBytesAvailable {..}) => {
+                                // Stop as tailing isn't needed here.
                                 break;
                             }
+                            Err(err) => {
+                                return Err(format!("Error while processing DLT file. {err}"));
+                            },
                         }
                     }
                 }
@@ -81,7 +85,7 @@ pub async fn scan_dlt_ft(
                 return Ok(Vec::new());
             }
 
-            Ok(attachments)
+            Ok(collector.attachments)
         }
         Err(error) => Err(format!("failed to open file: {error}")),
     }

application/apps/indexer/parsers/src/lib.rs

@@ -97,15 +97,6 @@ pub trait LogMessage: Display + Serialize {
     fn to_writer<W: Write>(&self, writer: &mut W) -> Result<usize, std::io::Error>;
 }
 
-#[derive(Debug)]
-pub enum MessageStreamItem<T: LogMessage> {
-    Item(ParseYield<T>),
-    Skipped,
-    Incomplete,
-    Empty,
-    Done,
-}
-
 /// A trait for parsers that extract one item at a time from a byte slice.
 ///
 /// Any type implementing this trait will automatically implement the [`Parser`] trait

application/apps/indexer/plugins_host/Cargo.toml

@@ -26,6 +26,7 @@ dir_checksum = { path = "../../../../cli/development-cli/dir_checksum" }
 
 [dev-dependencies]
 criterion = { workspace = true, features = ["async_tokio"] }
+processor = { path = "../processor"}
 
 [[bench]]
 name = "plugin_parser_init"

application/apps/indexer/plugins_host/benches/plugin_utls.rs

@@ -9,7 +9,7 @@ use stypes::{PluginConfigItem, PluginConfigValue};
 // Rust LSP may mark this as an error, but this is an issue with the LSP itself.
 // The code will compile without problems.
 
-#[path = "./../../sources/benches/bench_utls.rs"]
+#[path = "./../../processor/benches/bench_utls.rs"]
 mod bench_utls;
 
 /// Represents the needed configuration to run benchmarks on a plugin.

application/apps/indexer/processor/Cargo.toml

@@ -27,11 +27,50 @@ uuid = { workspace = true , features = ["serde", "v4"] }
 stypes = { path = "../stypes", features=["rustcore"] }
 
 [dev-dependencies]
-criterion.workspace = true
+criterion = { workspace = true, features = ["async_tokio"] }
 pretty_assertions = "1.4"
 rand.workspace = true
 tempfile.workspace = true
+tokio.workspace = true
+plugins_host = {path = "../plugins_host/"}
+toml.workspace = true
 
 [[bench]]
 name = "map_benchmarks"
 harness = false
+
+[[bench]]
+name = "dlt_producer"
+harness = false
+
+[[bench]]
+name = "someip_producer"
+harness = false
+
+[[bench]]
+name = "someip_legacy_producer"
+harness = false
+
+[[bench]]
+name = "text_producer"
+harness = false
+
+[[bench]]
+name = "plugin_praser_producer"
+harness = false
+
+[[bench]]
+name = "mocks_once_producer"
+harness = false
+
+[[bench]]
+name = "mocks_once_parallel"
+harness = false
+
+[[bench]]
+name = "mocks_multi_producer"
+harness = false
+
+[[bench]]
+name = "mocks_multi_parallel"
+harness = false

…ps/indexer/sources/benches/bench_utls.rs …/indexer/processor/benches/bench_utls.rsapplication/apps/indexer/sources/benches/bench_utls.rs renamed to application/apps/indexer/processor/benches/bench_utls.rs application/apps/indexer/sources/benches/bench_utls.rs renamed to application/apps/indexer/processor/benches/bench_utls.rs

@@ -3,15 +3,17 @@
 #![allow(unused)]
 
 use std::{
+    alloc::Layout,
     fs::File,
     io::{Cursor, Read},
     path::PathBuf,
     time::Duration,
 };
 
 use criterion::Criterion;
-use parsers::{LogMessage, MessageStreamItem};
-use sources::{binary::raw::BinaryByteSource, producer::MessageProducer};
+use parsers::{LogMessage, ParseYield};
+use processor::producer::{GeneralLogCollector, MessageProducer};
+use sources::binary::raw::BinaryByteSource;
 
 pub const INPUT_SOURCE_ENV_VAR: &str = "CHIPMUNK_BENCH_SOURCE";
 pub const CONFIG_ENV_VAR: &str = "CHIPMUNK_BENCH_CONFIG";
@@ -63,12 +65,12 @@ pub fn get_config() -> Option<String> {
 /// The purpose of this struct is to convince the compiler that we are using all input
 /// possibilities of producer to avoid unwanted optimizations.
 pub struct ProducerCounter {
+    pub items: usize,
     pub msg: usize,
     pub txt: usize,
     pub att: usize,
-    pub skipped: usize,
-    pub incomplete: usize,
-    pub empty: usize,
+    pub loaded_bytes: usize,
+    pub skipped_bytes: usize,
 }
 
 /// Run producer until the end converting messages into strings too, while counting all the
@@ -80,28 +82,48 @@ where
     T: LogMessage,
 {
     let mut counter = ProducerCounter::default();
+    let mut collector = GeneralLogCollector::default();
+
+    loop {
+        collector.get_records().clear();
+        match producer.produce_next(&mut collector).await.unwrap() {
+            processor::producer::ProduceSummary::Processed {
+                bytes_consumed,
+                messages_count,
+                skipped_bytes,
+            } => {
+                counter.items += messages_count;
+                counter.loaded_bytes += bytes_consumed;
+                counter.skipped_bytes += skipped_bytes;
+            }
+            processor::producer::ProduceSummary::NoBytesAvailable { skipped_bytes } => {
+                counter.skipped_bytes += skipped_bytes;
+                break;
+            }
+            processor::producer::ProduceSummary::Done {
+                loaded_bytes,
+                skipped_bytes,
+                produced_messages,
+            } => {
+                counter.loaded_bytes = loaded_bytes;
+                counter.skipped_bytes = skipped_bytes;
+                counter.items = produced_messages;
+                break;
+            }
+        }
 
-    while let Some(items) = producer.read_next_segment().await {
-        for (_, i) in items {
-            match i {
-                MessageStreamItem::Item(item) => match item {
-                    parsers::ParseYield::Message(msg) => {
-                        counter.msg += 1;
-                        counter.txt += msg.to_string().len();
-                    }
-                    parsers::ParseYield::Attachment(att) => counter.att += att.size,
-                    parsers::ParseYield::MessageAndAttachment((msg, att)) => {
-                        counter.msg += 1;
-                        counter.txt += msg.to_string().len();
-                        counter.att += att.size;
-                    }
-                },
-                MessageStreamItem::Skipped => {
-                    counter.skipped += 1;
+        for item in collector.get_records() {
+            match item {
+                ParseYield::Message(msg) => {
+                    counter.msg += 1;
+                    counter.txt += msg.to_string().len();
+                }
+                ParseYield::Attachment(att) => counter.att += att.size,
+                ParseYield::MessageAndAttachment((msg, att)) => {
+                    counter.msg += 1;
+                    counter.txt += msg.to_string().len();
+                    counter.att += att.size;
                 }
-                MessageStreamItem::Incomplete => counter.incomplete += 1,
-                MessageStreamItem::Empty => counter.empty += 1,
-                MessageStreamItem::Done => break,
             }
         }
     }

…/indexer/sources/benches/dlt_producer.rs …ndexer/processor/benches/dlt_producer.rsapplication/apps/indexer/sources/benches/dlt_producer.rs renamed to application/apps/indexer/processor/benches/dlt_producer.rs application/apps/indexer/sources/benches/dlt_producer.rs renamed to application/apps/indexer/processor/benches/dlt_producer.rs

@@ -5,7 +5,7 @@ use bench_utls::{
     bench_standrad_config, create_binary_bytesource, get_config, read_binary, run_producer,
 };
 use parsers::dlt::{self, DltParser};
-use sources::producer::MessageProducer;
+use processor::producer::MessageProducer;
 
 mod bench_utls;