Basic Lan Implmentation

src/threading/CMakeLists.txt

@@ -2,5 +2,7 @@ set(JTHREAD_SRCS
 	${CMAKE_CURRENT_SOURCE_DIR}/event.cpp
 	${CMAKE_CURRENT_SOURCE_DIR}/thread.cpp
 	${CMAKE_CURRENT_SOURCE_DIR}/semaphore.cpp
+	${CMAKE_CURRENT_SOURCE_DIR}/lock.cpp
+	${CMAKE_CURRENT_SOURCE_DIR}/thread_vector.cpp
 	PARENT_SCOPE)
 

src/threading/concurrent_map.h

@@ -0,0 +1,209 @@
+/*
+Copyright (C) 2024 proller <proler@gmail.com>
+*/
+
+/*
+This file is part of Freeminer.
+
+Freeminer is free software: you can redistribute it and/or modify
+it under the terms of the GNU General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+Freeminer  is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License
+along with Freeminer.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#pragma once
+
+#include <map>
+
+#include "lock.h"
+
+template <class LOCKER, class Key, class T, class Compare = std::less<Key>,
+		class Allocator = std::allocator<std::pair<const Key, T>>>
+class concurrent_map_ : public std::map<Key, T, Compare, Allocator>, public LOCKER
+{
+public:
+	typedef typename std::map<Key, T, Compare, Allocator> full_type;
+	typedef Key key_type;
+	typedef T mapped_type;
+
+	mapped_type &operator[](const key_type &k) = delete;
+	mapped_type &operator[](key_type &&k) = delete;
+
+	mapped_type nothing = {};
+
+	template <typename... Args>
+	mapped_type& get(Args &&...args)
+	{
+		auto lock = LOCKER::lock_shared_rec();
+
+		//if (!full_type::contains(std::forward<Args>(args)...))
+		if (full_type::find(std::forward<Args>(args)...) == full_type::end())
+			return nothing;
+
+		return full_type::operator[](std::forward<Args>(args)...);
+	}
+
+	template <typename... Args>
+	decltype(auto) at(Args &&...args)
+	{
+		auto lock = LOCKER::lock_shared_rec();
+		return full_type::at(std::forward<Args>(args)...);
+	}
+
+	template <typename... Args>
+	decltype(auto) assign(Args &&...args)
+	{
+		auto lock = LOCKER::lock_unique_rec();
+		return full_type::assign(std::forward<Args>(args)...);
+	}
+
+	template <typename... Args>
+	decltype(auto) insert(Args &&...args)
+	{
+		auto lock = LOCKER::lock_unique_rec();
+		return full_type::insert(std::forward<Args>(args)...);
+	}
+
+	template <typename... Args>
+	decltype(auto) emplace(Args &&...args)
+	{
+		auto lock = LOCKER::lock_unique_rec();
+		return full_type::emplace(std::forward<Args>(args)...);
+	}
+
+	template <typename... Args>
+	decltype(auto) emplace_try(Args &&...args)
+	{
+		auto lock = LOCKER::try_lock_unique_rec();
+		if (!lock->owns_lock())
+			return false;
+		return full_type::emplace(std::forward<Args>(args)...).second;
+	}
+
+	template <typename... Args>
+	decltype(auto) insert_or_assign(Args &&...args)
+	{
+		auto lock = LOCKER::lock_unique_rec();
+		return full_type::insert_or_assign(std::forward<Args>(args)...);
+	}
+
+	template <typename... Args>
+	decltype(auto) empty(Args &&...args) const noexcept
+	{
+		auto lock = LOCKER::lock_shared_rec();
+		return full_type::empty(std::forward<Args>(args)...);
+	}
+
+	template <typename... Args>
+	decltype(auto) size(Args &&...args) const
+	{
+		auto lock = LOCKER::lock_shared_rec();
+		return full_type::size(std::forward<Args>(args)...);
+	}
+
+	template <typename... Args>
+	decltype(auto) count(Args &&...args) const
+	{
+		auto lock = LOCKER::lock_shared_rec();
+		return full_type::count(std::forward<Args>(args)...);
+	}
+
+	template <typename... Args>
+	decltype(auto) contains(Args &&...args) const
+	{
+		auto lock = LOCKER::lock_shared_rec();
+		return full_type::contains(std::forward<Args>(args)...);
+	}
+
+	template <typename... Args>
+	decltype(auto) find(Args &&...args)
+	{
+		auto lock = LOCKER::lock_shared_rec();
+		return full_type::find(std::forward<Args>(args)...);
+	}
+
+	template <typename... Args>
+	decltype(auto) begin(Args &&...args)
+	{
+		auto lock = LOCKER::lock_shared_rec();
+		return full_type::begin(std::forward<Args>(args)...);
+	}
+
+	template <typename... Args>
+	decltype(auto) rbegin(Args &&...args)
+	{
+		auto lock = LOCKER::lock_shared_rec();
+		return full_type::rbegin(std::forward<Args>(args)...);
+	}
+
+	template <typename... Args>
+	decltype(auto) end(Args &&...args)
+	{
+		auto lock = LOCKER::lock_shared_rec();
+		return full_type::end(std::forward<Args>(args)...);
+	}
+
+	template <typename... Args>
+	decltype(auto) rend(Args &&...args)
+	{
+		auto lock = LOCKER::lock_shared_rec();
+		return full_type::rend(std::forward<Args>(args)...);
+	}
+
+
+	template <typename... Args>
+	decltype(auto) erase(Args &&...args)
+	{
+		auto lock = LOCKER::lock_unique_rec();
+		return full_type::erase(std::forward<Args>(args)...);
+	}
+
+	template <typename... Args>
+	decltype(auto) clear(Args &&...args)
+	{
+		auto lock = LOCKER::lock_unique_rec();
+		return full_type::clear(std::forward<Args>(args)...);
+	}
+};
+
+template <class Key, class T, class Compare = std::less<Key>,
+		class Allocator = std::allocator<std::pair<const Key, T>>>
+using concurrent_map = concurrent_map_<locker<>, Key, T, Compare, Allocator>;
+
+template <class Key, class T, class Compare = std::less<Key>, class Allocator = std::allocator<std::pair<const Key, T>>>
+using concurrent_shared_map = concurrent_map_<shared_locker, Key, T, Compare, Allocator>;
+
+#if ENABLE_THREADS
+
+template <class Key, class T, class Compare = std::less<Key>,
+		class Allocator = std::allocator<std::pair<const Key, T>>>
+using maybe_concurrent_map = concurrent_map<Key, T, Compare, Allocator>;
+
+#else
+
+template <class Key, class T, class Compare = std::less<Key>,
+		class Allocator = std::allocator<std::pair<const Key, T>>>
+class not_concurrent_map : public std::map<Key, T, Compare, Allocator>,
+						   public dummy_locker
+{
+public:
+	typedef typename std::map<Key, T, Compare, Allocator> full_type;
+	typedef Key key_type;
+	typedef T mapped_type;
+
+	mapped_type &get(const key_type &k) { return full_type::operator[](k); }
+};
+
+template <class Key, class T, class Compare = std::less<Key>,
+		class Allocator = std::allocator<std::pair<const Key, T>>>
+using maybe_concurrent_map = not_concurrent_map<Key, T, Compare, Allocator>;
+
+#endif

src/threading/lock.cpp

@@ -0,0 +1,131 @@
+#include "lock.h"
+#include "log.h"
+#include "profiler.h"
+
+#if !defined(NDEBUG) && !defined(LOCK_PROFILE)
+//#define LOCK_PROFILE 1
+#endif
+
+#if LOCK_PROFILE
+#define SCOPE_PROFILE(a) ScopeProfiler scp___(g_profiler, "Lock: " a);
+#else
+#define SCOPE_PROFILE(a)
+#endif
+
+template<class GUARD, class MUTEX>
+recursive_lock<GUARD, MUTEX>::recursive_lock(MUTEX & mtx, std::atomic<std::size_t> & thread_id_, bool try_lock):
+	thread_id(thread_id_) {
+	auto thread_me = std::hash<std::thread::id>()(std::this_thread::get_id());
+	if(thread_me != thread_id) {
+		if (try_lock) {
+			SCOPE_PROFILE("try_lock");
+			lock = new GUARD(mtx, try_to_lock);
+			if (lock->owns_lock()) {
+				thread_id = thread_me;
+				return;
+			} else {
+#if LOCK_PROFILE
+				g_profiler->add("Lock: try_lock fail", 1);
+#endif
+				//infostream<<"not locked "<<" thread="<<thread_id<<" lock="<<lock<<std::endl;
+			}
+			delete lock;
+		} else {
+			SCOPE_PROFILE("lock");
+			lock = new GUARD(mtx);
+			thread_id = thread_me;
+			return;
+		}
+	} else {
+#if LOCK_PROFILE
+		g_profiler->add("Lock: recursive", 1);
+#endif
+	}
+	lock = nullptr;
+}
+
+template<class GUARD, class MUTEX>
+recursive_lock<GUARD, MUTEX>::~recursive_lock() {
+	unlock();
+}
+
+template<class GUARD, class MUTEX>
+bool recursive_lock<GUARD, MUTEX>::owns_lock() {
+	if (lock)
+		return lock;
+	auto thread_me = std::hash<std::thread::id>()(std::this_thread::get_id());
+	return thread_id == thread_me;
+}
+
+template<class GUARD, class MUTEX>
+void recursive_lock<GUARD, MUTEX>::unlock() {
+	if(lock) {
+		thread_id = 0;
+		lock->unlock();
+		delete lock;
+		lock = nullptr;
+	}
+}
+
+
+template<class mutex, class unique_lock, class shared_lock>
+locker<mutex, unique_lock, shared_lock>::locker() {
+	thread_id = 0;
+}
+
+template<class mutex, class unique_lock, class shared_lock>
+std::unique_ptr<unique_lock> locker<mutex, unique_lock, shared_lock>::lock_unique() {
+	return std::make_unique<unique_lock>(mtx);
+}
+
+template<class mutex, class unique_lock, class shared_lock>
+std::unique_ptr<unique_lock> locker<mutex, unique_lock, shared_lock>::try_lock_unique() {
+	SCOPE_PROFILE("locker::try_lock_unique");
+	return std::make_unique<unique_lock>(mtx, std::try_to_lock);
+}
+
+template<class mutex, class unique_lock, class shared_lock>
+std::unique_ptr<shared_lock> locker<mutex, unique_lock, shared_lock>::lock_shared() const {
+	SCOPE_PROFILE("locker::lock_shared");
+	return std::make_unique<shared_lock>(mtx);
+}
+
+template<class mutex, class unique_lock, class shared_lock>
+std::unique_ptr<shared_lock> locker<mutex, unique_lock, shared_lock>::try_lock_shared() {
+	SCOPE_PROFILE("locker::try_lock_shared");
+	return std::make_unique<shared_lock>(mtx, std::try_to_lock);
+}
+
+template<class mutex, class unique_lock, class shared_lock>
+std::unique_ptr<recursive_lock<unique_lock, mutex>> locker<mutex, unique_lock, shared_lock>::lock_unique_rec() const {
+	SCOPE_PROFILE("locker::lock_unique_rec");
+	return std::make_unique<lock_rec_unique>(mtx, thread_id);
+}
+
+template<class mutex, class unique_lock, class shared_lock>
+std::unique_ptr<recursive_lock<unique_lock, mutex>> locker<mutex, unique_lock, shared_lock>::try_lock_unique_rec() {
+	SCOPE_PROFILE("locker::try_lock_unique_rec");
+	return std::make_unique<lock_rec_unique>(mtx, thread_id, true);
+}
+
+template<class mutex, class unique_lock, class shared_lock>
+std::unique_ptr<recursive_lock<shared_lock, mutex>> locker<mutex, unique_lock, shared_lock>::lock_shared_rec() const {
+	SCOPE_PROFILE("locker::lock_shared_rec");
+	return std::make_unique<lock_rec_shared>(mtx, thread_id);
+}
+
+template<class mutex, class unique_lock, class shared_lock>
+std::unique_ptr<recursive_lock<shared_lock, mutex>> locker<mutex, unique_lock, shared_lock>::try_lock_shared_rec() {
+	SCOPE_PROFILE("locker::try_lock_shared_rec");
+	return std::make_unique<lock_rec_shared>(mtx, thread_id, true);
+}
+
+
+template class recursive_lock<std::unique_lock<use_mutex>>;
+template class locker<>;
+#if LOCK_TWO
+template class recursive_lock<try_shared_lock, try_shared_mutex>;
+template class recursive_lock<std::unique_lock<try_shared_mutex>, try_shared_mutex>;
+
+template class locker<try_shared_mutex, std::unique_lock<try_shared_mutex>, std::shared_lock<try_shared_mutex>>;
+#endif

src/threading/lock.h

@@ -0,0 +1,164 @@
+/*
+This file is part of Freeminer.
+
+Freeminer is free software: you can redistribute it and/or modify
+it under the terms of the GNU General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+Freeminer  is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License
+along with Freeminer.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#pragma once
+
+#include <mutex>
+#include <atomic>
+#include <thread>
+#include <memory>
+
+#include "../config.h"
+
+#ifdef _WIN32
+
+//#include "../threading/mutex.h"
+using use_mutex = std::mutex;
+using try_shared_mutex = use_mutex;
+using try_shared_lock = std::unique_lock<try_shared_mutex>;
+using unique_lock = std::unique_lock<try_shared_mutex>;
+const auto try_to_lock = std::try_to_lock;
+
+#else
+
+typedef std::mutex use_mutex;
+
+
+#if USE_BOOST // not finished
+
+//#include <ctime>
+#include <boost/thread.hpp>
+//#include <boost/thread/locks.hpp>
+typedef boost::shared_mutex try_shared_mutex;
+typedef boost::shared_lock<try_shared_mutex> try_shared_lock;
+typedef boost::unique_lock<try_shared_mutex> unique_lock;
+const auto try_to_lock = boost::try_to_lock;
+#define LOCK_TWO 1
+
+#elif HAVE_SHARED_MUTEX
+//#elif __cplusplus >= 201305L
+
+#include <shared_mutex>
+using try_shared_mutex = std::shared_mutex;
+using try_shared_lock = std::shared_lock<try_shared_mutex>;
+using unique_lock = std::unique_lock<try_shared_mutex>;
+const auto try_to_lock = std::try_to_lock;
+#define LOCK_TWO 1
+
+#else
+
+using try_shared_mutex = use_mutex;
+using try_shared_lock = std::unique_lock<try_shared_mutex> ;
+using unique_lock = std::unique_lock<try_shared_mutex> ;
+const auto try_to_lock = std::try_to_lock;
+#endif
+
+#endif
+
+
+// http://stackoverflow.com/questions/4792449/c0x-has-no-semaphores-how-to-synchronize-threads
+/* uncomment when need
+#include <condition_variable>
+class semaphore {
+private:
+	std::mutex mtx;
+	std::condition_variable cv;
+	int count;
+
+public:
+	semaphore(int count_ = 0):count(count_){;}
+	void notify() {
+	std::unique_lock<std::mutex> lck(mtx);
+		++count;
+		cv.notify_one();
+	}
+	void wait() {
+		std::unique_lock<std::mutex> lck(mtx);
+		while(count == 0){
+			cv.wait(lck);
+		}
+		count--;
+	}
+};
+*/
+
+template<class GUARD, class MUTEX = use_mutex>
+class recursive_lock {
+public:
+	GUARD * lock;
+	std::atomic<std::size_t> & thread_id;
+	recursive_lock(MUTEX & mtx, std::atomic<std::size_t> & thread_id_, bool try_lock = false);
+	~recursive_lock();
+	bool owns_lock();
+	void unlock();
+};
+
+template<class mutex = use_mutex, class unique_lock = std::unique_lock<mutex> , class shared_lock = std::unique_lock<mutex> >
+class locker {
+public:
+	using lock_rec_shared = recursive_lock<shared_lock, mutex>;
+	using lock_rec_unique = recursive_lock<unique_lock, mutex>;
+
+	mutable mutex mtx;
+	mutable std::atomic<std::size_t> thread_id;
+
+	locker();
+	std::unique_ptr<unique_lock> lock_unique();
+	std::unique_ptr<unique_lock> try_lock_unique();
+	std::unique_ptr<shared_lock> lock_shared() const;
+	std::unique_ptr<shared_lock> try_lock_shared();
+	std::unique_ptr<lock_rec_unique> lock_unique_rec() const;
+	std::unique_ptr<lock_rec_unique> try_lock_unique_rec();
+	std::unique_ptr<lock_rec_shared> lock_shared_rec() const;
+	std::unique_ptr<lock_rec_shared> try_lock_shared_rec();
+};
+
+using shared_locker = locker<try_shared_mutex, unique_lock, try_shared_lock>;
+
+class dummy_lock {
+public:
+	~dummy_lock() {}; //no unused variable warning
+	bool owns_lock() {return true;}
+	bool operator!() {return true;}
+	dummy_lock * operator->() {return this; }
+	void unlock() {};
+};
+
+class dummy_locker {
+public:
+	dummy_lock lock_unique() { return {}; };
+	dummy_lock try_lock_unique() { return {}; };
+	dummy_lock lock_shared() { return {}; };
+	dummy_lock try_lock_shared() { return {}; };
+	dummy_lock lock_unique_rec() { return {}; };
+	dummy_lock try_lock_unique_rec() { return {}; };
+	dummy_lock lock_shared_rec() { return {}; };
+	dummy_lock try_lock_shared_rec() { return {}; };
+};
+
+
+#if ENABLE_THREADS
+
+using maybe_locker = locker<>;
+using maybe_shared_locker = shared_locker;
+
+#else
+
+using maybe_locker = dummy_locker;
+using maybe_shared_locker = dummy_locker;
+
+#endif

