Skip to content
This repository was archived by the owner on Jul 8, 2022. It is now read-only.

[Backport] Fix race conditions between polling threads and user threads pushing events #665

Merged
merged 3 commits into from
Jan 28, 2020
Merged

[Backport] Fix race conditions between polling threads and user threads pushing events #665

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0af852d
Protect storing last attribute value with mutex
mliszcz Jan 23, 2020
82f1b89
Change mutex used for EventSupplier::detect_change
mliszcz Jan 23, 2020
5f89566
Protect event subscription timestamps with mutex
mliszcz Dec 16, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
190 changes: 110 additions & 80 deletions cppapi/server/attribute.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3774,9 +3774,13 @@ void Attribute::fire_change_event(DevFailed *except)
time_t change3_subscription,change4_subscription,change5_subscription;

now = time(NULL);
change3_subscription = now - event_change3_subscription;
change4_subscription = now - event_change4_subscription;
change5_subscription = now - event_change5_subscription;

{
omni_mutex_lock oml(EventSupplier::get_event_mutex());
change3_subscription = now - event_change3_subscription;
change4_subscription = now - event_change4_subscription;
change5_subscription = now - event_change5_subscription;
}

//
// Get the event supplier(s)
Expand Down Expand Up @@ -4017,54 +4021,58 @@ void Attribute::fire_change_event(DevFailed *except)
bool force_change = false;
bool quality_change = false;

if ((except != NULL) ||
(quality == Tango::ATTR_INVALID) ||
((except == NULL) && (prev_change_event.err == true)) ||
((quality != Tango::ATTR_INVALID) &&
(prev_change_event.quality == Tango::ATTR_INVALID)))
{
force_change = true;
}

vector<string> filterable_names;
vector<double> filterable_data;
vector<string> filterable_names_lg;
vector<long> filterable_data_lg;

if (except != NULL)
{
prev_change_event.err = true;
prev_change_event.except = *except;
}
else
{
Tango::AttrQuality the_quality;
omni_mutex_lock oml(EventSupplier::get_event_mutex());

if (send_attr_5 != NULL)
if ((except != NULL) ||
(quality == Tango::ATTR_INVALID) ||
((except == NULL) && (prev_change_event.err == true)) ||
((quality != Tango::ATTR_INVALID) &&
(prev_change_event.quality == Tango::ATTR_INVALID)))
{
the_quality = send_attr_5->quality;
prev_change_event.value_4 = send_attr_5->value;
force_change = true;
}
else if (send_attr_4 != NULL)

if (except != NULL)
{
the_quality = send_attr_4->quality;
prev_change_event.value_4 = send_attr_4->value;
prev_change_event.err = true;
prev_change_event.except = *except;
}
else
{
the_quality = send_attr->quality;
prev_change_event.value = send_attr->value;
}
Tango::AttrQuality the_quality;

if (prev_change_event.quality != the_quality)
{
quality_change = true;
}
if (send_attr_5 != NULL)
{
the_quality = send_attr_5->quality;
prev_change_event.value_4 = send_attr_5->value;
}
else if (send_attr_4 != NULL)
{
the_quality = send_attr_4->quality;
prev_change_event.value_4 = send_attr_4->value;
}
else
{
the_quality = send_attr->quality;
prev_change_event.value = send_attr->value;
}

prev_change_event.quality = the_quality;
prev_change_event.err = false;
if (prev_change_event.quality != the_quality)
{
quality_change = true;
}

prev_change_event.quality = the_quality;
prev_change_event.err = false;
}
prev_change_event.inited = true;
}
prev_change_event.inited = true;

filterable_names.push_back("forced_event");
if (force_change == true)
Expand Down Expand Up @@ -4199,9 +4207,12 @@ void Attribute::fire_archive_event(DevFailed *except)

now = time(NULL);

archive3_subscription = now - event_archive3_subscription;
archive4_subscription = now - event_archive4_subscription;
archive5_subscription = now - event_archive5_subscription;
{
omni_mutex_lock oml(EventSupplier::get_event_mutex());
archive3_subscription = now - event_archive3_subscription;
archive4_subscription = now - event_archive4_subscription;
archive5_subscription = now - event_archive5_subscription;
}

