Skip to content
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.

Commit 44b9535

Browse files
AmmarAbouZorAmmarAbouZor
committedJun 12, 2024
Make parser async & Use with Arc wit Dlt parser & Adjustments on plugin hosts
- Changing parser trait to async caused an error in the DLT parser because it returns type has a reference in it. Because of that we got a compiler error about that the results doesn't implements Send trait enough. - This Error is possible to be a bug in rust that would be fixed in the future. See issues: - rust-lang/rust#64552 - rust-lang/rust#96865 - For now I replaced the references with Arcs in the results of DLT-Parser - implements for parser Plugin hosts now awaits on the async call from the plugin instead of using `futures::executer::block_on()`. However, this change didn't improve the performance of the plugins
1 parent 0e5b24a commit 44b9535

File tree

14 files changed

+225
-218
lines changed

14 files changed

+225
-218
lines changed
 

Diff for: ‎application/apps/indexer/Cargo.lock

+1
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

Diff for: ‎application/apps/indexer/parsers/Cargo.toml

+1
Original file line numberDiff line numberDiff line change
@@ -22,3 +22,4 @@ someip-payload = { git = "https://github.com/esrlabs/someip-payload" }
2222

2323
[dev-dependencies]
2424
stringreader = "0.1.1"
25+
tokio = { version = "1.24", features = ["full"] }

Diff for: ‎application/apps/indexer/parsers/src/dlt/fmt.rs

+10-9
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ use serde::ser::{Serialize, SerializeStruct, Serializer};
3030
use std::{
3131
fmt::{self, Formatter},
3232
str,
33+
sync::Arc,
3334
};
3435

3536
const DLT_COLUMN_SENTINAL: char = '\u{0004}';
@@ -194,13 +195,13 @@ impl From<Option<&String>> for FormatOptions {
194195
}
195196

196197
/// A dlt message that can be formatted with optional FIBEX data support
197-
pub struct FormattableMessage<'a> {
198+
pub struct FormattableMessage {
198199
pub message: Message,
199-
pub fibex_metadata: Option<&'a FibexMetadata>,
200-
pub options: Option<&'a FormatOptions>,
200+
pub fibex_metadata: Option<Arc<FibexMetadata>>,
201+
pub options: Option<Arc<FormatOptions>>,
201202
}
202203

