niklas/cyclonedds
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

missing m_observer_lock on (re)setting statuses

Browse source 
Most of the places where the status flags were reset, this happened
without holding m_observer_lock protecting these status flags.  For most
of these statuses, they are only ever set/reset while also holding the
entity lock, but this is not true for all of them (DATA_AVAILABLE for
example), and thus there are some cases where retrieving the status
could lead to losing the raising of a (at least a DATA_AVAILABLE)
status.

The problem was introduced in ba46cb1140.

Signed-off-by: Erik Boasson <eb@ilities.com>
This commit is contained in:
Erik Boasson 2019-04-20 17:17:47 +02:00 • committed by eboasson
parent 1a3d5c7aba
commit 46f61e09f5
4 changed files with 28 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

6
src/core/ddsc/src/dds_read.c
View file

@ -159,11 +159,14 @@ dds_read_impl(
/* read/take resets data available status -- must reset before reading because
the actual writing is protected by RHC lock, not by rd->m_entity.m_lock */
ddsrt_mutex_lock (&rd->m_entity.m_observers_lock);
dds_entity_status_reset (&rd->m_entity, DDS_DATA_AVAILABLE_STATUS);
/* reset DATA_ON_READERS status on subscriber after successful read/take */
if (dds_entity_kind (rd->m_entity.m_parent) == DDS_KIND_SUBSCRIBER) {
dds_entity_status_reset (rd->m_entity.m_parent, DDS_DATA_ON_READERS_STATUS);
}
ddsrt_mutex_unlock (&rd->m_entity.m_observers_lock);
if (take) {
ret = (dds_return_t)dds_rhc_take(rd->m_rd->rhc, lock, buf, si, maxs, mask, hand, cond);
} else {
@ -206,11 +209,14 @@ dds_readcdr_impl(
if (rc == DDS_RETCODE_OK) {
/* read/take resets data available status -- must reset before reading because
the actual writing is protected by RHC lock, not by rd->m_entity.m_lock */
ddsrt_mutex_lock (&rd->m_entity.m_observers_lock);
dds_entity_status_reset (&rd->m_entity, DDS_DATA_AVAILABLE_STATUS);
/* reset DATA_ON_READERS status on subscriber after successful read/take */
if (dds_entity_kind (rd->m_entity.m_parent) == DDS_KIND_SUBSCRIBER) {
dds_entity_status_reset (rd->m_entity.m_parent, DDS_DATA_ON_READERS_STATUS);
}
ddsrt_mutex_unlock (&rd->m_entity.m_observers_lock);
ret = dds_rhc_takecdr
(
rd->m_rd->rhc, lock, buf, si, maxs,

12
src/core/ddsc/src/dds_reader.c
View file

@ -643,11 +643,13 @@ dds_get_subscription_matched_status (
if (status) {
*status = rd->m_subscription_matched_status;
}
ddsrt_mutex_lock (&rd->m_entity.m_observers_lock);
if (rd->m_entity.m_status_enable & DDS_SUBSCRIPTION_MATCHED_STATUS) {
rd->m_subscription_matched_status.total_count_change = 0;
rd->m_subscription_matched_status.current_count_change = 0;
dds_entity_status_reset(&rd->m_entity, DDS_SUBSCRIPTION_MATCHED_STATUS);
}
ddsrt_mutex_unlock (&rd->m_entity.m_observers_lock);
dds_reader_unlock(rd);
fail:
return ret;
@ -672,11 +674,13 @@ dds_get_liveliness_changed_status (
if (status) {
*status = rd->m_liveliness_changed_status;
}
ddsrt_mutex_lock (&rd->m_entity.m_observers_lock);
if (rd->m_entity.m_status_enable & DDS_LIVELINESS_CHANGED_STATUS) {
rd->m_liveliness_changed_status.alive_count_change = 0;
rd->m_liveliness_changed_status.not_alive_count_change = 0;
dds_entity_status_reset(&rd->m_entity, DDS_LIVELINESS_CHANGED_STATUS);
}
ddsrt_mutex_unlock (&rd->m_entity.m_observers_lock);
dds_reader_unlock(rd);
fail:
return ret;
@ -700,11 +704,13 @@ dds_return_t dds_get_sample_rejected_status (
if (status) {
*status = rd->m_sample_rejected_status;
}
ddsrt_mutex_lock (&rd->m_entity.m_observers_lock);
if (rd->m_entity.m_status_enable & DDS_SAMPLE_REJECTED_STATUS) {
rd->m_sample_rejected_status.total_count_change = 0;
rd->m_sample_rejected_status.last_reason = DDS_NOT_REJECTED;
dds_entity_status_reset(&rd->m_entity, DDS_SAMPLE_REJECTED_STATUS);
}
ddsrt_mutex_unlock (&rd->m_entity.m_observers_lock);
dds_reader_unlock(rd);
fail:
return ret;
@ -728,10 +734,12 @@ dds_return_t dds_get_sample_lost_status (
if (status) {
*status = rd->m_sample_lost_status;
}
ddsrt_mutex_lock (&rd->m_entity.m_observers_lock);
if (rd->m_entity.m_status_enable & DDS_SAMPLE_LOST_STATUS) {
rd->m_sample_lost_status.total_count_change = 0;
dds_entity_status_reset(&rd->m_entity, DDS_SAMPLE_LOST_STATUS);
}
ddsrt_mutex_unlock (&rd->m_entity.m_observers_lock);
dds_reader_unlock(rd);
fail:
return ret;
@ -755,10 +763,12 @@ dds_return_t dds_get_requested_deadline_missed_status (
if (status) {
*status = rd->m_requested_deadline_missed_status;
}
ddsrt_mutex_lock (&rd->m_entity.m_observers_lock);
if (rd->m_entity.m_status_enable & DDS_REQUESTED_DEADLINE_MISSED_STATUS) {
rd->m_requested_deadline_missed_status.total_count_change = 0;
dds_entity_status_reset(&rd->m_entity, DDS_REQUESTED_DEADLINE_MISSED_STATUS);
}
ddsrt_mutex_unlock (&rd->m_entity.m_observers_lock);
dds_reader_unlock(rd);
fail:
return ret;
@ -782,10 +792,12 @@ dds_return_t dds_get_requested_incompatible_qos_status (
if (status) {
*status = rd->m_requested_incompatible_qos_status;
}
ddsrt_mutex_lock (&rd->m_entity.m_observers_lock);
if (rd->m_entity.m_status_enable & DDS_REQUESTED_INCOMPATIBLE_QOS_STATUS) {
rd->m_requested_incompatible_qos_status.total_count_change = 0;
dds_entity_status_reset(&rd->m_entity, DDS_REQUESTED_INCOMPATIBLE_QOS_STATUS);
}
ddsrt_mutex_unlock (&rd->m_entity.m_observers_lock);
dds_reader_unlock(rd);
fail:
return ret;

2
src/core/ddsc/src/dds_topic.c
View file

@ -669,10 +669,12 @@ dds_get_inconsistent_topic_status(
if (status) {
*status = t->m_inconsistent_topic_status;
}
ddsrt_mutex_lock (&t->m_entity.m_observers_lock);
if (t->m_entity.m_status_enable & DDS_INCONSISTENT_TOPIC_STATUS) {
t->m_inconsistent_topic_status.total_count_change = 0;
dds_entity_status_reset(&t->m_entity, DDS_INCONSISTENT_TOPIC_STATUS);
}
ddsrt_mutex_unlock (&t->m_entity.m_observers_lock);
dds_topic_unlock(t);
fail:
return ret;

8
src/core/ddsc/src/dds_writer.c
View file

@ -518,11 +518,13 @@ dds_get_publication_matched_status (
if (status) {
*status = wr->m_publication_matched_status;
}
ddsrt_mutex_lock (&wr->m_entity.m_observers_lock);
if (wr->m_entity.m_status_enable & DDS_PUBLICATION_MATCHED_STATUS) {
wr->m_publication_matched_status.total_count_change = 0;
wr->m_publication_matched_status.current_count_change = 0;
dds_entity_status_reset(&wr->m_entity, DDS_PUBLICATION_MATCHED_STATUS);
}
ddsrt_mutex_unlock (&wr->m_entity.m_observers_lock);
dds_writer_unlock(wr);
fail:
return ret;
@ -547,10 +549,12 @@ dds_get_liveliness_lost_status (
if (status) {
*status = wr->m_liveliness_lost_status;
}
ddsrt_mutex_lock (&wr->m_entity.m_observers_lock);
if (wr->m_entity.m_status_enable & DDS_LIVELINESS_LOST_STATUS) {
wr->m_liveliness_lost_status.total_count_change = 0;
dds_entity_status_reset(&wr->m_entity, DDS_LIVELINESS_LOST_STATUS);
}
ddsrt_mutex_unlock (&wr->m_entity.m_observers_lock);
dds_writer_unlock(wr);
fail:
return ret;
@ -575,10 +579,12 @@ dds_get_offered_deadline_missed_status(
if (status) {
*status = wr->m_offered_deadline_missed_status;
}
ddsrt_mutex_lock (&wr->m_entity.m_observers_lock);
if (wr->m_entity.m_status_enable & DDS_OFFERED_DEADLINE_MISSED_STATUS) {
wr->m_offered_deadline_missed_status.total_count_change = 0;
dds_entity_status_reset(&wr->m_entity, DDS_OFFERED_DEADLINE_MISSED_STATUS);
}
ddsrt_mutex_unlock (&wr->m_entity.m_observers_lock);
dds_writer_unlock(wr);
fail:
return ret;
@ -603,10 +609,12 @@ dds_get_offered_incompatible_qos_status (
if (status) {
*status = wr->m_offered_incompatible_qos_status;
}
ddsrt_mutex_lock (&wr->m_entity.m_observers_lock);
if (wr->m_entity.m_status_enable & DDS_OFFERED_INCOMPATIBLE_QOS_STATUS) {
wr->m_offered_incompatible_qos_status.total_count_change = 0;
dds_entity_status_reset(&wr->m_entity, DDS_OFFERED_INCOMPATIBLE_QOS_STATUS);
}
ddsrt_mutex_unlock (&wr->m_entity.m_observers_lock);
dds_writer_unlock(wr);
fail:
return ret;