Network: Batch individual particle packets (#16458)

also bumps proto ver

builtin/game/misc_s.lua

@@ -131,6 +131,7 @@ core.protocol_versions = {
 	["5.11.0"] = 47,
 	["5.12.0"] = 48,
 	["5.13.0"] = 49,
+	["5.14.0"] = 50,
 }
 
 setmetatable(core.protocol_versions, {__newindex = function()

src/client/client.h

@@ -194,6 +194,7 @@ public:
 	void handleCommand_DetachedInventory(NetworkPacket* pkt);
 	void handleCommand_ShowFormSpec(NetworkPacket* pkt);
 	void handleCommand_SpawnParticle(NetworkPacket* pkt);
+	void handleCommand_SpawnParticleBatch(NetworkPacket *pkt);
 	void handleCommand_AddParticleSpawner(NetworkPacket* pkt);
 	void handleCommand_DeleteParticleSpawner(NetworkPacket* pkt);
 	void handleCommand_HudAdd(NetworkPacket* pkt);

src/network/clientopcodes.cpp

@@ -111,6 +111,7 @@ const ToClientCommandHandler toClientCommandTable[TOCLIENT_NUM_MSG_TYPES] =
 	{ "TOCLIENT_FORMSPEC_PREPEND",         TOCLIENT_STATE_CONNECTED, &Client::handleCommand_FormspecPrepend }, // 0x61,
 	{ "TOCLIENT_MINIMAP_MODES",            TOCLIENT_STATE_CONNECTED, &Client::handleCommand_MinimapModes }, // 0x62,
 	{ "TOCLIENT_SET_LIGHTING",             TOCLIENT_STATE_CONNECTED, &Client::handleCommand_SetLighting }, // 0x63,
+	{ "TOCLIENT_SPAWN_PARTICLE_BATCH",     TOCLIENT_STATE_CONNECTED, &Client::handleCommand_SpawnParticleBatch }, // 0x64,
 };
 
 const static ServerCommandFactory null_command_factory = { nullptr, 0, false };

src/network/clientpackethandler.cpp

@@ -34,6 +34,7 @@
 #include "skyparams.h"
 #include "particles.h"
 #include <memory>
+#include <sstream>
 
 const char *accessDeniedStrings[SERVER_ACCESSDENIED_MAX] = {
 	N_("Invalid password"),
@@ -979,6 +980,29 @@ void Client::handleCommand_SpawnParticle(NetworkPacket* pkt)
 	m_client_event_queue.push(event);
 }
 
+void Client::handleCommand_SpawnParticleBatch(NetworkPacket *pkt)
+{
+	std::stringstream particle_batch_data(std::ios::binary | std::ios::in | std::ios::out);
+	{
+		std::istringstream compressed(pkt->readLongString(), std::ios::binary);
+		decompressZstd(compressed, particle_batch_data);
+	}
+
+	while (particle_batch_data.peek() != EOF) {
+		auto p = std::make_unique<ParticleParameters>();
+		{
+			std::istringstream particle_data(deSerializeString32(particle_batch_data), std::ios::binary);
+			p->deSerialize(particle_data, m_proto_ver);
+		}
+
+		ClientEvent *event = new ClientEvent();
+		event->type = CE_SPAWN_PARTICLE;
+		event->spawn_particle = p.release();
+
+		m_client_event_queue.push(event);
+	}
+}
+
 void Client::handleCommand_AddParticleSpawner(NetworkPacket* pkt)
 {
 	std::string datastring(pkt->getString(0), pkt->getSize());

src/network/networkprotocol.cpp

@@ -68,10 +68,13 @@
 	PROTOCOL VERSION 49
 		Support of showing a player inventory using 'core.show_formspec'
 		[scheduled bump for 5.13.0]
+	PROTOCOL VERSION 50
+		Support for TOCLIENT_SPAWN_PARTICLE_BATCH
+		[scheduled bump for 5.14.0]
 */
 
 // Note: Also update core.protocol_versions in builtin when bumping
-const u16 LATEST_PROTOCOL_VERSION = 49;
+const u16 LATEST_PROTOCOL_VERSION = 50;
 
 // See also formspec [Version History] in doc/lua_api.md
 const u16 FORMSPEC_API_VERSION = 10;

src/network/networkprotocol.h

@@ -288,6 +288,8 @@ enum ToClientCommand : u16
 
 	TOCLIENT_SPAWN_PARTICLE = 0x46,
 	/*
+		ParticleParameters params:
+
 		using range<T> = RangedParameter<T> {
 			T min, max
 			f32 bias
@@ -692,7 +694,14 @@ enum ToClientCommand : u16
 			f32 center_weight_power
 	*/
 
-	TOCLIENT_NUM_MSG_TYPES = 0x64,
+	TOCLIENT_SPAWN_PARTICLE_BATCH = 0x64,
+	/*
+		std::string data, zstd-compressed, for each particle:
+			u32 len
+			u8[len] serialized ParticleParameters
+	*/
+
+	TOCLIENT_NUM_MSG_TYPES = 0x65,
 };
 
 enum ToServerCommand : u16

src/network/serveropcodes.cpp

@@ -212,4 +212,5 @@ const ClientCommandFactory clientCommandFactoryTable[TOCLIENT_NUM_MSG_TYPES] =
 	{ "TOCLIENT_FORMSPEC_PREPEND",         0, true }, // 0x61
 	{ "TOCLIENT_MINIMAP_MODES",            0, true }, // 0x62
 	{ "TOCLIENT_SET_LIGHTING",             0, true }, // 0x63
+	{ "TOCLIENT_SPAWN_PARTICLE_BATCH",     0, true }, // 0x64
 };

src/server.cpp

@@ -3,9 +3,6 @@
 // Copyright (C) 2010-2013 celeron55, Perttu Ahola <celeron55@gmail.com>
 
 #include "server.h"
-#include <iostream>
-#include <queue>
-#include <algorithm>
 #include "irr_v2d.h"
 #include "network/connection.h"
 #include "network/networkpacket.h"
@@ -65,6 +62,10 @@
 #include "gettext.h"
 #include "util/tracy_wrapper.h"
 
+#include <iostream>
+#include <queue>
+#include <algorithm>
+#include <sstream>
 #include <csignal>
 
 class ClientNotFoundException : public BaseException
@@ -805,6 +806,12 @@ void Server::AsyncRunStep(float dtime, bool initial_step)
 	}
 #endif
 
