tot0k
commented
10 months ago With Keyed topics, when I publish with a first "early" writer (that doesn't wait for the DataReader to be ready), then with a second "late" writer (that waits some time before starting publishing), I see a lag on the samples published by the early writer, and no lag for the late one:
Creating DomainParticipant
Creating subscriber
Creating data_writer
Starting Subscriber thread
Starting Publishing
Publishing Early: 0
DDS receiver thread started
Publishing Early: 1
Publishing Early: 2
Publishing Early: 3
Publishing Early: 4
DataReader event: SubscriptionMatched { total: CountWithChange { count: 1, count_change: 1 }, current: CountWithChange { count: 1, count_change: 1 } }
DataReader event: SubscriptionMatched { total: CountWithChange { count: 2, count_change: 1 }, current: CountWithChange { count: 2, count_change: 1 } }
Publishing Early: 5
Publishing Early: 6
Received: Value(ExampleMessage { id: 1, count: 4 })
Publishing Early: 7
Received: Value(ExampleMessage { id: 1, count: 5 })
Publishing Early: 8
Received: Value(ExampleMessage { id: 1, count: 6 })
Publishing Early: 9
Publishing Late: 1000
Received: Value(ExampleMessage { id: 1, count: 7 })
Received: Value(ExampleMessage { id: 2, count: 1000 })
Publishing Early: 10
Publishing Late: 1001
Received: Value(ExampleMessage { id: 1, count: 8 })
Received: Value(ExampleMessage { id: 2, count: 1001 })
Publishing Early: 11
Publishing Late: 1002
Received: Value(ExampleMessage { id: 1, count: 9 })
Received: Value(ExampleMessage { id: 2, count: 1002 })
Publishing Early: 12
Publishing Late: 1003
Received: Value(ExampleMessage { id: 1, count: 10 })
Received: Value(ExampleMessage { id: 2, count: 1003 })
Publishing Early: 13
Publishing Late: 1004
Received: Value(ExampleMessage { id: 1, count: 11 })
Received: Value(ExampleMessage { id: 2, count: 1004 })
Publishing Early: 14
Publishing Late: 1005
Received: Value(ExampleMessage { id: 1, count: 12 })
Received: Value(ExampleMessage { id: 2, count: 1005 })Code
```rust
use futures::{executor, select, FutureExt, StreamExt};
use rustdds::{
with_key::{DataReader, DataWriter},
policy::Reliability,
policy::{Durability, History},
DomainParticipant, Duration, QosPolicies, QosPolicyBuilder, TopicKind, Keyed,
};
use serde::{Deserialize, Serialize};
use std::thread;
#[derive(Serialize, Deserialize, Debug, Default, Clone)]
struct ExampleMessage {
id: u8,
count: u32,
}
impl Keyed for ExampleMessage {
type K = u8;
fn key(&self) -> Self::K {
self.id
}
}
static DOMAIN_ID: u16 = 0;
/// Default QoS settings
fn qos() -> QosPolicies {
QosPolicyBuilder::new()
//.durability(Durability::TransientLocal)
//.history(History::KeepLast { depth: 10 })
.reliability(Reliability::Reliable {
max_blocking_time: Duration::DURATION_INFINITE,
})
.build()
}
/// Create a publisher return its writer.
fn create_data_writer
SelimV
Problem description
With some QoS settings, when sending samples before a
DataReaderis ready, i see the following behavior:We notice that after the
SubscriptionMatchedevent appears, we receive the samples with a lag of 2 that stays for the wholeDataWriterlife.Here are the QoS settings I tested, and the behavior I observed
nsamples are lost, then the acquisition is working properly without lag.nsamples are received all at once when the reader is ready, then the acquisition is working properly without lagnsamples are lost, then acquisition lag of 2 samples.nsamples are received all at once when the reader is ready, then the acquisition is working properly without lagIt seems that when using the Reliability QoS without
History, we experience a 2 samples lag in the reader.I can't understand this lag behavior. Is it normal ?
Reproducible example
Comment out lines in the
qos()function to change QoS settings.Code
```rust use futures::{executor, select, FutureExt, StreamExt}; use rustdds::{ no_key::{DataReader, DataWriter}, policy::Reliability, policy::{Durability, History}, DomainParticipant, Duration, QosPolicies, QosPolicyBuilder, TopicKind, }; use serde::{Deserialize, Serialize}; use std::thread; #[derive(Serialize, Deserialize, Debug, Default, Clone)] struct ExampleMessage { count: u32, } static DOMAIN_ID: u16 = 0; /// Default QoS settings fn qos() -> QosPolicies { QosPolicyBuilder::new() //.durability(Durability::TransientLocal) .history(History::KeepLast { depth: 10 }) .reliability(Reliability::Reliable { max_blocking_time: Duration::DURATION_INFINITE, }) .build() } /// Create a publisher return its writer. fn create_data_writer(
ctx: &DomainParticipant,
topic_name: &str,
) -> DataWriter {
let topic_type = std::any::type_name::();
let topic = ctx
.create_topic(
topic_name.to_string(),
topic_type.to_string(),
&qos(),
TopicKind::NoKey,
)
.unwrap();
let publisher = ctx.create_publisher(&qos()).unwrap();
publisher
.create_datawriter_no_key_cdr::(&topic, Some(qos()))
.unwrap()
}
///Publish a sample using the given data_writer
fn publish(
data_writer: &DataWriter,
sample: TopicType,
) {
let result = data_writer.write(sample, None);
result
.map_err(|e| println!("Failed to publish: {:?}", e))
.ok();
}
/// Create a subscriber and return its reader
fn create_data_reader Deserialize<'de> + 'static>(
ctx: &DomainParticipant,
topic_name: &str,
) -> DataReader {
let topic_type = std::any::type_name::();
let topic = ctx
.create_topic(
topic_name.to_string(),
topic_type.to_string(),
&qos(),
TopicKind::NoKey,
)
.unwrap();
let subscriber = ctx.create_subscriber(&qos()).unwrap();
subscriber
.create_datareader_no_key_cdr::(&topic, Some(qos()))
.unwrap()
}
/// Create a domain participant, a publisher, and a subscriber.
/// Listen on the data_reader with an async sample stream in another thread and
/// publish every 0.5 sec on the main thread.
fn main() {
println!("Creating DomainParticipant");
let ctx = DomainParticipant::new(DOMAIN_ID).unwrap();
println!("Creating subscriber");
let data_reader = create_data_reader::(&ctx, "hello/world");
println!("Creating data_writer");
let data_writer = create_data_writer::(&ctx, "hello/world");
println!("Starting Subscriber thread");
// Run single thread with a future select
let _ = thread::Builder::new()
.name("ReceiverThread".to_string())
.spawn(move || {
println!("DDS receiver thread started");
executor::block_on(async_infinite_loop(data_reader));
println!("DDS receiver thread stopped");
})
.unwrap();
println!("Starting Publishing");
let mut sample = ExampleMessage { count: 0 };
loop {
println!("Publishing: {}", sample.count);
publish(&data_writer, sample.clone());
sample.count += 1;
std::thread::sleep(std::time::Duration::from_millis(500));
}
}
async fn async_infinite_loop(data_reader: DataReader) {
// Create the async streams
let mut dr_stream = data_reader.async_sample_stream();
let mut dr_event_stream = dr_stream.async_event_stream();
loop {
// Create the futures
let next_some_dr_stream = dr_stream.select_next_some();
let next_some_dr_event_stream = dr_event_stream.select_next_some();
select! {
result = next_some_dr_stream.fuse() => {
match result {
Ok(sample) => println!("Received: {:}", sample.count),
Err(error) => println!("Error on sample read: {:?}", error)
}
}
result = next_some_dr_event_stream.fuse() => {
match result {
Ok(event) => println!("DataReader event: {event:?}"),
Err(error) => println!("Error while on reader event: {}", error)
}
}
};
}
}
```
Cargo.toml
```toml [package] name = "rdds-302" version = "0.1.0" edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] futures = "0.3.29" rustdds = "0.8.4" serde = "1.0.190" ```
I also tested on latest RustDDS commit on master, this didn't change the behavior.
Packet captures and logs of each QoS setup
Default QoS settings
rdds-302_default_qos.zip
History only
rdds-302_history_only.zip
Reliable only
rdds-302_reliable_only.zip
Reliable with History
rdds-302_reliable_history.zip