<commit_before>/*
 * Copyright 2017, <NAME>
 *
 * This is PUBLIC DOMAIN software so use at your own risk as it comes
 * with no warranties. This code is yours to share, use and modify without
 * any restrictions or obligations.
 *
 * For more information see conwrap/LICENSE or refer refer to http://unlicense.org
 *
 * Author: gimesketvirtadieni at gmail dot com (<NAME>)
 */

#pragma once

#include <conwrap/ProcessorProxy.hpp>
#include <functional>
#include <memory>
#include <thread>

#include "slim/ContainerBase.hpp"
#include "slim/log/log.hpp"


namespace slim
{
	template <class ProducerType, class ConsumerType>
	class Scheduler
	{
		public:
			Scheduler(std::unique_ptr<ProducerType> pr, std::unique_ptr<ConsumerType> cn)
			: producerPtr{std::move(pr)}
			, consumerPtr{std::move(cn)}
			{
				LOG(DEBUG) << LABELS{"slim"} << "Scheduler object was created (id=" << this << ")";
			}

			// using Rule Of Zero
		   ~Scheduler()
			{
				LOG(DEBUG) << LABELS{"slim"} << "Scheduler object was deleted (id=" << this << ")";
			}

			Scheduler(const Scheduler&) = delete;             // non-copyable
			Scheduler& operator=(const Scheduler&) = delete;  // non-assignable
			Scheduler(Scheduler&& rhs) = delete;              // non-movable
			Scheduler& operator=(Scheduler&& rhs) = delete;   // non-move-assignable

			void setProcessorProxy(conwrap::ProcessorProxy<ContainerBase>* p)
			{
				processorProxyPtr = p;

				producerPtr->setProcessorProxy(p);
				consumerPtr->setProcessorProxy(p);
			}

			void start()
			{
				producerPtr->start();
				consumerPtr->start();

				// starting a single consumer thread for processing PCM data
				processingThread = std::thread([&]
				{
					LOG(DEBUG) << LABELS{"slim"} << "Processing thread was started (id=" << std::this_thread::get_id() << ")";

					// signalling back to the starting thread when processing thread is fully ready
					processingStarted = true;

					while (producerPtr->isRunning())
					{
						if (producerPtr->isAvailable())
						{
							processorProxyPtr->process([&producer = *producerPtr, &consumer = *consumerPtr]
							{
								producer.produce(consumer);
							});
						}
						else
						{
							// TODO: cruise control should be implemented
							// if no PCM data is available in any of the producers then pause processing
							std::this_thread::sleep_for(std::chrono::milliseconds{10});
						}
					}

					LOG(DEBUG) << LABELS{"slim"} << "Streamer thread was stopped (id=" << std::this_thread::get_id() << ")";
				});

				// making sure it is up and running
				while(processingThread.joinable() && !processingStarted)
				{
					std::this_thread::sleep_for(std::chrono::milliseconds{1});
				}
			}

			void stop(bool gracefully = true)
			{
				producerPtr->stop();
				consumerPtr->stop();

				// waiting for consumer's thread to terminate
				if (processingThread.joinable())
				{
					processingThread.join();
				}
			}

		private:
			std::unique_ptr<ProducerType>           producerPtr;
			std::unique_ptr<ConsumerType>           consumerPtr;
			conwrap::ProcessorProxy<ContainerBase>* processorProxyPtr{nullptr};
			std::thread                             processingThread;
			volatile bool                           processingStarted{false};
	};
}<commit_msg>Refactoring Scheduling algorithm<commit_after>/*
 * Copyright 2017, <NAME>
 *
 * This is PUBLIC DOMAIN software so use at your own risk as it comes
 * with no warranties. This code is yours to share, use and modify without
 * any restrictions or obligations.
 *
 * For more information see conwrap/LICENSE or refer refer to http://unlicense.org
 *
 * Author: gimesketvirtadieni at gmail dot com (<NAME>)
 */

#pragma once

#include <atomic>
#include <conwrap/ProcessorProxy.hpp>
#include <functional>
#include <memory>
#include <thread>

#include "slim/ContainerBase.hpp"
#include "slim/log/log.hpp"


namespace slim
{
	template <class ProducerType, class ConsumerType>
	class Scheduler
	{
		public:
			Scheduler(std::unique_ptr<ProducerType> pr, std::unique_ptr<ConsumerType> cn)
			: producerPtr{std::move(pr)}
			, consumerPtr{std::move(cn)}
			, monitorThread
			{[&]{
				// starting a single consumer thread for processing PCM data
				LOG(DEBUG) << LABELS{"slim"} << "Scheduler monitor thread was started (id=" << std::this_thread::get_id() << ")";

				while (!finish)
				{
					std::this_thread::sleep_for(std::chrono::milliseconds{50});

					if (requestTaskLater)
					{
						requestTaskLater = false;

						processorProxyPtr->process([&]
						{
							processTask();
						});
					}
				}

				LOG(DEBUG) << LABELS{"slim"} << "Scheduler monitor thread was stopped (id=" << std::this_thread::get_id() << ")";
			}}
			{
				LOG(DEBUG) << LABELS{"slim"} << "Scheduler object was created (id=" << this << ")";
			}

			// using Rule Of Zero
		   ~Scheduler()
			{
				// signalling monitor thread to stop and joining it
				finish = true;
				if (monitorThread.joinable())
				{
					monitorThread.join();
				}

				LOG(DEBUG) << LABELS{"slim"} << "Scheduler object was deleted (id=" << this << ")";
			}

			Scheduler(const Scheduler&) = delete;             // non-copyable
			Scheduler& operator=(const Scheduler&) = delete;  // non-assignable
			Scheduler(Scheduler&& rhs) = delete;              // non-movable
			Scheduler& operator=(Scheduler&& rhs) = delete;   // non-move-assignable

			void setProcessorProxy(conwrap::ProcessorProxy<ContainerBase>* p)
			{
				processorProxyPtr = p;

				producerPtr->setProcessorProxy(p);
				consumerPtr->setProcessorProxy(p);
			}

			void start()
			{
				producerPtr->start();
				consumerPtr->start();

				// asking monitor thread to submit a new task
				requestTaskLater = true;

				LOG(DEBUG) << LABELS{"slim"} << "Streaming was started";
			}

			void stop(bool gracefully = true)
			{
				producerPtr->stop();
				consumerPtr->stop();

				LOG(DEBUG) << LABELS{"slim"} << "Streaming was stopped";
			}

		protected:
			void processTask()
			{
				if (producerPtr->isRunning())
				{
					if (producerPtr->isAvailable())
					{
						producerPtr->produce(*consumerPtr);

						processorProxyPtr->process([&]
						{
							processTask();
						});
					}
					else
					{
						requestTaskLater = true;
					}
				}
			}

		private:
			std::unique_ptr<ProducerType>           producerPtr;
			std::unique_ptr<ConsumerType>           consumerPtr;
			std::thread                             monitorThread;
			std::atomic<bool>                       finish{false};
			std::atomic<bool>                       requestTaskLater{false};
			conwrap::ProcessorProxy<ContainerBase>* processorProxyPtr{nullptr};
	};
}
{'fraction_non_alphanumeric': 0.1341756919374248, 'fraction_numerical': 0.0019554753309265946, 'mean_word_length': 2.9436536180308424, 'pattern_counts': {'":': 0, '<': 104, '<?xml version=': 0, '>': 52, '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': '4516897', 'n_tokens_mistral': 2547, 'n_tokens_neox': 1984, 'n_words': 479}