<commit_before>#include <atomic>
#include <functional>
#include <complex> 

#include "threadpool.hpp"
namespace Execution 
{
StateMachineThreadPool::StateMachineThreadPool() :_stop(true), _sharedConditionVar(new std::condition_variable()), _stateMachines(_sharedConditionVar) {}

void StateMachineThreadPool::stopPool()
{
	_stop = true;
	_stateMachines.stopQueue();

}

void StateMachineThreadPool::startPool(int n)
{
	_stateMachines.startQueue();
	_stop = false;
	modifiedThreads(n);

}

void StateMachineThreadPool::stopUponCompletion(std::atomic_int* messages)
{
	std::unique_lock<std::mutex> lock(stop_request_mu);

	if (!((*messages) == 0 && (*worker_threads) == 0))
	{
		stop_request_cond->wait(lock, [this, messages] {return (*messages) == 0 && *(this->worker_threads) == 0; });
	}

	stopPool();

}

void StateMachineThreadPool::task()
{
	while (!this->_stop && !workers.isReadyToStop(std::this_thread::get_id()))
	{



		ES::StateMachineRef sm = nullptr;
		while (sm == nullptr && !this->_stop)
		{

			if (workers.isReadyToStop(std::this_thread::get_id()))
			{
				return;
			}

			_stateMachines.dequeue(sm);
			
		}


		if (sm != nullptr)
		{
			incrementWorkers();

			if (!sm->isInitialized()) sm->init();
			for (int i = 0; i < 5 && sm != nullptr && !sm->emptyMessageQueue(); ++i)
			{
				if (!sm->isDestoryed()) {
					sm->processEventVirtual();
				}
				else {
					sm->setPooled(false);
					delete sm;
					sm = nullptr;
				}
			}

			if (sm != nullptr) {

				if (!sm->emptyMessageQueue())
				{
					_stateMachines.enqueue(sm);
				}
				else
				{
					sm->setPooled(false);
				}
			}

			reduceWorkers();
			stop_request_cond->notify_one();
		}

	}

}

void StateMachineThreadPool::modifiedThreads(int n)
{
	if (!_stop)
	{
		std::unique_lock<std::mutex> mlock(modifie_mutex);

		workers.setExpectedThreads(n);
		if (workers.isTooManyWorkers())
		{
			workers.gettingThreadsReadyToStop(_sharedConditionVar);
		}
		while (workers.isTooFewWorkers())
		{
			workers.addThread(new std::thread(&StateMachineThreadPool::task, this));
		}
	}

}


void StateMachineThreadPool::incrementWorkers()<commit_msg>Thread delete and creation is execatued before a thread gets a new job<commit_after>#include <atomic>
#include <functional>
#include <complex> 

#include "threadpool.hpp"
namespace Execution 
{
StateMachineThreadPool::StateMachineThreadPool() :_stop(true), _sharedConditionVar(new std::condition_variable()), _stateMachines(_sharedConditionVar) {}

void StateMachineThreadPool::stopPool()
{
	_stop = true;
	_stateMachines.stopQueue();

}

void StateMachineThreadPool::startPool(int n)
{
	_stateMachines.startQueue();
	_stop = false;
	modifiedThreads(n);
	for(int i = 0; i < n; ++i)
	{
		workers.addThread(new std::thread(&StateMachineThreadPool::task, this));
	}

}

void StateMachineThreadPool::stopUponCompletion(std::atomic_int* messages)
{
	std::unique_lock<std::mutex> lock(stop_request_mu);

	if (!((*messages) == 0 && (*worker_threads) == 0))
	{
		stop_request_cond->wait(lock, [this, messages] {return (*messages) == 0 && *(this->worker_threads) == 0; });
	}

	stopPool();

}

void StateMachineThreadPool::task()
{
	while (!this->_stop && !workers.isReadyToStop(std::this_thread::get_id()))
	{
		if (modifie_mutex.try_lock()) {
			if (workers.isTooManyWorkers())
			{
				workers.gettingThreadsReadyToStop(_sharedConditionVar);
			}
			while (workers.isTooFewWorkers())
			{
				workers.addThread(new std::thread(&StateMachineThreadPool::task, this));
			}
			modifie_mutex.unlock();
		}

		ES::StateMachineRef sm = nullptr;
		while (sm == nullptr && !this->_stop)
		{

			if (workers.isReadyToStop(std::this_thread::get_id()))
			{
				return;
			}

			_stateMachines.dequeue(sm);
			
		}


		if (sm != nullptr)
		{
			incrementWorkers();

			if (!sm->isInitialized()) sm->init();
			for (int i = 0; i < 5 && sm != nullptr && !sm->emptyMessageQueue(); ++i)
			{
				if (!sm->isDestoryed()) {
					sm->processEventVirtual();
				}
				else {
					sm->setPooled(false);
					delete sm;
					sm = nullptr;
				}
			}

			if (sm != nullptr) {

				if (!sm->emptyMessageQueue())
				{
					_stateMachines.enqueue(sm);
				}
				else
				{
					sm->setPooled(false);
				}
			}

			reduceWorkers();
			stop_request_cond->notify_one();
		}

	}

}

void StateMachineThreadPool::modifiedThreads(int n)
{
	if (!_stop)
	{
		workers.setExpectedThreads(n);
	}
}


void StateMachineThreadPool::incrementWorkers()
{'fraction_non_alphanumeric': 0.161101083032491, 'fraction_numerical': 0.0029332129963898917, 'mean_word_length': 4.215294117647058, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 38, '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': '4263659', 'n_tokens_mistral': 1820, 'n_tokens_neox': 1538, 'n_words': 253}