Merge pull request #665 from mliszcz/backport-fix-511-crash-race-condition-event-push

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

Author: t-b

cppapi/server/attribute.cpp

@@ -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)
@@ -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)
@@ -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)
@@ -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)
@@ -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)
@@ -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
 

cppapi/server/eventsupplier.cpp

@@ -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

cppapi/server/eventsupplier.h

@@ -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: