<commit_before>#ifndef ELEKTRA_KDBTHREAD_HPP
#define ELEKTRA_KDBTHREAD_HPP

#include <kdbcontext.hpp>

#include <kdb.hpp>

#include <mutex>
#include <thread>
#include <vector>
#include <cassert>
#include <algorithm>
#include <functional>
#include <unordered_map>

namespace kdb
{

/// Subject from Observer pattern for ThreadContext
class ThreadSubject
{
public:
	virtual void notify(KeySet &ks) = 0;
};

/// A vector of layers
typedef std::vector<std::shared_ptr<Layer>> LayerVector;

typedef std::unordered_map<std::string, std::function<void()>> FunctionMap;

/// A data structure that is stored by context inside the Coordinator
struct PerContext
{
	KeySet toUpdate;
	LayerVector toActivate;
	LayerVector toDeactivate;
};

/**
 * @brief Thread safe coordination of ThreadContext per Threads.
 */
class Coordinator
{
public:
	void onLayerActivation(std::string layerid, std::function <void()> f)
	{
		std::lock_guard<std::mutex> lock (m_mutexOnActivate);
		m_onActivate[layerid] = f;
	}

	void onLayerDeactivation(std::string layerid, std::function <void()> f)
	{
		std::lock_guard<std::mutex> lock (m_mutexOnDeactivate);
		m_onDeactivate[layerid] = f;
	}

private:
	friend class ThreadContext;

	void attach(ThreadSubject *c)
	{
		std::lock_guard<std::mutex> lock (m_mutex);
		m_updates.insert(std::make_pair(c, PerContext()));
	}

	void detach(ThreadSubject *c)
	{
		std::lock_guard<std::mutex> lock (m_mutex);
		m_updates.erase(c);
	}

	/**
	 * @brief Update the given ThreadContext
	 */
	void update(ThreadSubject *c)
	{
		std::lock_guard<std::mutex> lock (m_mutex);
		KeySet & toUpdate = m_updates[c].toUpdate;
		// if (toUpdate.size() == 0) return;

		c->notify(toUpdate);
		toUpdate.clear();
	}

	/**
	 * @brief Receive a function to be executed and remember
	 * which keys need a update in the other ThreadContexts.
	 */
	Key post(Post postFkt)
	{
		std::lock_guard<std::mutex> lock (m_mutex);
		Key k = postFkt();
		for (auto & c: m_updates)
		{
			c.second.toUpdate.append(k);
		}
		return k;
	}

	/**
	 * @brief Request that some layer needs to be globally
	 * activated.
	 *
	 * @param cc requests it and already has it updated itself
	 * @param layer to activate for all threads
	 */
	void globalActivate(ThreadSubject *cc, std::shared_ptr<Layer> layer)
	{
		{
			std::lock_guard<std::mutex> lock (m_mutexOnActivate);
			for (auto && f: m_onActivate)
			{
				f.second();
			}
			m_onActivate.clear();
		}

		std::lock_guard<std::mutex> lock (m_mutex);
		for (auto & c: m_updates)
		{
			 // caller itself has it already activated
			if (cc == c.first) continue;
			c.second.toActivate.push_back(layer);
		}
	}

	void globalDeactivate(ThreadSubject *cc, std::shared_ptr<Layer> layer)
	{
		{
			std::lock_guard<std::mutex> lock (m_mutexOnDeactivate);
			for (auto && f: m_onDeactivate)
			{
				f.second();
			}
			m_onDeactivate.clear();
		}

		std::lock_guard<std::mutex> lock (m_mutex);
		for (auto & c: m_updates)
		{
			 // caller itself has it already deactivated
			if (cc == c.first) continue;
			c.second.toDeactivate.push_back(layer);
		}
	}

	/**
	 * @param cc requester of its updates
	 *
	 * @see globalActivate
	 * @return all layers for that subject
	 */
	LayerVector fetchGlobalActivation(ThreadSubject *cc)
	{
		std::lock_guard<std::mutex> lock (m_mutex);
		LayerVector ret;
		ret.swap(m_updates[cc].toActivate);
		return std::move(ret);
	}

