Use multiple threads for mesh generation (#13062)

Co-authored-by: sfan5 <sfan5@live.de>

8 files changed, 179 insertions, 53 deletions

diff --git a/builtin/settingtypes.txt b/builtin/settingtypes.txt
index b3398c831..a97d2c0f7 100644
--- a/builtin/settingtypes.txt
+++ b/builtin/settingtypes.txt
@@ -1673,6 +1673,11 @@ enable_mesh_cache (Mesh cache) bool false
 #    down the rate of mesh updates, thus reducing jitter on slower clients.
 mesh_generation_interval (Mapblock mesh generation delay) int 0 0 50
 
+#    Number of threads to use for mesh generation.
+#    Value of 0 (default) will let Minetest autodetect the number of available threads.
+mesh_generation_threads (Mapblock mesh generation threads) int 0 0 8
+
+
 #    Size of the MapBlock cache of the mesh generator. Increasing this will
 #    increase the cache hit %, reducing the data being copied from the main
 #    thread, thus reducing jitter.
diff --git a/src/client/client.cpp b/src/client/client.cpp
index 04686c43c..27b0ca852 100644
--- a/src/client/client.cpp
+++ b/src/client/client.cpp
@@ -111,7 +111,7 @@ Client::Client(
 	m_sound(sound),
 	m_event(event),
 	m_rendering_engine(rendering_engine),
-	m_mesh_update_thread(this),
+	m_mesh_update_manager(this),
 	m_env(
 		new ClientMap(this, rendering_engine, control, 666),
 		tsrc, this
@@ -312,7 +312,7 @@ void Client::Stop()
 	if (m_mods_loaded)
 		m_script->on_shutdown();
 	//request all client managed threads to stop
-	m_mesh_update_thread.stop();
+	m_mesh_update_manager.stop();
 	// Save local server map
 	if (m_localdb) {
 		infostream << "Local map saving ended." << std::endl;
@@ -325,7 +325,7 @@ void Client::Stop()
 
 bool Client::isShutdown()
 {
-	return m_shutdown || !m_mesh_update_thread.isRunning();
+	return m_shutdown || !m_mesh_update_manager.isRunning();
 }
 
 Client::~Client()
@@ -335,13 +335,12 @@ Client::~Client()
 
 	deleteAuthData();
 
-	m_mesh_update_thread.stop();
-	m_mesh_update_thread.wait();
-	while (!m_mesh_update_thread.m_queue_out.empty()) {
-		MeshUpdateResult r = m_mesh_update_thread.m_queue_out.pop_frontNoEx();
+	m_mesh_update_manager.stop();
+	m_mesh_update_manager.wait();
+	
+	MeshUpdateResult r;
+	while (m_mesh_update_manager.getNextResult(r))
 		delete r.mesh;
-	}
-
 
 	delete m_inventory_from_server;
 
@@ -547,14 +546,14 @@ void Client::step(float dtime)
 		int num_processed_meshes = 0;
 		std::vector<v3s16> blocks_to_ack;
 		bool force_update_shadows = false;
-		while (!m_mesh_update_thread.m_queue_out.empty())
+		MeshUpdateResult r;
+		while (m_mesh_update_manager.getNextResult(r))
 		{
 			num_processed_meshes++;
 
 			MinimapMapblock *minimap_mapblock = NULL;
 			bool do_mapper_update = true;
 
-			MeshUpdateResult r = m_mesh_update_thread.m_queue_out.pop_frontNoEx();
 			MapBlock *block = m_env.getMap().getBlockNoCreateNoEx(r.p);
 			if (block) {
 				// Delete the old mesh
@@ -1655,12 +1654,12 @@ void Client::addUpdateMeshTask(v3s16 p, bool ack_to_server, bool urgent)
 	if (b == NULL)
 		return;
 
-	m_mesh_update_thread.updateBlock(&m_env.getMap(), p, ack_to_server, urgent);
+	m_mesh_update_manager.updateBlock(&m_env.getMap(), p, ack_to_server, urgent);
 }
 
 void Client::addUpdateMeshTaskWithEdge(v3s16 blockpos, bool ack_to_server, bool urgent)
 {
-	m_mesh_update_thread.updateBlock(&m_env.getMap(), blockpos, ack_to_server, urgent, true);
+	m_mesh_update_manager.updateBlock(&m_env.getMap(), blockpos, ack_to_server, urgent, true);
 }
 
 void Client::addUpdateMeshTaskForNode(v3s16 nodepos, bool ack_to_server, bool urgent)
@@ -1674,7 +1673,7 @@ void Client::addUpdateMeshTaskForNode(v3s16 nodepos, bool ack_to_server, bool ur
 
 	v3s16 blockpos = getNodeBlockPos(nodepos);
 	v3s16 blockpos_relative = blockpos * MAP_BLOCKSIZE;
-	m_mesh_update_thread.updateBlock(&m_env.getMap(), blockpos, ack_to_server, urgent, false);
+	m_mesh_update_manager.updateBlock(&m_env.getMap(), blockpos, ack_to_server, urgent, false);
 	// Leading edge
 	if (nodepos.X == blockpos_relative.X)
 		addUpdateMeshTask(blockpos + v3s16(-1, 0, 0), false, urgent);
@@ -1793,7 +1792,7 @@ void Client::afterContentReceived()
 
 	// Start mesh update thread after setting up content definitions
 	infostream<<"- Starting mesh update thread"<<std::endl;
-	m_mesh_update_thread.start();
+	m_mesh_update_manager.start();
 
 	m_state = LC_Ready;
 	sendReady();
diff --git a/src/client/client.h b/src/client/client.h
index 52ca87625..c05e6d325 100644
--- a/src/client/client.h
+++ b/src/client/client.h
@@ -313,7 +313,7 @@ public:
 	void addUpdateMeshTaskForNode(v3s16 nodepos, bool ack_to_server=false, bool urgent=false);
 
 	void updateCameraOffset(v3s16 camera_offset)
-	{ m_mesh_update_thread.m_camera_offset = camera_offset; }
+	{ m_mesh_update_manager.m_camera_offset = camera_offset; }
 
 	bool hasClientEvents() const { return !m_client_event_queue.empty(); }
 	// Get event from queue. If queue is empty, it triggers an assertion failure.
@@ -483,7 +483,7 @@ private:
 	RenderingEngine *m_rendering_engine;
 
 
-	MeshUpdateThread m_mesh_update_thread;
+	MeshUpdateManager m_mesh_update_manager;
 	ClientEnvironment m_env;
 	ParticleManager m_particle_manager;
 	std::unique_ptr<con::Connection> m_con;
diff --git a/src/client/mesh_generator_thread.cpp b/src/client/mesh_generator_thread.cpp
index ec567c8c1..1456b26ef 100644
--- a/src/client/mesh_generator_thread.cpp
+++ b/src/client/mesh_generator_thread.cpp
@@ -146,16 +146,26 @@ QueuedMeshUpdate *MeshUpdateQueue::pop()
 	for (std::vector<QueuedMeshUpdate*>::iterator i = m_queue.begin();
 			i != m_queue.end(); ++i) {
 		QueuedMeshUpdate *q = *i;
-		if(must_be_urgent && m_urgents.count(q->p) == 0)
+		if (must_be_urgent && m_urgents.count(q->p) == 0)
+			continue;
+		// Make sure no two threads are processing the same mapblock, as that causes racing conditions
+		if (m_inflight_blocks.find(q->p) != m_inflight_blocks.end())
 			continue;
 		m_queue.erase(i);
 		m_urgents.erase(q->p);
+		m_inflight_blocks.insert(q->p);
 		fillDataFromMapBlockCache(q);
 		return q;
 	}
 	return NULL;
 }
 
+void MeshUpdateQueue::done(v3s16 pos)
+{
+	MutexAutoLock lock(m_mutex);
+	m_inflight_blocks.erase(pos);
+}
+
 CachedMapBlockData* MeshUpdateQueue::cacheBlock(Map *map, v3s16 p, UpdateMode mode,
 			size_t *cache_hit_counter)
 {
@@ -264,18 +274,62 @@ void MeshUpdateQueue::cleanupCache()
 }
 
 /*
-	MeshUpdateThread
+	MeshUpdateWorkerThread
 */
 
-MeshUpdateThread::MeshUpdateThread(Client *client):
-	UpdateThread("Mesh"),
-	m_queue_in(client)
+MeshUpdateWorkerThread::MeshUpdateWorkerThread(MeshUpdateQueue *queue_in, MeshUpdateManager *manager, v3s16 *camera_offset) :
+		UpdateThread("Mesh"), m_queue_in(queue_in), m_manager(manager), m_camera_offset(camera_offset)
 {
 	m_generation_interval = g_settings->getU16("mesh_generation_interval");
 	m_generation_interval = rangelim(m_generation_interval, 0, 50);
 }
 
-void MeshUpdateThread::updateBlock(Map *map, v3s16 p, bool ack_block_to_server,
+void MeshUpdateWorkerThread::doUpdate()
+{
+	QueuedMeshUpdate *q;
+	while ((q = m_queue_in->pop())) {
+		if (m_generation_interval)
+			sleep_ms(m_generation_interval);
+		ScopeProfiler sp(g_profiler, "Client: Mesh making (sum)");
+
+		MapBlockMesh *mesh_new = new MapBlockMesh(q->data, *m_camera_offset);
+
+		
+
+		MeshUpdateResult r;
+		r.p = q->p;
+		r.mesh = mesh_new;
+		r.ack_block_to_server = q->ack_block_to_server;
+		r.urgent = q->urgent;
+
+		m_manager->putResult(r);
+		m_queue_in->done(q->p);
+		delete q;
+	}
+}
+
+/*
+	MeshUpdateManager
+*/
+
+MeshUpdateManager::MeshUpdateManager(Client *client):
+	m_queue_in(client)
+{
+	int number_of_threads = rangelim(g_settings->getS32("mesh_generation_threads"), 0, 8);
+
+	// Automatically use 33% of the system cores for mesh generation, max 4
+	if (number_of_threads == 0)
+		number_of_threads = MYMIN(4, Thread::getNumberOfProcessors() / 3);
+	
+	// use at least one thread
+	number_of_threads = MYMAX(1, number_of_threads);
+	infostream << "MeshUpdateManager: using " << number_of_threads << " threads" << std::endl;
+
+	for (int i = 0; i < number_of_threads; i++)
+		m_workers.push_back(std::make_unique<MeshUpdateWorkerThread>(&m_queue_in, this, &m_camera_offset));
+}
+
+void MeshUpdateManager::updateBlock(Map *map, v3s16 p, bool ack_block_to_server,
 		bool urgent, bool update_neighbors)
 {
 	static thread_local const bool many_neighbors =
@@ -298,24 +352,57 @@ void MeshUpdateThread::updateBlock(Map *map, v3s16 p, bool ack_block_to_server,
 	deferUpdate();
 }
 
-void MeshUpdateThread::doUpdate()
+void MeshUpdateManager::putResult(const MeshUpdateResult &result)
 {
-	QueuedMeshUpdate *q;
-	while ((q = m_queue_in.pop())) {
-		if (m_generation_interval)
-			sleep_ms(m_generation_interval);
-		ScopeProfiler sp(g_profiler, "Client: Mesh making (sum)");
+	if (result.urgent)
+		m_queue_out_urgent.push_back(result);
+	else
+		m_queue_out.push_back(result);
+}
 
-		MapBlockMesh *mesh_new = new MapBlockMesh(q->data, m_camera_offset);
+bool MeshUpdateManager::getNextResult(MeshUpdateResult &r)
+{
+	if (!m_queue_out_urgent.empty()) {
+		r = m_queue_out_urgent.pop_frontNoEx();
+		return true;
+	}
 
-		MeshUpdateResult r;
-		r.p = q->p;
-		r.mesh = mesh_new;
-		r.ack_block_to_server = q->ack_block_to_server;
-		r.urgent = q->urgent;
+	if (!m_queue_out.empty()) {
+		r = m_queue_out.pop_frontNoEx();
+		return true;
+	}
 
-		m_queue_out.push_back(r);
+	return false;
+}
 
-		delete q;
-	}
+void MeshUpdateManager::deferUpdate()
+{
+	for (auto &thread : m_workers)
+		thread->deferUpdate();
+}
+
+void MeshUpdateManager::start()
+{
+	for (auto &thread: m_workers)
+		thread->start();
+}
+
+void MeshUpdateManager::stop()
+{
+	for (auto &thread: m_workers)
+		thread->stop();
+}
+
+void MeshUpdateManager::wait()
+{
+	for (auto &thread: m_workers)
+		thread->wait();
+}
+
+bool MeshUpdateManager::isRunning()
+{
+	for (auto &thread: m_workers)
+		if (thread->isRunning())
+			return true;
+	return false;
 }
diff --git a/src/client/mesh_generator_thread.h b/src/client/mesh_generator_thread.h
index 09400196d..bbb84b74a 100644
--- a/src/client/mesh_generator_thread.h
+++ b/src/client/mesh_generator_thread.h
@@ -26,6 +26,8 @@ with this program; if not, write to the Free Software Foundation, Inc.,
 #include "mapblock_mesh.h"
 #include "threading/mutex_auto_lock.h"
 #include "util/thread.h"
+#include <vector>
+#include <memory>
 
 struct CachedMapBlockData
 {
@@ -75,6 +77,9 @@ public:
 	// Returns NULL if queue is empty
 	QueuedMeshUpdate *pop();
 
+	// Marks a position as finished, unblocking the next update
+	void done(v3s16 pos);
+
 	u32 size()
 	{
 		MutexAutoLock lock(m_mutex);
@@ -86,6 +91,7 @@ private:
 	std::vector<QueuedMeshUpdate *> m_queue;
 	std::unordered_set<v3s16> m_urgents;
 	std::unordered_map<v3s16, CachedMapBlockData *> m_cache;
+	std::unordered_set<v3s16> m_inflight_blocks;
 	u64 m_next_cache_cleanup; // milliseconds
 	std::mutex m_mutex;
 
@@ -111,25 +117,53 @@ struct MeshUpdateResult
 	MeshUpdateResult() = default;
 };
 
-class MeshUpdateThread : public UpdateThread
+class MeshUpdateManager;
+
+class MeshUpdateWorkerThread : public UpdateThread
+{
+public:
+	MeshUpdateWorkerThread(MeshUpdateQueue *queue_in, MeshUpdateManager *manager, v3s16 *camera_offset);
+
+protected:
+	virtual void doUpdate();
+
+private:
+	MeshUpdateQueue *m_queue_in;
+	MeshUpdateManager *m_manager;
+	v3s16 *m_camera_offset;
+
+	// TODO: Add callback to update these when g_settings changes
+	int m_generation_interval;
+};
+
+class MeshUpdateManager
 {
 public:
-	MeshUpdateThread(Client *client);
+	MeshUpdateManager(Client *client);
 
 	// Caches the block at p and its neighbors (if needed) and queues a mesh
 	// update for the block at p
 	void updateBlock(Map *map, v3s16 p, bool ack_block_to_server, bool urgent,
 			bool update_neighbors = false);
+	void putResult(const MeshUpdateResult &r);
+	bool getNextResult(MeshUpdateResult &r);
+
 
 	v3s16 m_camera_offset;
-	MutexedQueue<MeshUpdateResult> m_queue_out;
+
+	void start();
+	void stop();
+	void wait();
+
+	bool isRunning();
 
 private:
-	MeshUpdateQueue m_queue_in;
+	void deferUpdate();
 
-	// TODO: Add callback to update these when g_settings changes
-	int m_generation_interval;
 
-protected:
-	virtual void doUpdate();
+	MeshUpdateQueue m_queue_in;
+	MutexedQueue<MeshUpdateResult> m_queue_out;
+	MutexedQueue<MeshUpdateResult> m_queue_out_urgent;
+
+	std::vector<std::unique_ptr<MeshUpdateWorkerThread>> m_workers;
 };
diff --git a/src/client/tile.cpp b/src/client/tile.cpp
index 0336bd82a..582219bb8 100644
--- a/src/client/tile.cpp
+++ b/src/client/tile.cpp
@@ -491,16 +491,16 @@ u32 TextureSource::getTextureId(const std::string &name)
 	infostream<<"getTextureId(): Queued: name=\""<<name<<"\""<<std::endl;
 
 	// We're gonna ask the result to be put into here
-	static ResultQueue<std::string, u32, u8, u8> result_queue;
+	static thread_local ResultQueue<std::string, u32, u8, u8> result_queue;
 
 	// Throw a request in
 	m_get_texture_queue.add(name, 0, 0, &result_queue);
 
 	try {
 		while(true) {
-			// Wait result for a second
+			// Wait for result for up to 4 seconds (empirical value)
 			GetResult<std::string, u32, u8, u8>
-				result = result_queue.pop_front(1000);
+				result = result_queue.pop_front(4000);
 
 			if (result.key == name) {
 				return result.item;
diff --git a/src/defaultsettings.cpp b/src/defaultsettings.cpp
index e8170788a..3639ae292 100644
--- a/src/defaultsettings.cpp
+++ b/src/defaultsettings.cpp
@@ -44,6 +44,7 @@ void set_default_settings()
 	settings->setDefault("mute_sound", "false");
 	settings->setDefault("enable_mesh_cache", "false");
 	settings->setDefault("mesh_generation_interval", "0");
+	settings->setDefault("mesh_generation_threads", "0");
 	settings->setDefault("meshgen_block_cache_size", "20");
 	settings->setDefault("enable_vbo", "true");
 	settings->setDefault("free_move", "false");
diff --git a/src/network/clientpackethandler.cpp b/src/network/clientpackethandler.cpp
index a98fa0733..829b1c3e6 100644
--- a/src/network/clientpackethandler.cpp
+++ b/src/network/clientpackethandler.cpp
@@ -673,7 +673,7 @@ void Client::handleCommand_AnnounceMedia(NetworkPacket* pkt)
 
 	// Mesh update thread must be stopped while
 	// updating content definitions
-	sanity_check(!m_mesh_update_thread.isRunning());
+	sanity_check(!m_mesh_update_manager.isRunning());
 
 	for (u16 i = 0; i < num_files; i++) {
 		std::string name, sha1_base64;
@@ -733,7 +733,7 @@ void Client::handleCommand_Media(NetworkPacket* pkt)
 	if (init_phase) {
 		// Mesh update thread must be stopped while
 		// updating content definitions
-		sanity_check(!m_mesh_update_thread.isRunning());
+		sanity_check(!m_mesh_update_manager.isRunning());
 	}
 
 	for (u32 i = 0; i < num_files; i++) {
@@ -770,7 +770,7 @@ void Client::handleCommand_NodeDef(NetworkPacket* pkt)
 
 	// Mesh update thread must be stopped while
 	// updating content definitions
-	sanity_check(!m_mesh_update_thread.isRunning());
+	sanity_check(!m_mesh_update_manager.isRunning());
 
 	// Decompress node definitions
 	std::istringstream tmp_is(pkt->readLongString(), std::ios::binary);
@@ -789,7 +789,7 @@ void Client::handleCommand_ItemDef(NetworkPacket* pkt)
 
 	// Mesh update thread must be stopped while
 	// updating content definitions
-	sanity_check(!m_mesh_update_thread.isRunning());
+	sanity_check(!m_mesh_update_manager.isRunning());
 
 	// Decompress item definitions
 	std::istringstream tmp_is(pkt->readLongString(), std::ios::binary);