Move producer benches to processor + Produce Error in Export

* Add producer error as a variant for export errors.
* Move all benchmarks related to producer to processor crate since
  producer is moved there as well.
* Corresponding changes in development CLI tool configuration file for
  benchmarks.
* Apply adjustments in the producer loop with records collectors and
  state delivering in benchmarks.

Author: AmmarAbouZor

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,12 +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;
 

…xer/sources/benches/mocks/mock_parser.rs …r/processor/benches/mocks/mock_parser.rsapplication/apps/indexer/sources/benches/mocks/mock_parser.rs renamed to application/apps/indexer/processor/benches/mocks/mock_parser.rs application/apps/indexer/sources/benches/mocks/mock_parser.rs renamed to application/apps/indexer/processor/benches/mocks/mock_parser.rs

@@ -10,7 +10,7 @@ use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
 
 use bench_utls::{bench_standrad_config, run_producer};
 use mocks::{mock_parser::MockParser, mock_source::MockByteSource};
-use sources::producer::MessageProducer;
+use processor::producer::MessageProducer;
 
 mod bench_utls;
 mod mocks;

…xer/sources/benches/mocks/mock_source.rs …r/processor/benches/mocks/mock_source.rsapplication/apps/indexer/sources/benches/mocks/mock_source.rs renamed to application/apps/indexer/processor/benches/mocks/mock_source.rs application/apps/indexer/sources/benches/mocks/mock_source.rs renamed to application/apps/indexer/processor/benches/mocks/mock_source.rs

@@ -6,7 +6,7 @@
 use bench_utls::{bench_standrad_config, run_producer};
 use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
 use mocks::{mock_parser::MockParser, mock_source::MockByteSource};
-use sources::producer::MessageProducer;
+use processor::producer::MessageProducer;
 mod bench_utls;
 mod mocks;