+	// Send queued particles
+	{
+		EnvAutoLock envlock(this);
+		SendSpawnParticles();
+	}
+
 	/*
 		Check added and deleted active objects
 	*/
@@ -1603,47 +1610,69 @@ void Server::SendShowFormspecMessage(session_t peer_id, const std::string &forms
 	Send(&pkt);
 }
 
-// Spawns a particle on peer with peer_id
+void Server::SendSpawnParticles(RemotePlayer *player,
-void Server::SendSpawnParticle(session_t peer_id, u16 protocol_version,
+		const std::vector<ParticleParameters> &particles)
-	const ParticleParameters &p)
 {
 	static thread_local const float radius =
 		g_settings->getS16("max_block_send_distance") * MAP_BLOCKSIZE * BS;
-
-	if (peer_id == PEER_ID_INEXISTENT) {
-		std::vector<session_t> clients = m_clients.getClientIDs();
-		const v3f pos = p.pos * BS;
 	const float radius_sq = radius * radius;
 
-		for (const session_t client_id : clients) {
-			RemotePlayer *player = m_env->getPlayer(client_id);
-			if (!player)
-				continue;
-
 	PlayerSAO *sao = player->getPlayerSAO();
 	if (!sao)
-				continue;
-
-			// Do not send to distant clients
-			if (sao->getBasePosition().getDistanceFromSQ(pos) > radius_sq)
-				continue;
-
-			SendSpawnParticle(client_id, player->protocol_version, p);
-		}
 		return;
-	}
-	assert(protocol_version != 0);
 
-	NetworkPacket pkt(TOCLIENT_SPAWN_PARTICLE, 0, peer_id);
+	std::ostringstream particle_batch_data(std::ios_base::binary);
-
+	for (const auto &particle : particles) {
-	{
+		if (sao->getBasePosition().getDistanceFromSQ(particle.pos * BS) > radius_sq)
-		// NetworkPacket and iostreams are incompatible...
+			continue; // out of range
-		std::ostringstream oss(std::ios_base::binary);
-		p.serialize(oss, protocol_version);
-		pkt.putRawString(oss.str());
-	}
 
+		std::ostringstream particle_data(std::ios_base::binary);
+		particle.serialize(particle_data, player->protocol_version);
+		std::string particle_data_str = particle_data.str();
+		SANITY_CHECK(particle_data_str.size() < U32_MAX);
+		if (player->protocol_version < 50) {
+			// Client only supports TOCLIENT_SPAWN_PARTICLE,
+			// so turn the written particle into a packet immediately
+			NetworkPacket pkt(TOCLIENT_SPAWN_PARTICLE, particle_data_str.size(), player->getPeerId());
+			pkt.putRawString(particle_data_str);
 			Send(&pkt);
+		} else {
+			particle_batch_data << serializeString32(particle_data_str);
+		}
+	}
+
+	if (particle_batch_data.tellp() == 0)
+		return; // no batch to send
+
+	// Client supports TOCLIENT_SPAWN_PARTICLE_BATCH
+	assert(player->protocol_version >= 50);
+	std::ostringstream compressed(std::ios_base::binary);
+	compressZstd(particle_batch_data.str(), compressed);
+
+	NetworkPacket pkt(TOCLIENT_SPAWN_PARTICLE_BATCH,
+			4 + compressed.tellp(), player->getPeerId());
+	pkt.putLongString(compressed.str());
+	Send(&pkt);
+}
+
+void Server::SendSpawnParticles()
+{
+	for (const auto &[pname, particles] : m_particles_to_send) {
+		if (pname.empty())
+			continue; // sent to all clients
+
+		RemotePlayer *player = m_env->getPlayer(pname.c_str());
+		if (!player)
+			continue;
+
+		SendSpawnParticles(player, particles);
+	}
+
+	for (auto *player : m_env->getPlayers()) {
+		SendSpawnParticles(player, m_particles_to_send[""]);
+	}
+
+	m_particles_to_send.clear();
 }
 
 void Server::SendAddParticleSpawner(const std::string &to_player,
@@ -3621,17 +3650,7 @@ void Server::spawnParticle(const std::string &playername,
 	if (!m_env)
 		return;
 
-	session_t peer_id = PEER_ID_INEXISTENT;
+	m_particles_to_send[playername].push_back(p);
-	u16 proto_ver = 0;
-	if (!playername.empty()) {
-		RemotePlayer *player = m_env->getPlayer(playername.c_str());
-		if (!player)
-			return;
-		peer_id = player->getPeerId();
-		proto_ver = player->protocol_version;
-	}
-
-	SendSpawnParticle(peer_id, proto_ver, p);
 }
 
 u32 Server::addParticleSpawner(const ParticleSpawnerParameters &p,

src/server.h

@@ -605,9 +605,11 @@ private:
 
 	void SendDeleteParticleSpawner(session_t peer_id, u32 id);
 
-	// Spawns particle on peer with peer_id (PEER_ID_INEXISTENT == all)
+	// Spawn particles for a specific client, batching them if clients support it.
-	void SendSpawnParticle(session_t peer_id, u16 protocol_version,
+	void SendSpawnParticles(RemotePlayer *player,
-		const ParticleParameters &p);
+			const std::vector<ParticleParameters> &particles);
+	// Spawn all particles for this step, batching them if clients support it.
+	void SendSpawnParticles();
 
 	void SendActiveObjectRemoveAdd(RemoteClient *client, PlayerSAO *playersao);
 	void SendActiveObjectMessages(session_t peer_id, const std::string &datas,
@@ -805,6 +807,10 @@ private:
 	MetricCounterPtr m_packet_recv_counter;
 	MetricCounterPtr m_packet_recv_processed_counter;
 	MetricCounterPtr m_map_edit_event_counter;
+
+	// Particles to send this server step
+	// [playername] = list of params, empty playername for broadcast
+	std::unordered_map<std::string, std::vector<ParticleParameters>> m_particles_to_send;
 };
 
 /*