Lots of thread safety

Author: Kelwan

ecsact/runtime/async.h

@@ -77,9 +77,10 @@ typedef enum {
  */
 typedef void (*ecsact_async_error_callback)(
 	//
-	ecsact_async_error      async_err,
-	ecsact_async_request_id request_id,
-	void*                   callback_user_data
+	ecsact_async_error       async_err,
+	int                      request_ids_length,
+	ecsact_async_request_id* request_ids,
+	void*                    callback_user_data
 );
 
 /**

reference/async_reference/async_reference.cc

@@ -10,19 +10,20 @@ ecsact_async_request_id async_reference::connect(const char* connection_string
 ) {
 	std::string connect_str(connection_string);
 
+	registry_id = ecsact_create_registry("async_reference_impl_reg");
+	pending_registry_id =
+		ecsact_create_registry("pending_async_reference_impl_reg");
+
 	auto req_id = next_request_id();
 	// The good and bad strings simulate the outcome of connections
 	if(connect_str == "good") {
 		is_connected = true;
 
-		registry_id = ecsact_create_registry("async_reference_impl_reg");
-		pending_registry_id =
-			ecsact_create_registry("pending_async_reference_impl_reg");
-
 		execute_systems();
 	} else {
 		// Same thing that happens in enqueue? Callback next flush?
 		is_connected = false;
+		is_connected_notified = false;
 	}
 
 	return req_id;
@@ -33,57 +34,43 @@ ecsact_async_request_id async_reference::enqueue_execution_options(
 ) {
 	auto req_id = next_request_id();
 
-	if(is_connected == false) {
+	if(is_connected == false && is_connected_notified == false) {
 		types::async_error async_err{
 			.error = ECSACT_ASYNC_ERR_PERMISSION_DENIED,
-			.request_id = req_id};
-
-		// ecsact_add_component(
-		// 	*pending_registry_id,
-		// 	entity_id,
-		// 	component_id,
-
-		// )
+			.request_ids = {req_id},
+		};
 
+		is_connected_notified = true;
 		// Could block here
 		async_callbacks.add(async_err);
 		return req_id;
 	}
 
-	auto cpp_options = util::c_to_cpp_execution_options(options);
-	auto error = util::validate_options(cpp_options);
+	auto cpp_options =
+		util::c_to_cpp_execution_options(options, *pending_registry_id);
 
-	auto async_err = types::async_error{
-		.error = error,
+	types::pending_execution_options pending_options{
 		.request_id = req_id,
+		.options = cpp_options,
 	};
 
-	if(error != ECSACT_ASYNC_OK) {
-		// Could block here
-		async_callbacks.add(async_err);
-
-		disconnect();
-		return req_id;
-	}
-
 	// Could block here
-	error = tick_manager.try_add_options(cpp_options);
-	async_err.error = error;
-
-	if(error != ECSACT_ASYNC_OK) {
-		// Could block here
-		async_callbacks.add(async_err);
-
-		disconnect();
-		return req_id;
-	}
+	tick_manager.add_pending_options(pending_options);
 	return req_id;
 }
 
 void async_reference::execute_systems() {
 	execution_thread = std::thread([this] {
 		while(is_connected == true) {
 			// Could block here
+			auto async_err = tick_manager.validate_pending_options();
+
+			if(async_err.error != ECSACT_ASYNC_OK) {
+				async_callbacks.add(async_err);
+
+				disconnect();
+			}
+
 			auto cpp_options = tick_manager.get_options_now();
 
 			ecsact_execution_events_collector collector;
@@ -99,22 +86,14 @@ void async_reference::execute_systems() {
 
 			if(cpp_options) {
 				options = std::make_unique<ecsact_execution_options>(
-					util::cpp_to_c_execution_options(*cpp_options, *registry_id)
+					util::cpp_to_c_execution_options(
+						*cpp_options,
+						*registry_id,
+						*pending_registry_id
+					)
 				);
 			}
 
-			auto systems_error =
-				ecsact_execute_systems(*registry_id, 1, options.get(), &collector);
-
-			if(systems_error != ECSACT_EXEC_SYS_OK) {
-				async_callbacks.add(systems_error);
-				disconnect();
-				return;
-			}
-
-			// Could block here
-			tick_manager.increment_and_merge_tick();
-
 			std::vector<ecsact_async_request_id> pending_entities;
 
 			// Could block here
@@ -134,6 +113,15 @@ void async_reference::execute_systems() {
 				// Could block here
 				async_callbacks.add(created_entity);
 			}
+
+			auto systems_error =
+				ecsact_execute_systems(*registry_id, 1, options.get(), &collector);
+
+			if(systems_error != ECSACT_EXEC_SYS_OK) {
+				async_callbacks.add(systems_error);
+				disconnect();
+				return;
+			}
 		}
 	});
 }
@@ -142,33 +130,24 @@ void async_reference::flush_events(
 	const ecsact_execution_events_collector* execution_events,
 	const ecsact_async_events_collector*     async_events
 ) {
-	if(async_events != nullptr) {
-		// Could block here
-		bool breaking_error = async_callbacks.invoke(*async_events);
-
-		if(breaking_error) {
-			// Just disconnect immediately
-			disconnect();
-			return;
-		}
-	}
-
-	if(execution_events != nullptr) {
-		// Could block here
-		exec_callbacks.invoke(*execution_events, *registry_id);
+	async_callbacks.invoke(async_events);
+	if(registry_id) {
+		exec_callbacks.invoke(execution_events, *registry_id);
 	}
 }
 
 ecsact_async_request_id async_reference::create_entity_request() {
 	// NOTE: Add entity to both registries
 	// Consider ensure entity
 	auto req_id = next_request_id();
-	if(is_connected == false) {
+	if(is_connected == false && is_connected_notified == false) {
 		types::async_error async_err{
 			.error = ECSACT_ASYNC_ERR_PERMISSION_DENIED,
-			.request_id = req_id};
+			.request_ids = {req_id},
+		};
 
 		async_callbacks.add(async_err);
+		is_connected_notified = true;
 
 		return req_id;
 	}

reference/async_reference/async_reference.hh

@@ -62,6 +62,7 @@ private:
 	std::thread      execution_thread;
 	std::mutex       pending_m;
 	std::atomic_bool is_connected = false;
+	std::atomic_bool is_connected_notified = false;
 
 	ecsact_async_request_id next_request_id();
 	ecsact_async_request_id convert_request_id(int32_t id);

reference/async_reference/callbacks/async_callbacks.cc

@@ -5,9 +5,13 @@ void async_callbacks::add(const types::async_requests type) {
 	requests.insert(requests.end(), type);
 }
 
-bool async_callbacks::invoke(const ecsact_async_events_collector& async_events
+void async_callbacks::invoke(const ecsact_async_events_collector* async_events
 ) {
-	bool breaking_error = false;
+	if(async_events == nullptr) {
+		std::unique_lock lk(async_m);
+		requests.clear();
+		return;
+	}
 
 	std::vector<types::async_requests> pending_requests;
 
@@ -18,33 +22,31 @@ bool async_callbacks::invoke(const ecsact_async_events_collector& async_events
 
 	for(auto& request : pending_requests) {
 		std::visit(
-			[request, &async_events, &breaking_error](auto&& error) {
+			[request, &async_events](auto&& error) {
 				using T = std::decay_t<decltype(error)>;
 				if constexpr(std::is_same_v<T, types::async_error>) {
-					async_events.async_error_callback(
+					async_events->async_error_callback(
 						error.error,
-						static_cast<ecsact_async_request_id>(error.request_id),
-						async_events.async_error_callback_user_data
+						error.request_ids.size(),
+						error.request_ids.data(),
+						async_events->async_error_callback_user_data
 					);
-					breaking_error = true;
 				} else if constexpr(std::is_same_v<T, ecsact_execute_systems_error>) {
-					async_events.system_error_callback(
+					async_events->system_error_callback(
 						error,
-						async_events.system_error_callback_user_data
+						async_events->system_error_callback_user_data
 					);
-					breaking_error = true;
 				} else if constexpr(std::is_same_v<T, types::entity>) {
-					async_events.async_entity_callback(
+					async_events->async_entity_callback(
 						*error.entity_id,
 						error.request_id,
-						async_events.async_entity_error_callback_user_data
+						async_events->async_entity_error_callback_user_data
 					);
 				}
 			},
 			request
 		);
 	}
-	return breaking_error;
 }
 
 void async_callbacks::add_many(const std::vector<types::async_requests>& types

reference/async_reference/callbacks/async_callbacks.hh

@@ -9,7 +9,7 @@
 
 class async_callbacks {
 public:
-	bool invoke(const ecsact_async_events_collector& async_events);
+	void invoke(const ecsact_async_events_collector* async_events);
 	void add(const types::async_requests type);
 	void add_many(const std::vector<types::async_requests>& types);
 

reference/async_reference/callbacks/execution_callbacks.cc

@@ -9,9 +9,17 @@ std::vector<types::callback_info> execution_callbacks::remove_callbacks_info =
 std::mutex execution_callbacks::execution_m = {};
 
 void execution_callbacks::invoke(
-	const ecsact_execution_events_collector& execution_events,
+	const ecsact_execution_events_collector* execution_events,
 	ecsact_registry_id                       registry_id
 ) {
+	if(execution_events == nullptr) {
+		std::unique_lock lk(execution_m);
+		init_callbacks_info.clear();
+		update_callbacks_info.clear();
+		remove_callbacks_info.clear();
+		return;
+	}
+
 	std::vector<types::callback_info> init_callbacks;
 	std::vector<types::callback_info> update_callbacks;
 	std::vector<types::callback_info> remove_callbacks;
@@ -35,12 +43,12 @@ void execution_callbacks::invoke(
 			component_info.component_id
 		);
 
-		execution_events.init_callback(
+		execution_events->init_callback(
 			component_info.event,
 			component_info.entity_id,
 			component_info.component_id,
 			component_data,
-			execution_events.init_callback_user_data
+			execution_events->init_callback_user_data
 		);
 	}
 
@@ -51,12 +59,12 @@ void execution_callbacks::invoke(
 			component_info.component_id
 		);
 
-		execution_events.update_callback(
+		execution_events->update_callback(
 			component_info.event,
 			component_info.entity_id,
 			component_info.component_id,
 			component_data,
-			execution_events.update_callback_user_data
+			execution_events->update_callback_user_data
 		);
 	}
 
@@ -67,12 +75,12 @@ void execution_callbacks::invoke(
 			component_info.component_id
 		);
 
-		execution_events.remove_callback(
+		execution_events->remove_callback(
 			component_info.event,
 			component_info.entity_id,
 			component_info.component_id,
 			component_data,
-			execution_events.remove_callback_user_data
+			execution_events->remove_callback_user_data
 		);
 	}
 }

reference/async_reference/callbacks/execution_callbacks.hh

@@ -9,7 +9,7 @@
 class execution_callbacks {
 public:
 	void invoke(
-		const ecsact_execution_events_collector& execution_events,
+		const ecsact_execution_events_collector* execution_events,
 		ecsact_registry_id                       registry_id
 	);