HISTORY QoS Problem

[jwcesign]

Hi,
In my expriment, I want to test HISTORY QoS, but when I test, it seems not work as I want.
Here is the description of HISTORY QoS in OMG specification.

If the kind is set to KEEP_ALL, then the Service will attempt to maintain and deliver all the values of the instance to existing subscribers. The resources that the Service can use to keep this history are limited by the settings of the RESOURCE_LIMITS QoS. If the limit is reached, then the behavior of the Service will depend on the RELIABILITY QoS. If the reliability kind is BEST_EFFORT, then the old values will be discarded. If reliability is RELIABLE, then the Service will block the DataWriter until it can deliver the necessary old values to all subscribers.

So I set the datawriter QoS as belows and change helloworld examples

#dw qos
    dwQos = dds_qos_create ();
    dds_qset_reliability (dwQos, DDS_RELIABILITY_RELIABLE, DDS_INFINITY);
    dds_qset_history (dwQos, DDS_HISTORY_KEEP_ALL, 0);
    dds_qset_resource_limits (dwQos, 100, DDS_LENGTH_UNLIMITED, DDS_LENGTH_UNLIMITED);
#my idl
	module HelloWorldData
	{
	  struct Msg
	  {
	    long userID;
	    string message;
	  };
	  #pragma keylist Msg userID
	};
