fixed deadlock in zone update

Makefile

@@ -18,11 +18,11 @@ run: build
 eclipse:
 	$(Q)./fips config linux-eclipse-debug
 
-server:
-	$(Q)./fips run server
+server: build
+	$(Q)./fips run server -- -set core_loglevel 4
 
-client:
-	$(Q)./fips run client
+client: build
+	$(Q)./fips run client -- -set core_loglevel 2
 
-generate:
+generate: build
 	$(Q)./fips run worldgenerator -- -set seed 1 -set size 64

cmake/common.cmake

@@ -22,3 +22,40 @@ macro(check_lua_files TARGET files)
 		message(WARNING "No ${DEFAULT_LUA_EXECUTABLE} found")
 	endif()
 endmacro()
+
+include(CheckCCompilerFlag)
+
+# thread sanitizer doesn't work in combination with address and leak
+set(CMAKE_REQUIRED_FLAGS "-Werror -fsanitize=thread")
+check_c_compiler_flag("-fsanitize=thread" HAVE_FLAG_SANITIZE_THREAD)
+set(CMAKE_REQUIRED_FLAGS "-Werror -fsanitize=undefined")
+check_c_compiler_flag("-fsanitize=undefined" HAVE_FLAG_SANITIZE_UNDEFINED)
+#set(CMAKE_REQUIRED_FLAGS "-Werror -fsanitize=address")
+#check_c_compiler_flag("-fsanitize=address" HAVE_FLAG_SANITIZE_ADDRESS)
+#set(CMAKE_REQUIRED_FLAGS "-Werror -fsanitize=leak")
+#check_c_compiler_flag("-fsanitize=leak" HAVE_FLAG_SANITIZE_LEAK)
+set(CMAKE_REQUIRED_FLAGS "-Werror -fexpensive-optimizations")
+check_c_compiler_flag("-fexpensive-optimizations" HAVE_EXPENSIVE_OPTIMIZATIONS)
+unset(CMAKE_REQUIRED_FLAGS)
+
+#-Wthread-safety - http://clang.llvm.org/docs/ThreadSafetyAnalysis.html
+
+if (HAVE_FLAG_SANITIZE_UNDEFINED)
+	set(SANITIZE_FLAGS "${SANITIZE_FLAGS} -fsanitize=undefined")
+	message("Support undefined sanitizer")
+endif()
+
+if (HAVE_FLAG_SANITIZE_LEAK)
+	set(SANITIZE_FLAGS "${SANITIZE_FLAGS} -fsanitize=leak")
+	message("Support leak sanitizer")
+endif()
+
+if (HAVE_FLAG_SANITIZE_THREAD)
+	set(SANITIZE_FLAGS "${SANITIZE_FLAGS} -fsanitize=thread")
+	message("Support thread sanitizer")
+endif()
+
+if (HAVE_FLAG_SANITIZE_ADDRESS)
+	set(SANITIZE_FLAGS "${SANITIZE_FLAGS} -fsanitize=address")
+	message("Support address sanitizer")
+endif()

src/modules/backend/entity/ai/condition/IsCloseToSelection.h

@@ -31,17 +31,12 @@ public:
 		if (selection.empty())
 			return false;
 
-		auto func = [&] (const ai::AIPtr& ai) -> bool {
-			const Npc& npc = ai->getCharacterCast<AICharacter>().getNpc();
-			const glm::vec3& pos = npc.pos();
-			const ai::Vector3f& ownPos = entity->getCharacter()->getPosition();
-			const int distance = ownPos.distance(ai::Vector3f(pos.x, pos.y, pos.z));
-			return distance <= _distance;
-		};
-
-		const AIPtr& selected = zone->getAI(selection[0]);
-		auto future = zone->executeAsync(selected, func);
-		return future.get();
+		const AIPtr& ai = zone->getAI(selection[0]);
+		const Npc& npc = ai->getCharacterCast<AICharacter>().getNpc();
+		const glm::vec3& pos = npc.pos();
+		const ai::Vector3f& ownPos = entity->getCharacter()->getPosition();
+		const int distance = ownPos.distance(ai::Vector3f(pos.x, pos.y, pos.z));
+		return distance <= _distance;
 	}
 };
 

src/modules/backend/entity/ai/condition/IsSelectionAlive.h

@@ -17,14 +17,10 @@ public:
 		const FilteredEntities& selection = entity->getFilteredEntities();
 		if (selection.empty())
 			return true;
-		static const auto func = [] (const AIPtr& ai) -> bool {
-			AICharacter& chr = ai->getCharacterCast<AICharacter>();
-			return !chr.getNpc().dead();
-		};
 		const Zone* zone = entity->getZone();
-		const AIPtr& selected = zone->getAI(selection[0]);
-		auto future = zone->executeAsync(selected, func);
-		return future.get();
+		const AIPtr& ai = zone->getAI(selection[0]);
+		AICharacter& chr = ai->getCharacterCast<AICharacter>();
+		return !chr.getNpc().dead();
 	}
 };
 

src/modules/backend/loop/ServerLoop.cpp