	LayerVector fetchGlobalDeactivation(ThreadSubject *cc)
	{
		std::lock_guard<std::mutex> lock (m_mutex);
		LayerVector ret;
		ret.swap(m_updates[cc].toDeactivate);<commit_msg>allow arbitrary many callbacks to be registered<commit_after>#ifndef ELEKTRA_KDBTHREAD_HPP
#define ELEKTRA_KDBTHREAD_HPP

#include <kdbcontext.hpp>

#include <kdb.hpp>

#include <mutex>
#include <thread>
#include <vector>
#include <cassert>
#include <algorithm>
#include <functional>
#include <unordered_map>

namespace kdb
{

/// Subject from Observer pattern for ThreadContext
class ThreadSubject
{
public:
	virtual void notify(KeySet &ks) = 0;
};

/// A vector of layers
typedef std::vector<std::shared_ptr<Layer>> LayerVector;

typedef std::unordered_map<std::string, std::vector<std::function<void()>>> FunctionMap;

/// A data structure that is stored by context inside the Coordinator
struct PerContext
{
	KeySet toUpdate;
	LayerVector toActivate;
	LayerVector toDeactivate;
};

/**
 * @brief Thread safe coordination of ThreadContext per Threads.
 */
class Coordinator
{
public:
	void onLayerActivation(std::string layerid, std::function <void()> f)
	{
		std::lock_guard<std::mutex> lock (m_mutexOnActivate);
		m_onActivate[layerid].push_back(f);
	}

	void onLayerDeactivation(std::string layerid, std::function <void()> f)
	{
		std::lock_guard<std::mutex> lock (m_mutexOnDeactivate);
		m_onDeactivate[layerid].push_back(f);
	}

private:
	friend class ThreadContext;

	void attach(ThreadSubject *c)
	{
		std::lock_guard<std::mutex> lock (m_mutex);
		m_updates.insert(std::make_pair(c, PerContext()));
	}

	void detach(ThreadSubject *c)
	{
		std::lock_guard<std::mutex> lock (m_mutex);
		m_updates.erase(c);
	}

	/**
	 * @brief Update the given ThreadContext
	 */
	void update(ThreadSubject *c)
	{
		std::lock_guard<std::mutex> lock (m_mutex);
		KeySet & toUpdate = m_updates[c].toUpdate;
		// if (toUpdate.size() == 0) return;

		c->notify(toUpdate);
		toUpdate.clear();
	}

	/**
	 * @brief Receive a function to be executed and remember
	 * which keys need a update in the other ThreadContexts.
	 */
	Key post(Post postFkt)
	{
		std::lock_guard<std::mutex> lock (m_mutex);
		Key k = postFkt();
		for (auto & c: m_updates)
		{
			c.second.toUpdate.append(k);
		}
		return k;
	}

	void runOnActivate(std::shared_ptr<Layer> layer)
	{
		std::lock_guard<std::mutex> lock (m_mutexOnActivate);
		for (auto && f: m_onActivate[layer->id()])
		{
			f();
		}
		m_onActivate.clear();
	}

	/**
	 * @brief Request that some layer needs to be globally
	 * activated.
	 *
	 * @param cc requests it and already has it updated itself
	 * @param layer to activate for all threads
	 */
	void globalActivate(ThreadSubject *cc, std::shared_ptr<Layer> layer)
	{
		runOnActivate(layer);

		std::lock_guard<std::mutex> lock (m_mutex);
		for (auto & c: m_updates)
		{
			 // caller itself has it already activated
			if (cc == c.first) continue;
			c.second.toActivate.push_back(layer);
		}
	}

	void runOnDeactivate(std::shared_ptr<Layer> layer)
	{
		std::lock_guard<std::mutex> lock (m_mutexOnDeactivate);
		for (auto && f: m_onDeactivate[layer->id()])
		{
			f();
		}
		m_onDeactivate.clear();
	}


	void globalDeactivate(ThreadSubject *cc, std::shared_ptr<Layer> layer)
	{
		runOnDeactivate(layer);

		std::lock_guard<std::mutex> lock (m_mutex);
		for (auto & c: m_updates)
		{
			 // caller itself has it already deactivated
			if (cc == c.first) continue;
			c.second.toDeactivate.push_back(layer);
		}
	}

	/**
	 * @param cc requester of its updates
	 *
	 * @see globalActivate
	 * @return all layers for that subject
	 */
	LayerVector fetchGlobalActivation(ThreadSubject *cc)
	{
		std::lock_guard<std::mutex> lock (m_mutex);
		LayerVector ret;
		ret.swap(m_updates[cc].toActivate);
		return std::move(ret);
	}

	LayerVector fetchGlobalDeactivation(ThreadSubject *cc)
	{
		std::lock_guard<std::mutex> lock (m_mutex);
		LayerVector ret;
		ret.swap(m_updates[cc].toDeactivate);
{'fraction_non_alphanumeric': 0.13563029805851792, 'fraction_numerical': 0.0005468963631391851, 'mean_word_length': 4.46711509715994, 'pattern_counts': {'":': 0, '<': 64, '<?xml version=': 0, '>': 68, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6873342', 'n_tokens_mistral': 2813, 'n_tokens_neox': 2575, 'n_words': 609}