203-
impl<'a> Serialize for FormattableMessage<'a> {
204+
impl Serialize for FormattableMessage {
204205
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
205206
where
206207
S: Serializer,
@@ -286,7 +287,7 @@ impl<'a> Serialize for FormattableMessage<'a> {
286287
}
287288
}
288289

289-
impl<'a> From<Message> for FormattableMessage<'a> {
290+
impl From<Message> for FormattableMessage {
290291
fn from(message: Message) -> Self {
291292
FormattableMessage {
292293
message,
@@ -319,7 +320,7 @@ impl<'a> PrintableMessage<'a> {
319320
}
320321
}
321322

322-
impl<'a> FormattableMessage<'a> {
323+
impl FormattableMessage {
323324
pub fn printable_parts<'b>(
324325
&'b self,
325326
ext_h_app_id: &'b str,
@@ -462,7 +463,7 @@ impl<'a> FormattableMessage<'a> {
462463
}
463464

464465
fn info_from_metadata<'b>(&'b self, id: u32, data: &[u8]) -> Option<NonVerboseInfo<'b>> {
465-
let fibex = self.fibex_metadata?;
466+
let fibex = self.fibex_metadata.as_ref()?;
466467
let md = extract_metadata(fibex, id, self.message.extended_header.as_ref())?;
467468
let msg_type: Option<MessageType> = message_type(&self.message, md.message_info.as_deref());
468469
let app_id = md.application_id.as_deref().or_else(|| {
@@ -511,7 +512,7 @@ impl<'a> FormattableMessage<'a> {
511512
}
512513
}
513514

514-
impl<'a> fmt::Display for FormattableMessage<'a> {
515+
impl fmt::Display for FormattableMessage {
515516
/// will format dlt Message with those fields:
516517
/// ********* storage-header ********
517518
/// date-time
@@ -530,7 +531,7 @@ impl<'a> fmt::Display for FormattableMessage<'a> {
530531
/// payload
531532
fn fmt(&self, f: &mut Formatter) -> Result<(), fmt::Error> {
532533
if let Some(h) = &self.message.storage_header {
533-
let tz = self.options.map(|o| o.tz);
534+
let tz = self.options.as_ref().map(|o| o.tz);
534535
match tz {
535536
Some(Some(tz)) => {
536537
write_tz_string(f, &h.timestamp, &tz)?;

Diff for: ‎application/apps/indexer/parsers/src/dlt/mod.rs

+18-18
Original file line numberDiff line numberDiff line change
@@ -13,11 +13,11 @@ use dlt_core::{
1313
parse::{dlt_consume_msg, dlt_message},
1414
};
1515
use serde::Serialize;
16-
use std::{io::Write, ops::Range};
16+
use std::{io::Write, ops::Range, sync::Arc};
1717

1818
use self::{attachment::FtScanner, fmt::FormatOptions};
1919

20-
impl LogMessage for FormattableMessage<'_> {
20+
impl LogMessage for FormattableMessage {
2121
fn to_writer<W: Write>(&self, writer: &mut W) -> Result<usize, std::io::Error> {
2222
let bytes = self.message.as_bytes();
2323
let len = bytes.len();
@@ -66,10 +66,10 @@ impl LogMessage for RawMessage {
6666

6767
#[derive(Default)]
6868
//TODO AAZ: This is the parser that should be replaced
69-
pub struct DltParser<'m> {
69+
pub struct DltParser {
7070
pub filter_config: Option<ProcessedDltFilterConfig>,
71-
pub fibex_metadata: Option<&'m FibexMetadata>,
72-
pub fmt_options: Option<&'m FormatOptions>,
71+
pub fibex_metadata: Option<Arc<FibexMetadata>>,
72+
pub fmt_options: Option<Arc<FormatOptions>>,
7373
pub with_storage_header: bool,
7474
ft_scanner: FtScanner,
7575
offset: usize,
@@ -98,30 +98,30 @@ impl DltRangeParser {
9898
}
9999
}
100100

101-
impl<'m> DltParser<'m> {
101+
impl DltParser {
102102
pub fn new(
103103
filter_config: Option<ProcessedDltFilterConfig>,
104-
fibex_metadata: Option<&'m FibexMetadata>,
105-
fmt_options: Option<&'m FormatOptions>,
104+
fibex_metadata: Option<FibexMetadata>,
105+
fmt_options: Option<FormatOptions>,
106106
with_storage_header: bool,
107107
) -> Self {
108108
Self {
109109
filter_config,
110-
fibex_metadata,
110+
fibex_metadata: fibex_metadata.map(Arc::new),
111111
with_storage_header,
112-
fmt_options,
112+
fmt_options: fmt_options.map(Arc::new),
113113
ft_scanner: FtScanner::new(),
114114
offset: 0,
115115
}
116116
}
117117
}
118118

119-
impl<'m> Parser<FormattableMessage<'m>> for DltParser<'m> {
120-
fn parse<'b>(
119+
impl Parser<FormattableMessage> for DltParser {
120+
async fn parse<'a>(
121121
&mut self,
122-
input: &'b [u8],
122+
input: &'a [u8],
123123
timestamp: Option<u64>,
124-
) -> Result<(&'b [u8], Option<ParseYield<FormattableMessage<'m>>>), Error> {
124+
) -> Result<(&'a [u8], Option<ParseYield<FormattableMessage>>), Error> {
125125
match dlt_message(input, self.filter_config.as_ref(), self.with_storage_header)
126126
.map_err(|e| Error::Parse(format!("{e}")))?
127127
{
@@ -142,8 +142,8 @@ impl<'m> Parser<FormattableMessage<'m>> for DltParser<'m> {
142142

143143
let msg = FormattableMessage {
144144
message: msg_with_storage_header,
145-
fibex_metadata: self.fibex_metadata,
146-
options: self.fmt_options,
145+
fibex_metadata: self.fibex_metadata.clone(),
146+
options: self.fmt_options.clone(),
147147
};
148148
self.offset += input.len() - rest.len();
149149
Ok((
@@ -160,7 +160,7 @@ impl<'m> Parser<FormattableMessage<'m>> for DltParser<'m> {
160160
}
161161

162162
impl Parser<RangeMessage> for DltRangeParser {
163-
fn parse<'b>(
163+
async fn parse<'b>(
164164
&mut self,
165165
input: &'b [u8],
166166
_timestamp: Option<u64>,
@@ -180,7 +180,7 @@ impl Parser<RangeMessage> for DltRangeParser {
180180
}
181181

182182
impl Parser<RawMessage> for DltRawParser {
183-
fn parse<'b>(
183+
async fn parse<'b>(
184184
&mut self,
185185
input: &'b [u8],
186186
_timestamp: Option<u64>,

Diff for: ‎application/apps/indexer/parsers/src/lib.rs

+2-2
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ pub enum ParseYield<T> {
2525
MessageAndAttachment((T, Attachment)),
2626
}
2727

28-
impl<T> From<T> for ParseYield<T> {
28+
impl<T: Send + Unpin> From<T> for ParseYield<T> {
2929
fn from(item: T) -> Self {
3030
Self::Message(item)
3131
}
@@ -44,7 +44,7 @@ pub trait Parser<T> {
4444
&mut self,
4545
input: &'a [u8],
4646
timestamp: Option<u64>,
47-
) -> Result<(&'a [u8], Option<ParseYield<T>>), Error>;
47+
) -> impl std::future::Future<Output = Result<(&'a [u8], Option<ParseYield<T>>), Error>> + Send;
4848
}
4949

5050
#[derive(Debug, Clone, Serialize)]

Diff for: ‎application/apps/indexer/parsers/src/someip.rs

+25-25
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ unsafe impl Send for SomeipParser {}
4949
unsafe impl Sync for SomeipParser {}
5050

5151
impl Parser<SomeipLogMessage> for SomeipParser {
52-
fn parse<'a>(
52+
async fn parse<'a>(
5353
&mut self,
5454
input: &'a [u8],
5555
timestamp: Option<u64>,
@@ -397,8 +397,8 @@ mod test {
397397
FibexParser::parse(vec![reader]).expect("parse failed")
398398
}
399399

400-
#[test]
401-
fn parse_cookie_client() {
400+
#[tokio::test]
401+
async fn parse_cookie_client() {
402402
let input: &[u8] = &[
403403
0xFF, 0xFF, 0x00, 0x00, // serviceId(u16), methodId(u16)
404404
0x00, 0x00, 0x00, 0x08, // length(u32)
@@ -407,7 +407,7 @@ mod test {
407407
];
408408

409409
let mut parser = SomeipParser::new();
410-
let (output, message) = parser.parse(input, None).unwrap();
410+
let (output, message) = parser.parse(input, None).await.unwrap();
411411

412412
assert!(output.is_empty());
413413

@@ -418,8 +418,8 @@ mod test {
418418
}
419419
}
420420

421-
#[test]
422-
fn parse_cookie_server() {
421+
#[tokio::test]
422+
async fn parse_cookie_server() {
423423
let input: &[u8] = &[
424424
0xFF, 0xFF, 0x80, 0x00, // serviceId(u16), methodId(u16)
425425
0x00, 0x00, 0x00, 0x08, // length(u32)
@@ -428,7 +428,7 @@ mod test {
428428
];
429429

430430
let mut parser = SomeipParser::new();
431-
let (output, message) = parser.parse(input, None).unwrap();
431+
let (output, message) = parser.parse(input, None).await.unwrap();
432432

433433
assert!(output.is_empty());
434434

@@ -439,8 +439,8 @@ mod test {
439439
}
440440
}
441441

442-
#[test]
443-
fn parse_empty_rpc_message_no_model() {
442+
#[tokio::test]
443+
async fn parse_empty_rpc_message_no_model() {
444444
let input: &[u8] = &[
445445
0x01, 0x03, 0x80, 0x04, // serviceId(u16), methodId(u16)
446446
0x00, 0x00, 0x00, 0x08, // length(u32)
@@ -449,7 +449,7 @@ mod test {
449449
];
450450

451451
let mut parser = SomeipParser::new();
452-
let (output, message) = parser.parse(input, None).unwrap();
452+
let (output, message) = parser.parse(input, None).await.unwrap();
453453

454454
assert!(output.is_empty());
455455

@@ -462,8 +462,8 @@ mod test {
462462
}
463463
}
464464

465-
#[test]
466-
fn parse_empty_rpc_message() {
465+
#[tokio::test]
466+
async fn parse_empty_rpc_message() {
467467
let input: &[u8] = &[
468468
0x01, 0x03, 0x80, 0x04, // serviceId(u16), methodId(u16)
469469
0x00, 0x00, 0x00, 0x08, // length(u32)
@@ -474,7 +474,7 @@ mod test {
474474
let model = test_model();
475475

476476
let mut parser = SomeipParser { model: Some(model) };
477-
let (output, message) = parser.parse(input, None).unwrap();
477+
let (output, message) = parser.parse(input, None).await.unwrap();
478478

479479
assert!(output.is_empty());
480480

@@ -487,8 +487,8 @@ mod test {
487487
}
488488
}
489489

490-
#[test]
491-
fn parse_rpc_message_no_model() {
490+
#[tokio::test]
491+
async fn parse_rpc_message_no_model() {
492492
let input: &[u8] = &[
493493
0x01, 0x03, 0x80, 0x05, // serviceId(u16), methodId(u16)
494494
0x00, 0x00, 0x00, 0x0A, // length(u32)
@@ -498,7 +498,7 @@ mod test {
498498
];
499499

500500
let mut parser = SomeipParser::new();
501-
let (output, message) = parser.parse(input, None).unwrap();
501+
let (output, message) = parser.parse(input, None).await.unwrap();
502502

503503
assert!(output.is_empty());
504504

@@ -511,8 +511,8 @@ mod test {
511511
}
512512
}
513513

514-
#[test]
515-
fn parse_rpc_message() {
514+
#[tokio::test]
515+
async fn parse_rpc_message() {
516516
let input: &[u8] = &[
517517
0x01, 0x03, 0x80, 0x05, // serviceId(u16), methodId(u16)
518518
0x00, 0x00, 0x00, 0x0A, // length(u32)
@@ -524,7 +524,7 @@ mod test {
524524
let model = test_model();
525525

526526
let mut parser = SomeipParser { model: Some(model) };
527-
let (output, message) = parser.parse(input, None).unwrap();
527+
let (output, message) = parser.parse(input, None).await.unwrap();
528528

529529
assert!(output.is_empty());
530530

@@ -537,8 +537,8 @@ mod test {
537537
}
538538
}
539539

540-
#[test]
541-
fn parse_empty_sd_message() {
540+
#[tokio::test]
541+
async fn parse_empty_sd_message() {
542542
let input: &[u8] = &[
543543
0xFF, 0xFF, 0x81, 0x00, // serviceId(u16), methodId(u16)
544544
0x00, 0x00, 0x00, 0x14, // length(u32)
@@ -550,7 +550,7 @@ mod test {
550550
];
551551

552552
let mut parser = SomeipParser::new();
553-
let (output, message) = parser.parse(input, None).unwrap();
553+
let (output, message) = parser.parse(input, None).await.unwrap();
554554

555555
assert!(output.is_empty());
556556

@@ -563,8 +563,8 @@ mod test {
563563
}
564564
}
565565

566-
#[test]
567-
fn parse_sd_message() {
566+
#[tokio::test]
567+
async fn parse_sd_message() {
568568
let input: &[u8] = &[
569569
0xFF, 0xFF, 0x81, 0x00, // serviceId(u16), methodId(u16)
570570
0x00, 0x00, 0x00, 0x40, // length(u32)
@@ -592,7 +592,7 @@ mod test {
592592
];
593593

594594
let mut parser = SomeipParser::new();
595-
let (output, message) = parser.parse(input, None).unwrap();
595+
let (output, message) = parser.parse(input, None).await.unwrap();
596596

597597
assert!(output.is_empty());
598598

0 commit comments

Comments
 (0)