<commit_before>#ifndef SILICIUM_THREAD_HPP
#define SILICIUM_THREAD_HPP

#include <silicium/observer.hpp>
#include <silicium/override.hpp>
#include <silicium/config.hpp>
#include <silicium/exchange.hpp>
#include <silicium/yield_context.hpp>
#include <future>
#include <boost/thread/future.hpp>
#include <boost/optional.hpp>
#include <boost/concept_check.hpp>

namespace Si
{
	template <class Element, class ThreadingAPI>
	struct thread_observable
	{
		typedef Element element_type;

		thread_observable()
		{
		}

		template <class Action>
		explicit thread_observable(Action &&action)
			: state(make_unique<movable_state>(std::forward<Action>(action)))
		{
		}

		void wait()
		{
			assert(state);
			state->thread.get();
		}

		void async_get_one(observer<element_type> &receiver)
		{
			assert(state);
			typename ThreadingAPI::unique_lock lock(state->result_mutex);
			assert(!state->receiver_);
			state->receiver_ = &receiver;
			if (!state->cached_result)
			{
				if (!state->thread.valid())
				{
					lock.unlock();
					return Si::exchange(state->receiver_, nullptr)->ended();
				}
				return;
			}
			auto ready_result = std::move(*state->cached_result);
			state->result_retrieved.notify_one();
			lock.unlock();
			Si::exchange(state->receiver_, nullptr)->got_element(std::move(ready_result));
		}

		void cancel()
		{
			throw std::logic_error("to do");
		}

	private:

		struct movable_state : private detail::yield_context_impl<Element>
		{
			typename ThreadingAPI::template future<void> thread;
			typename ThreadingAPI::mutex result_mutex;
			typename ThreadingAPI::condition_variable result_retrieved;
			observer<element_type> *receiver_ = nullptr;
			boost::optional<Element> cached_result;

			template <class Action>
			explicit movable_state(Action &&action)
			{
				//start the background thread after the initialization of all the members
				thread = ThreadingAPI::template launch_async([this, action]() -> void
				{
					yield_context<Element> yield(*this);
					action(yield);
				});
			}

			virtual void push_result(Element result) SILICIUM_OVERRIDE
			{
				typename ThreadingAPI::unique_lock lock(result_mutex);
				while (cached_result.is_initialized())
				{
					result_retrieved.wait(lock);
				}
				if (receiver_)
				{
					auto receiver = Si::exchange(receiver_, nullptr);
					lock.unlock();
					receiver->got_element(std::move(result));
				}
				else
				{
					cached_result = std::move(result);
				}
			}

			virtual void get_one(observable<nothing> &target) SILICIUM_OVERRIDE
			{
				boost::ignore_unused_variable_warning(target);
				throw std::logic_error("todo");
			}
		};

		std::unique_ptr<movable_state> state;
	};

	struct std_threading
	{
		template <class T>
		using future = std::future<T>;
		using mutex = std::mutex;
		using condition_variable = std::condition_variable;
		using unique_lock = std::unique_lock<std::mutex>;
		template <class Action, class ...Args>
		static auto launch_async(Action &&action, Args &&...args) -> std::future<decltype(action(std::forward<Args>(args)...))>
		{
			return std::async(std::launch::async, std::forward<Action>(action), std::forward<Args>(args)...);
		}
	};

	struct boost_threading
	{
		template <class T>
		using future = boost::unique_future<T>;
		using mutex = boost::mutex;
		using condition_variable = boost::condition_variable;
		using unique_lock = boost::unique_lock<boost::mutex>;
		template <class Action, class ...Args>
		static auto launch_async(Action &&action, Args &&...args) -> boost::unique_future<decltype(action(std::forward<Args>(args)...))>
		{
			return boost::async(boost::launch::async, std::forward<Action>(action), std::forward<Args>(args)...);
		}
	};

	template <class Element, class ThreadingAPI, class Action>
	auto make_thread(Action &&action) -> thread_observable<Element, ThreadingAPI>
	{
		return thread_observable<Element, ThreadingAPI>(std::forward<Action>(action));
	}
}

#endif<commit_msg>finish make_thread implementation<commit_after>#ifndef SILICIUM_THREAD_HPP
#define SILICIUM_THREAD_HPP

#include <silicium/observer.hpp>
#include <silicium/override.hpp>
#include <silicium/config.hpp>
#include <silicium/exchange.hpp>
#include <silicium/yield_context.hpp>
#include <future>
#include <boost/thread/future.hpp>
#include <boost/optional.hpp>
#include <boost/concept_check.hpp>

namespace Si
{
	namespace detail
	{
		template <class Element>
		struct thread_yield_context_impl : yield_context_impl<Element>
		{
			virtual void push_result(Element result) SILICIUM_OVERRIDE
			{
				throw std::logic_error("to do");
			}

			virtual void get_one(observable<nothing> &target) SILICIUM_OVERRIDE
			{
				throw std::logic_error("to do");
			}
		};
	}

	template <class Element, class ThreadingAPI>
	struct thread_observable
	{
		typedef Element element_type;