@@ -1,6 +1,7 @@
 #include "ServerLoop.h"
 #include "core/Command.h"
 #include "core/Tokenizer.h"
+#include "core/Log.h"
 
 namespace backend {
 
@@ -24,10 +25,6 @@ bool ServerLoop::onInit() {
 
 	class ProgressMonitor: public util::IProgressMonitor {
 	public:
-		void step(int steps = 1) override {
-			IProgressMonitor::step(steps);
-			Log::info("max: %i, steps: %i => %f\r", _max, _steps, progress());
-		}
 		void done() override {
 			Log::info("\ndone");
 		}

src/modules/backend/network/UserConnectHandler.cpp

@@ -23,7 +23,7 @@ void UserConnectHandler::execute(ENetPeer* peer, const void* raw) {
 
 	const char *email = message->email()->c_str();
 	const char *password = message->password()->c_str();
-	Log::info("User %s tries to log into the gameserver", email);
+	Log::info("User \"%s\" tries to log into the gameserver", email);
 
 	UserPtr user = _entityStorage->login(peer, email, password);
 	if (!user) {

src/modules/backend/spawn/SpawnMgr.cpp

@@ -42,7 +42,7 @@ void SpawnMgr::spawnEntity(ai::Zone& zone, network::messages::NpcType start, net
 	const int offset = start + 1;
 	int count[end - offset];
 	memset(count, 0, sizeof(count));
-	zone.execute([&] (const ai::AIPtr& ai) {
+	zone.execute([start, end, offset, &count] (const ai::AIPtr& ai) {
 		const AICharacter& chr = ai::character_cast<AICharacter>(ai->getCharacter());
 		const Npc& npc = chr.getNpc();
 		const network::messages::NpcType type = npc.npcType();

src/modules/core/App.h

@@ -2,6 +2,7 @@
 
 #include <unordered_set>
 #include <chrono>
+#include "Common.h"
 #include "Trace.h"
 #include "EventBus.h"
 #include "io/Filesystem.h"

src/modules/core/CMakeLists.txt

@@ -16,7 +16,7 @@ fips_begin_module(core)
 		Log.cpp Log.h
 		Var.cpp Var.h
 		Trace.h
-		ThreadPool.h ThreadPool.cpp
+		ThreadPool.h
 		Tokenizer.h
 		TimeProvider.h TimeProvider.cpp
 	)

src/modules/core/ThreadPool.cpp

@@ -1,26 +0,0 @@
-#include "ThreadPool.h"
-
-namespace core {
-
-void ThreadPool::tick() {
-	FunctionType task;
-	for (;;) {
-		{
-			std::unique_lock<std::mutex> lock(_mutex);
-			while (!_join && _tasks.empty()) {
-				_cond.wait(lock);
-			}
-			if (_join) {
-				core_assert(_tasks.empty());
-				return;
-			}
-
-			task = _tasks.front();
-			_tasks.pop_front();
-		}
-
-		task();
-	}
-}
-
-}

src/modules/core/ThreadPool.h

@@ -1,61 +1,112 @@
+/*
+ Copyright (c) 2012 Jakob Progsch, Václav Zeman
+
+ This software is provided 'as-is', without any express or implied
+ warranty. In no event will the authors be held liable for any damages
+ arising from the use of this software.
+
+ Permission is granted to anyone to use this software for any purpose,
+ including commercial applications, and to alter it and redistribute it
+ freely, subject to the following restrictions:
+
+ 1. The origin of this software must not be misrepresented; you must not
+ claim that you wrote the original software. If you use this software
+ in a product, an acknowledgment in the product documentation would be
+ appreciated but is not required.
+
+ 2. Altered source versions must be plainly marked as such, and must not be
+ misrepresented as being the original software.
+
+ 3. This notice may not be removed or altered from any source
+ distribution.
+ */
+
 #pragma once
 
-#include "Common.h"
+#include <vector>
 #include <queue>
 #include <memory>
 #include <thread>
 #include <mutex>
 #include <condition_variable>
+#include <atomic>
 #include <future>
 #include <functional>
-#include <stdexcept>
 
 namespace core {
 
-class ThreadPool {
-private:
-	typedef std::function<void()> FunctionType;
-	bool _join;
-	std::vector<std::thread> _workers;
-	std::deque<FunctionType> _tasks;
-	std::condition_variable _cond;
-	std::mutex _mutex;
-
-	void tick();
+class ThreadPool final {
 public:
-	ThreadPool(unsigned int numThreads = 2) :
-			_join(false) {
-		for (unsigned int i = 0; i < numThreads; ++i) {
-			_workers.push_back(std::move(std::thread(std::bind(&ThreadPool::tick, this))));
-		}
-	}
+	explicit ThreadPool(size_t);
 
-	template<class Function, class ... Args>
-	auto push(Function&& func, Args&&... args) -> std::future<typename std::result_of<Function(Args...)>::type> {
-		core_assert(!_join);
-		using retType = typename std::result_of<Function(Args...)>::type;
+	/**
+	 * Enqueue functors or lambdas into the thread pool
+	 */
+	template<class F, class ... Args>
+	auto enqueue(F&& f, Args&&... args) -> std::future<typename std::result_of<F(Args...)>::type>;
 
-		auto task = std::make_shared<std::packaged_task<retType()> >(std::bind(std::forward<Function>(func), std::forward<Args>(args)...));
-		std::future<retType> res = task->get_future();
+	~ThreadPool();
+private:
+	// need to keep track of threads so we can join them
+	std::vector<std::thread> _workers;
+	// the task queue
+	std::queue<std::function<void()> > _tasks;
 
-		auto funcWrapper = [task]() {
-			(*task)();
-		};
-		{
-			std::unique_lock<std::mutex> lock(_mutex);
-			_tasks.push_back(funcWrapper);
-		}
-		_cond.notify_one();
-		return res;
-	}
-
-	~ThreadPool() {
-		_join = true;
-		_cond.notify_all();
-		for (std::thread &worker : _workers) {
-			worker.join();
-		}
-	}
+	// synchronization
+	std::mutex _queueMutex;
+	std::condition_variable _condition;
+	std::atomic_bool _stop;
 };
 
+// the constructor just launches some amount of workers
+inline ThreadPool::ThreadPool(size_t threads) :
+		_stop(false) {
+	_workers.reserve(threads);
+	for (size_t i = 0; i < threads; ++i) {
+		_workers.emplace_back([this] {
+			for (;;) {
+				std::function<void()> task;
+				{
+					std::unique_lock<std::mutex> lock(this->_queueMutex);
+					this->_condition.wait(lock, [this] {
+						return this->_stop || !this->_tasks.empty();
+					});
+					if (this->_stop && this->_tasks.empty()) {
+						return;
+					}
+					task = std::move(this->_tasks.front());
+					this->_tasks.pop();
+				}
+
+				task();
+			}
+		});
+	}
+}
+
+// add new work item to the pool
+template<class F, class ... Args>
+auto ThreadPool::enqueue(F&& f, Args&&... args)
+-> std::future<typename std::result_of<F(Args...)>::type> {
+	using return_type = typename std::result_of<F(Args...)>::type;
+
+	auto task = std::make_shared<std::packaged_task<return_type()> >(std::bind(std::forward<F>(f), std::forward<Args>(args)...));
+
+	std::future<return_type> res = task->get_future();
+	{
+		std::unique_lock<std::mutex> lock(_queueMutex);
+		_tasks.emplace([task]() {(*task)();});
+		_condition.notify_one();
+	}
+	return res;
+}
+
+// the destructor joins all threads
+inline ThreadPool::~ThreadPool() {
+	_stop = true;
+	_condition.notify_all();
+	for (std::thread &worker : _workers)
+		worker.join();
+}
+
 }

src/modules/core/tests/ThreadPoolTest.cpp

@@ -14,7 +14,7 @@ public:
 
 TEST_F(ThreadPoolTest, testPush) {
 	core::ThreadPool pool(1);
-	auto future = pool.push([this] () {_executed = true;});
+	auto future = pool.enqueue([this] () {_executed = true;});
 	future.get();
 	ASSERT_TRUE(_executed) << "Thread wasn't executed";
 }
@@ -24,7 +24,7 @@ TEST_F(ThreadPoolTest, testMultiplePush) {
 	{
 		core::ThreadPool pool(2);
 		for (int i = 0; i < x; ++i) {
-			pool.push([this] () {_count++;});
+			pool.enqueue([this] () {_count++;});
 		}
 	}
 	ASSERT_EQ(x, _count) << "Not all threads were executed";

src/modules/io/Filesystem.h

@@ -40,7 +40,7 @@ public:
 	 */
 	template<class CompleteHandle>
 	void loadAsync(const std::string& filename, CompleteHandle completeHandle) {
-		_threadPool.push([=]() {
+		_threadPool.enqueue([=]() {
 			io::FilePtr f(new io::File(filename));
 			completeHandle(f);
 		});

src/modules/video/MeshPool.cpp

@@ -30,7 +30,7 @@ MeshPtr MeshPool::getMesh(const std::string& id) {
 		return i->second;
 
 	const MeshPtr mesh(new Mesh());
-	_threadPool.push([=]() {mesh->loadMesh(name);});
+	_threadPool.enqueue([=]() {mesh->loadMesh(name);});
 	_meshes[name] = mesh;
 	return mesh;
 }

src/modules/voxel/World.cpp

@@ -84,7 +84,7 @@ void World::scheduleMeshExtraction(const glm::ivec2& p) {
 	_meshesExtracted.insert(pos);
 
 	const int delta = size - 1;
-	_threadPool.push([=] () {
+	_threadPool.enqueue([=] () {
 		core_trace_scoped("MeshExtraction");
 		const PolyVox::Vector3DInt32 mins(pos.x, 0, pos.y);
 		const PolyVox::Vector3DInt32 maxs(mins.getX() + delta, MAX_HEIGHT, mins.getZ() + delta);