#my write opretion
    msg.userID = 1;
    char s;
    printf ("=== [Publisher]  Writing : ");
    for (int i=0;i<50;i++)
    {
        s = (char)i;
        msg.message = &s;
        ret = dds_write (writer, &msg);
        DDS_ERR_CHECK (ret, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
        printf ("Message (%d, %d)\n", msg.userID, *msg.message);
        dds_sleepfor (DDS_MSECS (1000));
    }

As I think, all the msgs I write belongs to the same instance, so all the samples will be saved until all the samples are received by the subscriber's reader.
And I do expriment as belows:
(1)Run two docker containers and bridge to the same network interface---docker0.
(2)container_1 run HelloworldPublisher, then container_2 run HelloworldSubscriber.
(3)container_2 will receive message and till message 20, I use ifconfig docker0 down to turn off the network interface.
(4)container_2 will stop receiving data.
(5)I use ifconfig docker0 up to turn on the networkinterface.
(6)container_2 will start to receive message again, but the message starts from message 28.

So my problem is that container_2 doesn't receive message21-27, but in the specification, it said should receive all.

[eboasson]

Hi,

The key word here is “existing subscribers”: the other favourite term in the specs is “in steady state”. All this is with handwaving as to what these terms actually mean, and I can assure you that you are not the first to be caught be them. In a case like this, all implementations I know of interpret this to mean that the guarantee holds only while the writer and the reader have discovered each other.

Part of discovery is that each DDS participant declares how often, at minimum, it will send messages to maintain liveliness — a.k.a. “renew its lease” — and, in consequence, that other participants should consider it dead once no message has been received from it for that long — a.k.a. “lease expiry”. So, when you turn off the network interface in the experiment, at some point the leases expire because no messages are coming through anymore and all knowledge of the remote participant is removed.

As long as the writer considers the reader alive, it will hold on to the data until it receives an acknowledgement (in line with the sentence you quoted), but once it considers it dead it no longer needs an acknowledgement from that reader and can happily continue, only taking the remaining readers into account when deciding whether samples can be removed from the writer history.

You could argue that the subscriber still exists, and that therefore the data should be retained; but you can’t really do that because you don’t know if it will ever return, and so when the lease expires there is no record left the reader ever existed. Messaging systems with guaranteed delivery (like MQ series, or Kafka to some extent) have a different take on this.

That however, is not all there is to it.

DDS on the one hand gives you reliability between known/discovered/matching/live writers/readers (that’s “reliability” for whether or not retransmits will be requested for lost samples, and “history” for what data will be retained for retransmitting); and on the other hand gives you a mechanism for dealing with keeping data around for readers that join later on, the “durability” QoS.

Once the lease expires in your example, all knowledge of the reader is dropped, and consequently, when the connection is re-established, the reader re-joins from a system-level perspective, but as the writer has no memory of it anymore, to the writer it simply is a late joining reader.

If you were to set the durability QoS to “transient-local”, then the writer will be retaining samples for late-joining readers. On re-establishing the connection, the reader will synchronise with the writer by requesting all available old samples. What samples the writer retains for this purpose is controlled by the “durability service” QoS — a separate history kind, depth and resource limits. (Fair warning: I don’t think the resource limits are fully implemented yet in Cyclone DDS, but the history is.)

The upshot is that you can configure, e.g., “keep-all” for the “history” and “keep-last 100” for the “durability service history”. With those settings, a late joining (or re-joining) reader will always be able to obtain the last 100 samples, while all connected readers will receive everything. (If you’d set it the durability service history to “keep-all”, the writer would retain all data ever written, which might get a little tricky with memory usage.)

That said, should you try this, then you might get surprised again — then again, you might not, because it is actually fairly obvious that more likely than not some of the samples you read before the disconnecting will be read again after reconnecting. There is, after all, simply no record left of where you were in the sequence …

Just to make sure this response to your question really gives too much detail: there are two points in the receive path in which a sample can be dropped because it is too old. The first is when the sequence number is no greater than the last delivered sequence number — but the disappearance of the writer removes the knowledge of this sequence number.

The second is the timestamp of the latest sample available for the instance in the data reader history. If the “destination order” QoS is set to “by source timestamp”, then any samples arriving with an older timestamp will be dropped instead of added to the history. However, if you “take” samples (and you have little choice when you are trying to create a queue), then taking the last sample from an instance when there are no “live” writers left for that instance will cause the reader history to drop that instance. And with that, that second line of defence also falls. (E.g., see the instance lifecycle diagram in the spec, figure 2-11, “Statechart of the instance_state and view_state for a single instance”.)

What to do? You basically have two options at this point. One option is to solve it in the application logic; the other is to make the lease duration longer, so that the disconnection does not result in lease expiry. Which of the two is the better option depends on what you want to achieve.

Longer term, it should become possible for an application to have that queueing behaviour (so take the samples after reading), while also being able to keep the instance around until the application decides it is time for it to go. There are various ways one might go about achieving that, with the least intrusive one probably to implement sufficiently expressive query conditions — that is, more expressive than the query conditions from the DDS specification. But I that is a topic for another day …

[jwcesign]

HI,
As your advice, I check the source code and test, the default lease duration is qos->liveliness.lease_duration = nn_to_ddsi_duration (DDS_INFINITY);. In my expriment, I didn't change this qos setting, and I am sure the time between down and up network interface is not over 5 seconds, which means the participant does not expire. But samples still lost.

[jwcesign]

Hi, I solve this problem, your answer is right, I think I do something wrong with my examples. Can u tell me what qos(es) are implemented in cycloneDDS now? Thanks a lot.（Is deadline QoS implemented ? If no, How can I implement it? As for the deadline QoS experiment, I did as blows）:

  // qos
  dds_qos_t *qos = dds_qos_create();
  dds_reliability_kind_t reliability_kind = DDS_RELIABLE_RELIABILITY_QOS;
  dds_qset_reliability(qos, reliability_kind, duration);
  dds_qset_deadline(qos, DDS_SECS(1));                         // the deadline is 1 second
  //data writer
  writer = dds_create_writer(participant, topic, qos, NULL);
  //write oprtation
  for (int i=0; i<100; i++)
  {
    s = (char)i;
    msg.message = &s;
    printf("Message (%d, %d)\n", msg.userID, *msg.message);
    printf("[qostest]writer status:%d--%d\n", status_writer.total_count, matched_status_writer.total_count);
    dds_get_offered_deadline_missed_status(writer, &status_writer);
    dds_get_publication_matched_status(writer, &matched_status_writer);
    ret = dds_write(writer, &msg);
    DDS_ERR_CHECK (ret, DDS_CHECK_REPORT | DDS_CHECK_EXIT);
    dds_sleepfor(DDS_SECS (20));    // Here to sleep 20 seconds
  }

The result is status_writer.total_count is always 0.

[eboasson]

Hi,

Great that you managed to work around this:

As your advice, I check the source code and test, the default lease duration is qos->liveliness.lease_duration = nn_to_ddsi_duration (DDS_INFINITY);. In my expriment, I didn't change this qos setting, and I am sure the time between down and up network interface is not over 5 seconds, which means the participant does not expire. But samples still lost.

but the fact that the value of liveliness.lease_duration is completely ignored is not good. I've created #58 to not lose sight of it.

[eboasson]

Can u tell me what qos(es) are implemented in cycloneDDS now?

Good question ... definitely not all of them, though today you can set them all to valid values without getting an "unsupported" back (made #59 as a reminder), and it will all be properly communicated and taken into account when doing reader/writer matching.

Fully supported:

• USER_DATA, TOPIC_DATA, GROUP_DATA
• OWNERSHIP, OWNERSHIP_STRENGTH
• PARTITION
• RELIABILITY
• DESTINATION_ORDER
• HISTORY
• WRITER_DATA_LIFECYCLE

Partially supported:

• DURABILITY: volatile and transient-local, only
• DURABILITY_SERVICE: history settings yes, but not the service_cleanup_delay, nor the resource limits are respected
• LIVELINESS: only automatic; the participant lease duration is currently set by the configuration file (Domain/Lease/ExpiryTime) and the readers and writers always behave as if it is infinite — so they piggyback onto the participant liveliness
• RESOURCE_LIMITS: respected on the readers only (the writer currently only waits when the flow-control kicks in)

Not meaningfully supported (or ignored):

• PRESENTATION: only the default (INSTANCE, non-coherent, non-ordered)
• DEADLINE: implementing is not so hard; the easiest option would probably be to maintain a time-ordered list of instances with the deadlines and use the lease mechanism for the monitoring liveliness of the remote entities to raise a missed deadline event. An update of an instance moves it to the end of the list, so maintaining order is trivial, and you could simply renew the reader/writer "lease" to expiry at what has become the new earliest deadline. You could miss a deadline expiry if an update happens after the deadline expired but before the lease is checked; checking for lease expiry in the update code would help deal with that
• LATENCY_BUDGET: a proper implementation is desperately needed ... it can pack samples into messages without any problem (there is a somewhat useful global "batching" setting and a flush function). I think the most practical way would be to have the timed-event thread do the automatic flushing based on the latency budget, but I haven't worked out the impact of that yet
• TRANSPORT_PRIORITY: only interesting as soon as you can schedule things; there is support in the code for having multiple receive queues and threads with selection based on priority (thanks to OpenSplice's network channels), but here that code is not yet used
• LIFESPAN, READER_DATA_LIFECYCLE: I believe both can be implemented pretty easily once one has DEADLINE implemented.
• TIME_BASED_FILTER: simply dropping the a sample if it is too soon after the preceding one has the interesting effect of not seeing the latest state if the writer suddenly stops writing, so one needs to retain the most recent sample under all circumstances, but without making it visible to the application. Whether that behaviour would be ok or not is up for debate, but the principle of least surprise definitely suggests it is best to make the latest update visible automatically if there is no update for some time. This could be done just-in-time within the reader history cache were it not for the fact that there might never be a trigger that there is data available ... so it probably also requires the same mechanism as DEADLINE.
• ENTITY_FACTORY: you may need to enable many entities, and so behaviour in case of a failure halfway through enabling all those entities needs to be considered; but at least it is not needed very often :)

[jwcesign]

Hi, Thanks for explaining. I find my way to implement DEADLINE based on your advice. Thanks a lot

[eboasson]

👍🏻😁