		thread_observable()
		{
		}

		template <class Action>
		explicit thread_observable(Action &&action)
			: state(Si::make_unique<state_type>(std::forward<Action>(action)))
		{
		}

		void wait()
		{
			return state->wait();
		}

		void async_get_one(observer<element_type> &receiver)
		{
			return state->async_get_one(receiver);
		}

		void cancel()
		{
			return state->cancel();
		}

	private:

		struct state_type
			: private detail::yield_context_impl<Element>
			, private observer<nothing>
		{
			template <class Action>
			explicit state_type(Action &&action)
			{
				worker = ThreadingAPI::launch_async([&]
				{
					yield_context<Element> yield(*this);
					std::forward<Action>(action)(yield);

					typename ThreadingAPI::unique_lock lock(receiver_mutex);
					if (receiver)
					{
						Si::exchange(receiver, nullptr)->ended();
					}
					else
					{
						has_ended = true;
					}
				});
			}

			void async_get_one(observer<Element> &new_receiver)
			{
				typename ThreadingAPI::unique_lock lock(receiver_mutex);
				assert(!receiver);
				if (ready_result)
				{
					Element result = std::move(*ready_result);
					ready_result = boost::none;
					receiver_ready.notify_one();
					lock.unlock();
					new_receiver.got_element(std::move(result));
				}
				else
				{
					if (has_ended)
					{
						return new_receiver.ended();
					}
					receiver = &new_receiver;
					receiver_ready.notify_one();
				}
			}

			void cancel()
			{
				throw std::logic_error("to do");
			}

			void wait()
			{
				worker.get();
			}

		private:

			typename ThreadingAPI::template future<void> worker;
			typename ThreadingAPI::mutex receiver_mutex;
			observer<Element> *receiver = nullptr;
			boost::optional<Element> ready_result;
			typename ThreadingAPI::condition_variable receiver_ready;
			bool has_ended = false;

			typename ThreadingAPI::mutex got_something_mutex;
			typename ThreadingAPI::condition_variable got_something_set;
			bool got_something = false;

			virtual void push_result(Element result) SILICIUM_OVERRIDE
			{
				typename ThreadingAPI::unique_lock lock(receiver_mutex);
				while (ready_result && !receiver)
				{
					receiver_ready.wait(lock);
				}
				if (receiver)
				{
					auto * const receiver_ = Si::exchange(receiver, nullptr);
					lock.unlock();
					receiver_->got_element(std::move(result));
				}
				else
				{
					ready_result = std::move(result);
				}
			}

			virtual void get_one(observable<nothing> &target) SILICIUM_OVERRIDE
			{
				assert(![this]() { typename ThreadingAPI::unique_lock lock(got_something_mutex); return got_something; }());
				target.async_get_one(*this);

				typename ThreadingAPI::unique_lock lock(got_something_mutex);
				while (!got_something)
				{
					got_something_set.wait(lock);
				}
				got_something = false;
			}

			virtual void got_element(nothing) SILICIUM_OVERRIDE
			{
				wake_get_one();
			}

			virtual void ended() SILICIUM_OVERRIDE
			{
				wake_get_one();
			}

			void wake_get_one()
			{
				typename ThreadingAPI::unique_lock lock(got_something_mutex);
				got_something = true;
				got_something_set.notify_one();
			}
		};

		std::unique_ptr<state_type> state;
	};

	struct std_threading
	{
		template <class T>
		using future = std::future<T>;
		template <class T>
		using promise = std::promise<T>;
		using mutex = std::mutex;
		using condition_variable = std::condition_variable;
		using unique_lock = std::unique_lock<std::mutex>;
		template <class Action, class ...Args>
		static auto launch_async(Action &&action, Args &&...args) -> std::future<decltype(action(std::forward<Args>(args)...))>
		{
			return std::async(std::launch::async, std::forward<Action>(action), std::forward<Args>(args)...);
		}
	};

	struct boost_threading
	{
		template <class T>
		using future = boost::unique_future<T>;
		template <class T>
		using promise = boost::promise<T>;
		using mutex = boost::mutex;
		using condition_variable = boost::condition_variable;
		using unique_lock = boost::unique_lock<boost::mutex>;
		template <class Action, class ...Args>
		static auto launch_async(Action &&action, Args &&...args) -> boost::unique_future<decltype(action(std::forward<Args>(args)...))>
		{
			return boost::async(boost::launch::async, std::forward<Action>(action), std::forward<Args>(args)...);
		}
	};

	template <class Element, class ThreadingAPI, class Action>
	auto make_thread(Action &&action) -> thread_observable<Element, ThreadingAPI>
	{
		return thread_observable<Element, ThreadingAPI>(std::forward<Action>(action));
	}
}

#endif
{'fraction_non_alphanumeric': 0.13288432756040197, 'fraction_numerical': 0.0, 'mean_word_length': 4.249719416386083, 'pattern_counts': {'":': 0, '<': 97, '<?xml version=': 0, '>': 122, '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': '468263', 'n_tokens_mistral': 3695, 'n_tokens_neox': 3015, 'n_words': 563}