//
// Get the event supplier(s)
Expand Down Expand Up @@ -4458,60 +4469,73 @@ void Attribute::fire_archive_event(DevFailed *except)
{

//
// Execute detect_change only to calculate the delta_change_rel and
// delta_change_abs and force_change !
//

bool force_change = false;
bool quality_change = false;
double delta_change_rel = 0.0;
double delta_change_abs = 0.0;

if (event_supplier_nd != NULL)
event_supplier_nd->detect_change(*this, ad,true,delta_change_rel,delta_change_abs,except,force_change,dev);
else if (event_supplier_zmq != NULL)
event_supplier_zmq->detect_change(*this, ad,true,delta_change_rel,delta_change_abs,except,force_change,dev);


vector<string> filterable_names;
vector<double> filterable_data;
vector<string> filterable_names_lg;
vector<long> filterable_data_lg;

if (except != NULL)
{
prev_archive_event.err = true;
prev_archive_event.except = *except;
}
else
{
Tango::AttrQuality the_quality;
omni_mutex_lock oml(EventSupplier::get_event_mutex());

if (send_attr_5 != Tango_nullptr)
// Execute detect_change only to calculate the delta_change_rel and
// delta_change_abs and force_change !

if (event_supplier_nd || event_supplier_zmq)
{
prev_archive_event.value_4 = send_attr_5->value;
the_quality = send_attr_5->quality;
EventSupplier* event_supplier = event_supplier_nd ? event_supplier_nd : event_supplier_zmq;
event_supplier->detect_change(
*this,
ad,
true,
delta_change_rel,
delta_change_abs,
except,
force_change,
dev);
}
else if (send_attr_4 != Tango_nullptr)

if (except != NULL)
{
prev_archive_event.value_4 = send_attr_4->value;
the_quality = send_attr_4->quality;
prev_archive_event.err = true;
prev_archive_event.except = *except;
}
else
{
prev_archive_event.value = send_attr->value;
the_quality = send_attr->quality;
}
Tango::AttrQuality the_quality;

if (prev_archive_event.quality != the_quality)
{
quality_change = true;
}
if (send_attr_5 != Tango_nullptr)
{
prev_archive_event.value_4 = send_attr_5->value;
the_quality = send_attr_5->quality;
}
else if (send_attr_4 != Tango_nullptr)
{
prev_archive_event.value_4 = send_attr_4->value;
the_quality = send_attr_4->quality;
}
else
{
prev_archive_event.value = send_attr->value;
the_quality = send_attr->quality;
}

if (prev_archive_event.quality != the_quality)
{
quality_change = true;
}

prev_archive_event.quality = the_quality;
prev_archive_event.err = false;
prev_archive_event.quality = the_quality;
prev_archive_event.err = false;
}
prev_archive_event.inited = true;
}
prev_archive_event.inited = true;

vector<string> filterable_names;
vector<double> filterable_data;
vector<string> filterable_names_lg;
vector<long> filterable_data_lg;

filterable_names.push_back("forced_event");
if (force_change == true)
Expand Down Expand Up @@ -4642,9 +4666,12 @@ void Attribute::fire_event(vector<string> &filt_names,vector<double> &filt_vals,

now = time(NULL);

user3_subscription = now - event_user3_subscription;
user4_subscription = now - event_user4_subscription;
user5_subscription = now - event_user5_subscription;
{
omni_mutex_lock oml(EventSupplier::get_event_mutex());
user3_subscription = now - event_user3_subscription;
user4_subscription = now - event_user4_subscription;
user5_subscription = now - event_user5_subscription;
}

//
// Get the event supplier(s)
Expand Down Expand Up @@ -4914,9 +4941,12 @@ void Attribute::fire_error_periodic_event(DevFailed *except)

now = time(NULL);

periodic3_subscription = now - event_periodic3_subscription;
periodic4_subscription = now - event_periodic4_subscription;
periodic5_subscription = now - event_periodic5_subscription;
{
omni_mutex_lock oml(EventSupplier::get_event_mutex());
periodic3_subscription = now - event_periodic3_subscription;
periodic4_subscription = now - event_periodic4_subscription;
periodic5_subscription = now - event_periodic5_subscription;
}

vector<int> client_libs = get_client_lib(PERIODIC_EVENT); // We want a copy

Expand Down
6 changes: 0 additions & 6 deletions cppapi/server/eventsupplier.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1182,12 +1182,6 @@ bool EventSupplier::detect_change(Attribute &attr, struct SuppliedEventData &att
the_new_any = &(attr_value.attr_val->value);
}

//
// get the mutex to synchronize the sending of events
//

omni_mutex_lock l(detect_mutex);

//
// Send event, if the read_attribute failed or if it is the first time that the read_attribute succeed after a failure.
// Same thing if the attribute quality factor changes to INVALID
Expand Down
4 changes: 1 addition & 3 deletions cppapi/server/eventsupplier.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,9 +174,7 @@ protected :
static omni_mutex push_mutex;
static omni_condition push_cond;

// Added a mutex to synchronize the access to
// detect_event which is used
// from different threads
// Unused, replaced with event_mutex. Left for backward compatibility.
static omni_mutex detect_mutex;

private:
Expand Down