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

Fix DATA_AVAILABLE race condition

Browse source 
The DATA_AVAILABLE status was reset by read and take while holding the
upper-layer reader lock, but after completing the read/take operation on
the RHC.  As data can be written into the RHC without holding the
upper-layer reader lock, new data could arrive in between the
reading/taking and the resetting of the DATA_AVAILABLE status, leading
to a missed detection.  Resetting DATA_AVAILABLE prior to accessing the
RHC solves this.

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

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

@ -156,17 +156,19 @@ dds_read_impl(
rd->m_loan_out = true; rd->m_loan_out = true;
} }
} }
if (take) {
ret = (dds_return_t)dds_rhc_take(rd->m_rd->rhc, lock, buf, si, maxs, mask, hand, cond); /* read/take resets data available status -- must reset before reading because
} else { the actual writing is protected by RHC lock, not by rd->m_entity.m_lock */
ret = (dds_return_t)dds_rhc_read(rd->m_rd->rhc, lock, buf, si, maxs, mask, hand, cond);
}
/* read/take resets data available status */
dds_entity_status_reset (&rd->m_entity, DDS_DATA_AVAILABLE_STATUS); dds_entity_status_reset (&rd->m_entity, DDS_DATA_AVAILABLE_STATUS);
/* reset DATA_ON_READERS status on subscriber after successful read/take */ /* reset DATA_ON_READERS status on subscriber after successful read/take */
if (dds_entity_kind (rd->m_entity.m_parent) == DDS_KIND_SUBSCRIBER) { 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); dds_entity_status_reset (rd->m_entity.m_parent, DDS_DATA_ON_READERS_STATUS);
} }
if (take) {
ret = (dds_return_t)dds_rhc_take(rd->m_rd->rhc, lock, buf, si, maxs, mask, hand, cond);
} else {
ret = (dds_return_t)dds_rhc_read(rd->m_rd->rhc, lock, buf, si, maxs, mask, hand, cond);
}
dds_read_unlock(rd, cond); dds_read_unlock(rd, cond);
fail_awake: fail_awake:
@ -202,6 +204,13 @@ dds_readcdr_impl(
thread_state_awake (ts1); thread_state_awake (ts1);
rc = dds_read_lock(reader_or_condition, &rd, &cond, false); rc = dds_read_lock(reader_or_condition, &rd, &cond, false);
if (rc == DDS_RETCODE_OK) { 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 */
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);
}
ret = dds_rhc_takecdr ret = dds_rhc_takecdr
( (
rd->m_rd->rhc, lock, buf, si, maxs, rd->m_rd->rhc, lock, buf, si, maxs,
@ -211,15 +220,6 @@ dds_readcdr_impl(
hand hand
); );
/* read/take resets data available status */
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);
}
dds_read_unlock(rd, cond); dds_read_unlock(rd, cond);
} else { } else {
ret = DDS_ERRNO(rc); ret = DDS_ERRNO(rc);