Everything working again, added tick rate options and fixed some bugs

Author: Kelwan

ecsact/runtime/async.h

@@ -53,6 +53,11 @@ typedef enum {
 	 */
 	ECSACT_ASYNC_ERR_PERMISSION_DENIED,
 
+	/**
+	 * Client sent invalid connection options
+	 */
+	ECSACT_ASYNC_INVALID_CONNECTION_STRING,
+
 	/**
 	 * Connection to the client is closed
 	 */
@@ -70,7 +75,8 @@ typedef enum {
  * @param async_err when there is no async error this will be @ref
  * ECSACT_ASYNC_OK otherwise @see ecsact_async_error
 
- * @param request_id the request ID returned by an async request function that
+ * @param request_ids A list of request IDs returned by an async request
+ function that
  * was responsible for this error
  * @param callback_user_data the @ref
  * ecsact_async_events_collector::error_callback_user_data
@@ -120,7 +126,7 @@ typedef struct ecsact_async_events_collector {
 	ecsact_async_error_callback async_error_callback;
 
 	/**
-	 * `callback_user_data` passed to `error_callback`
+	 * `callback_user_data` passed to `async_error_callback`
 	 */
 	void* async_error_callback_user_data;
 
@@ -133,9 +139,9 @@ typedef struct ecsact_async_events_collector {
 	ecsact_async_create_entity_callback async_entity_callback;
 
 	/**
-	 * `callback_user_data` passed to `error_callback`
+	 * `callback_user_data` passed to `async_entity_callback`
 	 */
-	void* async_entity_error_callback_user_data;
+	void* async_entity_callback_user_data;
 
 	/**
 	 * invoked when a system execution error occurred.

reference/async_reference/async_reference.cc

@@ -3,27 +3,94 @@
 #include <ranges>
 #include <algorithm>
 #include <iterator>
+#include <chrono>
 
 #include "async_reference.hh"
 
+struct parsed_connection_string {
+	std::string                        host;
+	std::map<std::string, std::string> options;
+};
+
+static auto str_subrange(std::string_view str, auto start, auto end) {
+	return str.substr(start, end - start);
+}
+
+static auto parse_connection_string(std::string_view str)
+	-> parsed_connection_string {
+	auto result = parsed_connection_string{};
+
+	auto options_start = str.find("?");
+
+	if(options_start == std::string::npos) {
+		// No options
+		result.host = str;
+		return result;
+	}
+
+	result.host = str.substr(0, options_start);
+
+	auto amp_idx = options_start;
+
+	while(amp_idx != std::string::npos) {
+		auto next_amp_idx = str.find("&", amp_idx + 1);
+
+		auto name_str = std::string{};
+		auto value_str = std::string{};
+		auto eq_idx = str.find("=", amp_idx);
+
+		if(eq_idx >= next_amp_idx) {
+			name_str = str_subrange(str, amp_idx + 1, next_amp_idx);
+		} else {
+			name_str = str_subrange(str, amp_idx + 1, eq_idx);
+			value_str = str_subrange(str, eq_idx + 1, next_amp_idx);
+		}
+
+		result.options[name_str] = value_str;
+
+		amp_idx = next_amp_idx;
+	}
+
+	return result;
+}
+
 ecsact_async_request_id async_reference::connect(const char* connection_string
 ) {
+	auto req_id = next_request_id();
+
 	std::string connect_str(connection_string);
 
-	registry_id = ecsact_create_registry("async_reference_impl_reg");
+	auto result = parse_connection_string(connect_str);
 
-	auto req_id = next_request_id();
-	// The good and bad strings simulate the outcome of connections
-	if(connect_str == "good") {
-		is_connected = true;
+	if(result.options.contains("tick_rate")) {
+		auto tick_str = result.options.at("tick_rate");
 
-		execute_systems();
-	} else {
-		// Same thing that happens in enqueue? Callback next flush?
+		int tick_int = std::stoi(tick_str);
+
+		tick_rate = std::chrono::milliseconds(tick_int);
+	}
+
+	// The good and bad strings simulate the outcome of connections
+	if(result.host != "good") {
 		is_connected = false;
 		is_connected_notified = false;
+		return req_id;
 	}
 
+	if(tick_rate.count() == 0) {
+		types::async_error async_err{
+			.error = ECSACT_ASYNC_INVALID_CONNECTION_STRING,
+			.request_ids = {req_id},
+		};
+
+		async_callbacks.add(async_err);
+		return req_id;
+	}
+
+	registry_id = ecsact_create_registry("async_reference_impl_reg");
+	is_connected = true;
+	execute_systems();
+
 	return req_id;
 }
 
@@ -39,7 +106,6 @@ ecsact_async_request_id async_reference::enqueue_execution_options(
 		};
 
 		is_connected_notified = true;
-		// Could block here
 		async_callbacks.add(async_err);
 		return req_id;
 	}
@@ -51,24 +117,46 @@ ecsact_async_request_id async_reference::enqueue_execution_options(
 		.options = cpp_options,
 	};
 
-	// Could block here
 	tick_manager.add_pending_options(pending_options);
 	return req_id;
 }
 
 void async_reference::execute_systems() {
 	execution_thread = std::thread([this] {
+		using namespace std::chrono_literals;
+		using clock = std::chrono::high_resolution_clock;
+		using milliseconds = std::chrono::milliseconds;
+		using nanoseconds = std::chrono::nanoseconds;
+		using std::chrono::duration_cast;
+
+		nanoseconds execution_duration = {};
+		nanoseconds sleep_drift = {};
+
 		while(is_connected == true) {
-			// Could block here
 			auto async_err = tick_manager.validate_pending_options();
 
+			const auto sleep_duration = tick_rate - execution_duration;
+
+			auto wait_start = clock::now();
+			std::this_thread::sleep_for(sleep_duration - sleep_drift);
+			auto wait_end = clock::now();
+
+			sleep_drift =
+				sleep_duration - duration_cast<nanoseconds>(wait_start - wait_end);
+
 			if(async_err.error != ECSACT_ASYNC_OK) {
 				async_callbacks.add(async_err);
 
-				disconnect();
+				is_connected = false;
+				break;
 			}
 
-			auto cpp_options = tick_manager.get_options_now();
+			auto start = clock::now();
+
+			auto cpp_options = tick_manager.move_and_increment_tick();
+
+			// TODO(Kelwan): Add done callbacks so we can resolve all requests
+			// https://github.com/ecsact-dev/ecsact_runtime/issues/102
 
 			ecsact_execution_events_collector collector;
 			collector.init_callback = &execution_callbacks::init_callback;
@@ -87,31 +175,17 @@ void async_reference::execute_systems() {
 				);
 			}
 
-			std::vector<ecsact_async_request_id> pending_entities;
-
-			// Could block here
-			std::unique_lock lk(pending_m);
-			pending_entities = std::move(pending_entity_requests);
-			pending_entity_requests.clear();
-			lk.unlock();
-
-			for(auto& entity_request_id : pending_entities) {
-				auto entity = ecsact_create_entity(*registry_id);
-
-				types::entity created_entity{
-					.entity_id = entity,
-					.request_id = entity_request_id,
-				};
-				// Could block here
-				async_callbacks.add(created_entity);
-			}
+			process_entities();
 
 			auto systems_error =
 				ecsact_execute_systems(*registry_id, 1, options.get(), &collector);
 
+			auto end = clock::now();
+			execution_duration = duration_cast<nanoseconds>(end - start);
+
 			if(systems_error != ECSACT_EXEC_SYS_OK) {
 				async_callbacks.add(systems_error);
-				disconnect();
+				is_connected = false;
 				return;
 			}
 		}
@@ -129,8 +203,6 @@ void async_reference::flush_events(
 }
 
 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 && is_connected_notified == false) {
 		types::async_error async_err{
@@ -165,3 +237,23 @@ ecsact_async_request_id async_reference::next_request_id() {
 ecsact_async_request_id async_reference::convert_request_id(int32_t id) {
 	return static_cast<ecsact_async_request_id>(id);
 }
+
+void async_reference::process_entities() {
+	std::vector<ecsact_async_request_id> pending_entities;
+
+	std::unique_lock lk(pending_m);
+	pending_entities = std::move(pending_entity_requests);
+	pending_entity_requests.clear();
+	lk.unlock();
+
+	for(auto& entity_request_id : pending_entities) {
+		auto entity = ecsact_create_entity(*registry_id);
+
+		types::entity created_entity{
+			.entity_id = entity,
+			.request_id = entity_request_id,
+		};
+
+		async_callbacks.add(created_entity);
+	}
+}

reference/async_reference/async_reference.hh

@@ -42,6 +42,8 @@ private:
 
 	std::optional<ecsact_registry_id> registry_id;
 
+	void process_entities();
+
 	tick_manager        tick_manager;
 	execution_callbacks exec_callbacks;
 	async_callbacks     async_callbacks;
@@ -53,6 +55,8 @@ private:
 	std::atomic_bool is_connected = false;
 	std::atomic_bool is_connected_notified = false;
 
+	std::chrono::milliseconds tick_rate = {};
+
 	ecsact_async_request_id next_request_id();
 	ecsact_async_request_id convert_request_id(int32_t id);
 };

reference/async_reference/callbacks/async_callbacks.cc

@@ -2,7 +2,7 @@
 
 void async_callbacks::add(const types::async_requests type) {
 	std::unique_lock lk(async_m);
-	requests.insert(requests.end(), type);
+	requests.push_back(type);
 }
 
 void async_callbacks::invoke(const ecsact_async_events_collector* async_events
@@ -22,25 +22,34 @@ void async_callbacks::invoke(const ecsact_async_events_collector* async_events
 
 	for(auto& request : pending_requests) {
 		std::visit(
-			[request, &async_events](auto&& error) {
+			[&async_events](auto&& error) {
 				using T = std::decay_t<decltype(error)>;
 				if constexpr(std::is_same_v<T, types::async_error>) {
+					if(async_events->async_error_callback == nullptr) {
+						return;
+					}
 					async_events->async_error_callback(
 						error.error,
 						error.request_ids.size(),
 						error.request_ids.data(),
 						async_events->async_error_callback_user_data
 					);
 				} else if constexpr(std::is_same_v<T, ecsact_execute_systems_error>) {
+					if(async_events->system_error_callback == nullptr) {
+						return;
+					}
 					async_events->system_error_callback(
 						error,
 						async_events->system_error_callback_user_data
 					);
 				} else if constexpr(std::is_same_v<T, types::entity>) {
+					if(async_events->async_entity_callback == nullptr) {
+						return;
+					}
 					async_events->async_entity_callback(
 						*error.entity_id,
 						error.request_id,
-						async_events->async_entity_error_callback_user_data
+						async_events->async_entity_callback_user_data
 					);
 				}
 			},