src/threading/thread_vector.cpp

@@ -0,0 +1,118 @@
+#include "thread_vector.h"
+#include "fm_porting.h"
+#include "log.h"
+#include "porting.h"
+
+thread_vector::thread_vector(const std::string &name, int priority) :
+		m_name(name), m_priority(priority)
+{
+	request_stop = false;
+};
+
+thread_vector::~thread_vector()
+{
+	join();
+};
+
+void thread_vector::func()
+{
+	reg();
+	run();
+};
+
+void thread_vector::reg(const std::string &name, int priority)
+{
+	if (!name.empty())
+		m_name = name;
+
+	porting::setThreadName(m_name.c_str());
+	g_logger.registerThread(m_name);
+
+	if (priority)
+		m_priority = priority;
+	if (m_priority)
+		porting::setThreadPriority(m_priority);
+};
+
+void thread_vector::start(const size_t n)
+{
+#if !NDEBUG
+	infostream << "start thread " << m_name << " n=" << n << std::endl;
+#endif
+	request_stop = false;
+	for (size_t i = 0; i < n; ++i) {
+		workers.emplace_back(&thread_vector::func, this);
+	}
+}
+
+void thread_vector::stop()
+{
+	request_stop = true;
+}
+
+void thread_vector::join()
+{
+	stop();
+	for (auto &worker : workers) {
+		try {
+			if (worker.joinable()) {
+				worker.join();
+			}
+		} catch (...) {
+		}
+	}
+	workers.clear();
+}
+
+void thread_vector::restart(size_t n)
+{
+	join();
+	start(n);
+}
+void thread_vector::reanimate(size_t n)
+{
+	if (workers.empty()) {
+		start(n);
+	}
+}
+
+void thread_vector::sleep(const int seconds)
+{
+	for (int i = 0; i <= seconds; ++i) {
+		std::this_thread::sleep_for(std::chrono::seconds(1));
+		if (request_stop) {
+			return;
+		}
+	}
+}
+
+// JThread compat:
+bool thread_vector::stopRequested()
+{
+	return request_stop;
+}
+bool thread_vector::isRunning()
+{
+	return !workers.empty();
+}
+void thread_vector::wait()
+{
+	join();
+};
+void thread_vector::kill()
+{
+	join();
+};
+void *thread_vector::run()
+{
+	return nullptr;
+};
+
+bool thread_vector::isCurrentThread()
+{
+	auto thread_me = std::hash<std::thread::id>()(std::this_thread::get_id());
+	for (auto &worker : workers)
+		if (thread_me == std::hash<std::thread::id>()(worker.get_id()))
+			return true;
+	return false;
+}

src/threading/thread_vector.h

@@ -0,0 +1,37 @@
+#pragma once
+
+#include <atomic>
+#include <thread>
+#include <vector>
+#include <string>
+
+class thread_vector {
+public:
+	std::vector<std::thread> workers;
+	std::atomic_bool request_stop;
+
+	thread_vector(const std::string &name = "Unnamed", int priority = 0);
+	virtual ~thread_vector();
+
+	virtual void func();
+
+	void reg (const std::string &name = "", int priority = 0);
+	void start (const size_t n = 1);
+	void restart (const size_t n = 1);
+	void reanimate (const size_t n = 1);
+	void stop ();
+	void join ();
+
+	void sleep(const int second);
+// Thread compat:
+
+	bool stopRequested();
+	bool isRunning();
+	void wait();
+	void kill();
+	virtual void * run() = 0;
+	bool isCurrentThread();
+protected:
+	std::string m_name;
+	int m_priority = 0;
+};