WIP Controller refactor.

Author: AndreaCatania

core/network_interface.h

@@ -160,6 +160,7 @@ class NetworkInterface {
 template <typename... ARGs>
 void RpcHandle<ARGs...>::rpc(NetworkInterface &p_interface, int p_peer_id, ARGs... p_args) const {
 	ENSURE(p_interface.rpcs_info.size() > index);
+	ASSERT_COND_MSG(p_interface.fetch_local_peer_id() != p_peer_id, "Sending an rpc to self is not allowed.");
 
 	DataBuffer db;
 	db.begin_write(0);

core/object_data.h

@@ -69,10 +69,10 @@ struct ObjectData {
 	/// The sync variables of this node. The order of this vector matters
 	/// because the index is the `VarId`.
 	std::vector<VarDescriptor> vars;
-	NS::Processor<float> functions[PROCESSPHASE_COUNT];
+	NS::Processor<double> functions[PROCESSPHASE_COUNT];
 
 	std::function<void(DataBuffer & /*out_buffer*/, float /*update_rate*/)> func_trickled_collect;
-	std::function<void(float /*delta*/, float /*interpolation_alpha*/, DataBuffer & /*past_buffer*/, DataBuffer & /*future_buffer*/)> func_trickled_apply;
+	std::function<void(double /*delta*/, float /*interpolation_alpha*/, DataBuffer & /*past_buffer*/, DataBuffer & /*future_buffer*/)> func_trickled_apply;
 
 public:
 	void set_net_id(ObjectNetId p_id);

net_utilities.cpp

@@ -15,7 +15,7 @@ void NS::PeerData::set_latency(int p_latency) {
 }
 
 int NS::PeerData::get_latency() const {
-	return compressed_latency * 4.0;
+	return int(compressed_latency) * 4;
 }
 
 void NS::PeerData::make_controller(NS::SceneSynchronizerBase &p_scene_sync, int p_peer) {
@@ -63,6 +63,9 @@ void NS::SyncGroup::mark_changes_as_notified() {
 
 void NS::SyncGroup::add_listening_peer(int p_peer) {
 	NS::VecFunc::insert_unique(listening_peers, p_peer);
+	if (NS::VecFunc::insert_unique(networked_peers, p_peer)) {
+		NS::VecFunc::insert_unique(peers_with_newly_calculated_latency, p_peer);
+	}
 
 	// Make all the controlled objects as simulated.
 	const std::vector<ObjectData *> *controlled_objects_ptr = scene_sync->get_peer_controlled_objects_data(p_peer);
@@ -184,6 +187,7 @@ void NS::SyncGroup::remove_sync_object(std::size_t p_index, bool p_is_simulated)
 		// If no other simulated objects controlled by `associated_peer` remove it from
 		bool is_simulating = false;
 		bool is_networking = false;
+
 		for (auto &soi : simulated_sync_objects) {
 			if (soi.od->get_controlled_by_peer() == associted_peer) {
 				is_networking = true;
@@ -192,6 +196,12 @@ void NS::SyncGroup::remove_sync_object(std::size_t p_index, bool p_is_simulated)
 			}
 		}
 
+		if (!is_networking) {
+			if (NS::VecFunc::has(listening_peers, associted_peer)) {
+				is_networking = true;
+			}
+		}
+
 		if (!is_networking) {
 			for (auto &toi : trickled_sync_objects) {
 				if (toi.od->get_controlled_by_peer() == associted_peer) {

networked_controller.cpp

@@ -314,7 +314,7 @@ void NetworkedControllerBase::on_state_validated(FrameIndex p_frame_index) {
 }
 
 void NetworkedControllerBase::on_rewind_frame_begin(FrameIndex p_input_id, int p_index, int p_count) {
-	if (controller && is_realtime_enabled()) {
+	if (controller && can_simulate()) {
 		controller->queue_instant_process(p_input_id, p_index, p_count);
 	}
 }
@@ -374,13 +374,15 @@ bool NetworkedControllerBase::player_has_new_input() const {
 	return has_player_new_input;
 }
 
-bool NetworkedControllerBase::is_realtime_enabled() {
-	if (is_server_controller()) {
-		return true;
-	} else if (scene_synchronizer) {
-		// TODO optimize by avoiding fetching the controlled objects in this way?
-		const std::vector<ObjectData *> *controlled_objects = scene_synchronizer->get_peer_controlled_objects_data(get_authority_peer());
-		if (controlled_objects) {
+bool NetworkedControllerBase::can_simulate() {
+	NS_PROFILE
+
+	const std::vector<ObjectData *> *controlled_objects = scene_synchronizer ? scene_synchronizer->get_peer_controlled_objects_data(get_authority_peer()) : nullptr;
+	if (controlled_objects) {
+		if (is_server_controller() || is_player_controller()) {
+			return controlled_objects->size() > 0;
+		} else {
+			// TODO optimize by avoiding fetching the controlled objects in this way?
 			for (const ObjectData *od : *controlled_objects) {
 				if (od->realtime_sync_enabled_on_client) {
 					return true;

networked_controller.h

@@ -37,7 +37,7 @@ struct NoNetController;
 // instantiated.
 // The most important part is inside the `PlayerController`, `ServerController`,
 // `DollController`, `NoNetController`.
-class NetworkedControllerBase {
+class NetworkedControllerBase final {
 	friend class NS::SceneSynchronizerBase;
 	template <class NetInterfaceClass>
 	friend class NetworkedController;
@@ -214,7 +214,7 @@ class NetworkedControllerBase {
 public:
 	bool player_has_new_input() const;
 
-	bool is_realtime_enabled();
+	bool can_simulate();
 
 protected:
 	void notify_controller_reset();

scene_synchronizer.cpp

@@ -132,6 +132,9 @@ void SceneSynchronizerBase::setup(SynchronizerManager &p_synchronizer_interface)
 	// Make sure to reset all the assigned controllers.
 	reset_controllers();
 
+	// Spawn the self peer.
+	on_peer_connected(get_network_interface().fetch_local_peer_id());
+
 	// Init the peers already connected.
 	std::vector<int> peer_ids;
 	network_interface->fetch_connected_peers(peer_ids);
@@ -624,7 +627,7 @@ void SceneSynchronizerBase::untrack_variable_changes(ListenerHandle p_handle) {
 	delete listener;
 }
 
-NS::PHandler SceneSynchronizerBase::register_process(ObjectLocalId p_id, ProcessPhase p_phase, std::function<void(float)> p_func) {
+NS::PHandler SceneSynchronizerBase::register_process(ObjectLocalId p_id, ProcessPhase p_phase, std::function<void(double)> p_func) {
 	ERR_FAIL_COND_V(p_id == NS::ObjectLocalId::NONE, NS::NullPHandler);
 	ERR_FAIL_COND_V(!p_func, NS::NullPHandler);
 
@@ -651,7 +654,7 @@ void SceneSynchronizerBase::unregister_process(ObjectLocalId p_id, ProcessPhase
 void SceneSynchronizerBase::set_trickled_sync(
 		ObjectLocalId p_id,
 		std::function<void(DataBuffer & /*out_buffer*/, float /*update_rate*/)> p_func_trickled_collect,
-		std::function<void(float /*delta*/, float /*interpolation_alpha*/, DataBuffer & /*past_buffer*/, DataBuffer & /*future_buffer*/)> p_func_trickled_apply) {
+		std::function<void(double /*delta*/, float /*interpolation_alpha*/, DataBuffer & /*past_buffer*/, DataBuffer & /*future_buffer*/)> p_func_trickled_apply) {
 	ERR_FAIL_COND(p_id == ObjectLocalId::NONE);
 
 	NS::ObjectData *od = get_object_data(p_id);
@@ -884,7 +887,13 @@ bool SceneSynchronizerBase::is_peer_networking_enabled(int p_peer) const {
 void SceneSynchronizerBase::on_peer_connected(int p_peer) {
 	PeerData npd;
 	auto pd_it = MapFunc::insert_if_new(peer_data, p_peer, std::move(npd));
+	if (pd_it->second.get_controller()) {
+		// Nothing to do, already initialized.
+		return;
+	}
+
 	pd_it->second.make_controller(*this, p_peer);
+	reset_controller(*pd_it->second.get_controller());
 
 	event_peer_status_updated.broadcast(p_peer, true, true);
 
@@ -1372,7 +1381,7 @@ void SceneSynchronizerBase::process_functions__execute() {
 		cached_process_functions_valid = true;
 	}
 
-	SceneSynchronizerDebugger::singleton()->debug_print(network_interface, "Process functions START", true);
+	SceneSynchronizerDebugger::singleton()->print(INFO, "Process functions START");
 
 	// Pre process phase
 	for (int process_phase = PROCESSPHASE_EARLY; process_phase < PROCESSPHASE_PROCESS; ++process_phase) {
@@ -1386,9 +1395,9 @@ void SceneSynchronizerBase::process_functions__execute() {
 		const std::string info = "process phase -- CONTROLLER --";
 		NS_PROFILE_WITH_INFO(info);
 
-		// TODO optimize this, as `is_realtime_enabled` is expensive.
+		// TODO optimize this, as `can_simulate` is expensive.
 		for (auto &pd : peer_data) {
-			if (pd.second.get_controller() && pd.second.get_controller()->is_realtime_enabled()) {
+			if (pd.second.get_controller() && pd.second.get_controller()->can_simulate()) {
 				pd.second.get_controller()->process(get_fixed_frame_delta());
 			}
 		}
@@ -1935,6 +1944,11 @@ void ServerSynchronizer::process_snapshot_notificator() {
 		delta_snapshot.begin_write(MD_SIZE);
 
 		for (int peer_id : group.get_listening_peers()) {
+			if (peer_id == scene_synchronizer->get_network_interface().fetch_local_peer_id()) {
+				// Never send the snapshot to self (notice `self` is the server).
+				continue;
+			}
+
 			NS::PeerData *peer = MapFunc::get_or_null(scene_synchronizer->peer_data, peer_id);
 			if (peer == nullptr) {
 				SceneSynchronizerDebugger::singleton()->print(ERROR, "The `process_snapshot_notificator` failed to lookup the peer_id `" + std::to_string(peer_id) + "`. Was it removed but never cleared from sync_groups. Report this error, as this is a bug.");
@@ -2243,6 +2257,11 @@ void ServerSynchronizer::process_latency_update() {
 	const std::chrono::high_resolution_clock::time_point now = std::chrono::high_resolution_clock::now();
 
 	for (auto &[peer, peer_data] : scene_synchronizer->get_peers()) {
+		if (peer == scene_synchronizer->get_network_interface().fetch_local_peer_id()) {
+			// No need to update the ping for `self` (the server).
+			continue;
+		}
+
 		std::map<int, PeerServerData>::iterator peer_server_data_it = NS::MapFunc::insert_if_new(peers_data, peer, PeerServerData());
 		if (peer_server_data_it->second.latency_calculation_in_progress) {
 			continue;
@@ -2291,7 +2310,7 @@ void ServerSynchronizer::process_adjust_clients_controller_tick_rate(double p_de
 }
 
 void ServerSynchronizer::process_adjust_client_controller_tick_rate(double p_delta, int p_controller_peer, NetworkedControllerBase &p_controller) {
-	CRASH_COND(!p_controller.is_server_controller());
+	ASSERT_COND(p_controller.is_server_controller());
 
 	if (!p_controller.get_server_controller_unchecked()->streaming_paused) {
 		return;
@@ -2368,7 +2387,7 @@ void ClientSynchronizer::process(double p_delta) {
 	process_trickled_sync(p_delta);
 
 #if DEBUG_ENABLED
-	if (player_controller) {
+	if (player_controller && player_controller->can_simulate()) {
 		const int client_peer = scene_synchronizer->network_interface->fetch_local_peer_id();
 		SceneSynchronizerDebugger::singleton()->write_dump(client_peer, player_controller->get_current_frame_index().id);
 		SceneSynchronizerDebugger::singleton()->start_new_frame();
@@ -2447,7 +2466,7 @@ void ClientSynchronizer::signal_end_sync_changed_variables_events() {
 }
 
 void ClientSynchronizer::on_controller_reset(NetworkedControllerBase &p_controller) {
-	if (player_controller->is_player_controller()) {
+	if (p_controller.is_player_controller()) {
 		// This can't trigger because the reset function creates the player
 		// controller when the following condition is true.
 		ASSERT_COND(p_controller.get_authority_peer() == scene_synchronizer->get_network_interface().fetch_local_peer_id());
@@ -2461,7 +2480,7 @@ void ClientSynchronizer::on_controller_reset(NetworkedControllerBase &p_controll
 }
 
 const std::vector<ObjectData *> &ClientSynchronizer::get_active_objects() const {
-	if make_likely (player_controller && enabled) {
+	if make_likely (player_controller && player_controller->can_simulate() && enabled) {
 		return active_objects;
 	} else {
 		// Since there is no player controller or the sync is disabled, this
@@ -2901,7 +2920,7 @@ int ClientSynchronizer::calculates_sub_ticks(const double p_delta) {
 void ClientSynchronizer::process_simulation(double p_delta) {
 	NS_PROFILE
 
-	if make_unlikely (player_controller == nullptr || enabled == false) {
+	if make_unlikely (player_controller == nullptr || enabled == false || !player_controller->can_simulate()) {
 		// No player controller so can't process the simulation.
 		// TODO Remove this constraint?
 
@@ -3434,7 +3453,7 @@ bool ClientSynchronizer::parse_snapshot(DataBuffer &p_snapshot) {
 		return false;
 	}
 
-	if make_unlikely (received_snapshot.input_id == FrameIndex::NONE && player_controller != nullptr) {
+	if make_unlikely (received_snapshot.input_id == FrameIndex::NONE && player_controller && player_controller->can_simulate()) {
 		// We espect that the player_controller is updated by this new snapshot,
 		// so make sure it's done so.
 		SceneSynchronizerDebugger::singleton()->print(ERROR, "The player controller (" + std::to_string(player_controller->get_authority_peer()) + ") was not part of the received snapshot, this happens when the server destroys the peer controller.");

scene_synchronizer.h

@@ -211,7 +211,7 @@ class SceneSynchronizerBase {
 	std::vector<ChangesListener *> changes_listeners;
 
 	bool cached_process_functions_valid = false;
-	Processor<float> cached_process_functions[PROCESSPHASE_COUNT];
+	Processor<double> cached_process_functions[PROCESSPHASE_COUNT];
 
 	// Set at runtime by the constructor by reading the project settings.
 	bool debug_rewindings_enabled = false;
@@ -374,7 +374,7 @@ class SceneSynchronizerBase {
 	void untrack_variable_changes(ListenerHandle p_handle);
 
 	/// You can use the macro `callable_mp()` to register custom C++ function.
-	NS::PHandler register_process(ObjectLocalId p_id, ProcessPhase p_phase, std::function<void(float)> p_func);
+	NS::PHandler register_process(ObjectLocalId p_id, ProcessPhase p_phase, std::function<void(double)> p_func);
 	void unregister_process(ObjectLocalId p_id, ProcessPhase p_phase, NS::PHandler p_func_handler);
 
 	/// Setup the trickled sync method for this specific object.
@@ -383,7 +383,7 @@ class SceneSynchronizerBase {
 	void set_trickled_sync(
 			ObjectLocalId p_id,
 			std::function<void(DataBuffer & /*out_buffer*/, float /*update_rate*/)> p_func_trickled_collect,
-			std::function<void(float /*delta*/, float /*interpolation_alpha*/, DataBuffer & /*past_buffer*/, DataBuffer & /*future_buffer*/)> p_func_trickled_apply);
+			std::function<void(double /*delta*/, float /*interpolation_alpha*/, DataBuffer & /*past_buffer*/, DataBuffer & /*future_buffer*/)> p_func_trickled_apply);
 
 	/// Returns the latency (RTT in ms) for this peer or -1 if the latency is not available.
 	int get_peer_latency(int p_peer) const;

tests/local_network.cpp

@@ -3,6 +3,7 @@
 
 #include "core/error/error_macros.h"
 #include "core/math/vector3.h"
+#include "modules/network_synchronizer/core/ensure.h"
 #include "modules/network_synchronizer/core/network_interface.h"
 #include "modules/network_synchronizer/core/processor.h"
 #include "modules/network_synchronizer/core/var_data.h"
@@ -39,11 +40,25 @@ void LocalNetwork::start_as_client(LocalNetwork &p_server_network) {
 	this_peer = peer;
 	p_server_network.peer_counter += 1;
 
+	// Insert this peer into the server connected peer.
 	p_server_network.connected_peers[peer] = this;
+
+	// Put the server into the list of connected peers.
 	connected_peers[1] = &p_server_network;
 
+	// Emit the connected event
 	p_server_network.connected_event.broadcast(peer);
 	connected_event.broadcast(1);
+
+	// Mark all the other peers as connected too.
+	for (auto [other_peer, other_local_network] : p_server_network.connected_peers) {
+		if (peer != other_peer) {
+			connected_peers[other_peer] = other_local_network;
+			other_local_network->connected_peers[peer] = this;
+			other_local_network->connected_event.broadcast(peer);
+			connected_event.broadcast(other_peer);
+		}
+	}
 }
 
 void LocalNetwork::register_object(LocalNetworkInterface &p_interface) {
@@ -99,14 +114,16 @@ void LocalNetwork::process(float p_delta) {
 }
 
 void LocalNetwork::rpc_send_internal(const std::shared_ptr<PendingPacket> &p_packet) {
+	ASSERT_COND_MSG(p_packet->peer_recipient != get_peer(), "During the integration test was generated an RPC to self. This is a bug.");
+
 	auto recipient = connected_peers.find(p_packet->peer_recipient);
-	CRASH_COND(recipient == connected_peers.end());
+	ASSERT_COND(recipient != connected_peers.end());
 
 	auto object_map_it = registered_objects.find(p_packet->object_name);
-	CRASH_COND(object_map_it == registered_objects.end());
+	ASSERT_COND(object_map_it != registered_objects.end());
 
 	LocalNetworkInterface *object_net_interface = object_map_it->second;
-	CRASH_COND(object_net_interface == nullptr);
+	ASSERT_COND(object_net_interface);
 
 	recipient->second->rpc_receive_internal(this_peer, p_packet);
 }
@@ -155,7 +172,7 @@ void LocalNetworkInterface::fetch_connected_peers(std::vector<int> &p_connected_
 	p_connected_peers.clear();
 	// Get all the connected peers.
 	for (const auto &[peer_id, _] : network->get_connected_peers()) {
-		if (peer_id != get_unit_authority()) {
+		if (peer_id != fetch_local_peer_id()) {
 			p_connected_peers.push_back(peer_id);
 		}
 	}
@@ -288,14 +305,22 @@ void NS_Test::test_local_network() {
 	CRASH_COND(peer_2_connection_event[0] != server.get_peer());
 
 	// Check the connected peers list is valid
-	std::vector<int> connected_peers;
-	server_obj_1.fetch_connected_peers(connected_peers);
-	CRASH_COND(std::find(connected_peers.begin(), connected_peers.end(), peer_1.get_peer()) == connected_peers.end());
-	CRASH_COND(std::find(connected_peers.begin(), connected_peers.end(), peer_2.get_peer()) == connected_peers.end());
-	peer_1_obj_1.fetch_connected_peers(connected_peers);
-	CRASH_COND(std::find(connected_peers.begin(), connected_peers.end(), server.get_peer()) == connected_peers.end());
-	peer_2_obj_1.fetch_connected_peers(connected_peers);
-	CRASH_COND(std::find(connected_peers.begin(), connected_peers.end(), server.get_peer()) == connected_peers.end());
+	{
+		std::vector<int> connected_peers;
+		server_obj_1.fetch_connected_peers(connected_peers);
+		ASSERT_COND(NS::VecFunc::has(connected_peers, peer_1.get_peer()));
+		ASSERT_COND(NS::VecFunc::has(connected_peers, peer_2.get_peer()));
+
+		connected_peers.clear();
+		peer_1_obj_1.fetch_connected_peers(connected_peers);
+		ASSERT_COND(NS::VecFunc::has(connected_peers, server.get_peer()));
+		ASSERT_COND(NS::VecFunc::has(connected_peers, peer_2.get_peer()));
+
+		connected_peers.clear();
+		peer_2_obj_1.fetch_connected_peers(connected_peers);
+		ASSERT_COND(NS::VecFunc::has(connected_peers, server.get_peer()));
+		ASSERT_COND(NS::VecFunc::has(connected_peers, peer_1.get_peer()));
+	}
 
 	Vector<uint8_t> vec;
 	vec.push_back(1);

tests/local_scene.cpp

@@ -165,7 +165,7 @@ void LocalScene::remove_object(const char *p_object_name) {
 	}
 }
 
-void LocalScene::process(float p_delta) {
+void LocalScene::process(double p_delta) {
 	scene_sync->process(p_delta);
 	// Clear any pending RPC.
 	// NOTE: The network process is executed after the scene_sync so any pending

tests/local_scene.h

@@ -81,7 +81,7 @@ class LocalScene {
 
 	void remove_object(const char *p_object_name);
 
-	void process(float p_delta);
+	void process(double p_delta);
 };
 
 template <class T>