text: ﻿#pragma once
  
  namespace resumef
  {
  namespace detail
  {
  	template<class _Ty, class _Chty>
  	struct state_channel_t : public state_base_t
  	{
  		using value_type = _Ty;
  
  		state_channel_t(_Chty* ch, value_type& val) noexcept
  			: _channel(ch->shared_from_this())
  			, _value(std::addressof(val))
  		{
  		}
  
  		virtual void resume() override
  		{
  			coroutine_handle<> handler = _coro;
  			if (handler)
  			{
  				_coro = nullptr;
  				_scheduler->del_final(this);
  				handler.resume();
  			}
  		}
  
  		virtual bool has_handler() const  noexcept override
  		{
  			return (bool)_coro;
  		}
  
  		void on_notify()
  		{
  			assert(this->_scheduler != nullptr);
  			if (this->_coro)
  				this->_scheduler->add_generator(this);
  		}
  
  		void on_cancel() noexcept
  		{
  			this->_coro = nullptr;
  		}
  
  		template<class _PromiseT, typename = std::enable_if_t<traits::is_promise_v<_PromiseT>>>
  		void on_await_suspend(coroutine_handle<_PromiseT> handler) noexcept
  		{
  			_PromiseT& promise = handler.promise();
  			auto* parent_state = promise.get_state();
  			scheduler_t* sch = parent_state->get_scheduler();
  
  			this->_scheduler = sch;
  			this->_coro = handler;
  		}
  
  		friend _Chty;
  	public:
  		//为浸入式单向链表提供的next指针
  		state_channel_t* _next = nullptr;
  	protected:
  		//co_await产生的临时awaitor会引用state，管理state的生命周期
  		//state再引用channel
  		//从而在co_await channel.xxx()这条语句内，awaitor/state/channel均处于正确的生命周期
  		std::shared_ptr<_Chty> _channel;
  	protected:
  		value_type* _value;
  	};
  
  
  
  //-----------------------------------------------------------------------------------------------------------------------------------------
  
  	template<class _Ty, bool _Optional>
  	struct channel_impl_v2 : public std::enable_shared_from_this<channel_impl_v2<_Ty, _Optional>>
  	{
  		using value_type = _Ty;
  		using optional_type = std::conditional_t<_Optional, std::optional<value_type>, value_type>;
  		using this_type = channel_impl_v2<value_type, _Optional>;
  
  		using state_read_t = state_channel_t<optional_type, this_type>;
  		using state_write_t = state_channel_t<value_type, this_type>;
  
  		channel_impl_v2(size_t cache_size);
  
  		bool try_read(optional_type& val);
  		bool try_read_nolock(optional_type& val);
  		void add_read_list_nolock(state_read_t* state);
  
  		bool try_write(value_type& val);
  		bool try_write_nolock(value_type& val);
  		void add_write_list_nolock(state_write_t* state);
  
  		size_t capacity() const noexcept
  		{
  			return _max_counter;
  		}
  	private:
  		auto try_pop_reader_()->state_read_t*;
  		auto try_pop_writer_()->state_write_t*;
  		void awake_one_reader_();
  		bool awake_one_reader_(value_type& val);
  
  		void awake_one_writer_();
  		bool awake_one_writer_(optional_type& val);
  
  		channel_impl_v2(const channel_impl_v2&) = delete;
  		channel_impl_v2(channel_impl_v2&&) = delete;
  		channel_impl_v2& operator = (const channel_impl_v2&) = delete;
  		channel_impl_v2& operator = (channel_impl_v2&&) = delete;
  
  		static constexpr bool USE_SPINLOCK = true;
  		static constexpr bool USE_RING_QUEUE = true;
  		static constexpr bool USE_LINK_QUEUE = true;
  
  		using queue_type = std::conditional_t<USE_RING_QUEUE, ring_queue<value_type, _Optional, uint32_t>, std::deque<value_type>>;
  		using read_queue_type = std::conditional_t<USE_LINK_QUEUE, intrusive_link_queue<state_read_t>, std::list<state_read_t*>>;
  		using write_queue_type = std::conditional_t<USE_LINK_QUEUE, intrusive_link_queue<state_write_t>, std::list<state_write_t*>>;
  
  		const size_t _max_counter;							//数据队列的容量上限
  	public:
  		using lock_type = std::conditional_t<USE_SPINLOCK, spinlock, std::deque<std::recursive_mutex>>;
  		lock_type _lock;									//保证访问本对象是线程安全的
  	private:
  		queue_type _values;									//数据队列
  		read_queue_type _read_awakes;						//读队列
  		write_queue_type _write_awakes;						//写队列
  	};
  
  
  
  //-----------------------------------------------------------------------------------------------------------------------------------------
  
  	template<class _Ty, bool _Optional>
  	channel_impl_v2<_Ty, _Optional>::channel_impl_v2(size_t cache_size)
  		: _max_counter(cache_size)
  		, _values(USE_RING_QUEUE ? cache_size : 0)
  	{
  	}
  
  	template<class _Ty, bool _Optional>
  	inline bool channel_impl_v2<_Ty, _Optional>::try_read(optional_type& val)
  	{
  		scoped_lock<lock_type> lock_(this->_lock);
  		return try_read_nolock(val);
  	}
  
  	template<class _Ty, bool _Optional>
  	bool channel_impl_v2<_Ty, _Optional>::try_read_nolock(optional_type& val)
  	{
  		if constexpr (USE_RING_QUEUE)
  		{
  			if (_values.try_pop(val))
  			{
  				awake_one_writer_();
  				return true;
  			}
  		}
  		else
  		{
  			if (_values.size() > 0)
  			{
  				val = std::move(_values.front());
  				_values.pop_front();
  				awake_one_writer_();
  
  				return true;
  			}
  		}
  
  		return awake_one_writer_(val);
  	}
  
  	template<class _Ty, bool _Optional>
  	inline void channel_impl_v2<_Ty, _Optional>::add_read_list_nolock(state_read_t* state)
  	{
  		assert(state != nullptr);
  		_read_awakes.push_back(state);
  	}
  
  	template<class _Ty, bool _Optional>
  	inline bool channel_impl_v2<_Ty, _Optional>::try_write(value_type& val)
  	{
  		scoped_lock<lock_type> lock_(this->_lock);
  		return try_write_nolock(val);
  	}
  
  	template<class _Ty, bool _Optional>
  	bool channel_impl_v2<_Ty, _Optional>::try_write_nolock(value_type& val)
  	{
  		if constexpr (USE_RING_QUEUE)
  		{
  			if (_values.try_push(std::move(val)))
  			{
  				awake_one_reader_();
  				return true;
  			}
  		}
  		else
  		{
  			if (_values.size() < _max_counter)
  			{
  				_values.push_back(std::move(val));
  				awake_one_reader_();
  
  				return true;
  			}
  		}
  
  		return awake_one_reader_(val);
  	}
  
  	template<class _Ty, bool _Optional>
  	inline void channel_impl_v2<_Ty, _Optional>::add_write_list_nolock(state_write_t* state)
  	{
  		assert(state != nullptr);
  		_write_awakes.push_back(state);
  	}
  
  	template<class _Ty, bool _Optional>
  	auto channel_impl_v2<_Ty, _Optional>::try_pop_reader_()->state_read_t*
  	{
  		if constexpr (USE_LINK_QUEUE)
  		{
  			return _read_awakes.try_pop();
  		}
  		else
  		{
  			if (!_read_awakes.empty())
  			{
  				state_read_t* state = _read_awakes.front();
  				_read_awakes.pop_front();
  				return state;
  			}
  			return nullptr;
  		}
  	}
  
  	template<class _Ty, bool _Optional>
  	auto channel_impl_v2<_Ty, _Optional>::try_pop_writer_()->state_write_t*
  	{
  		if constexpr (USE_LINK_QUEUE)
  		{
  			return _write_awakes.try_pop();
  		}
  		else
  		{
  			if (!_write_awakes.empty())
  			{
  				state_write_t* state = _write_awakes.front();
  				_write_awakes.pop_front();
  				return state;
  			}
  			return nullptr;
  		}
  	}
  
  	template<class _Ty, bool _Optional>
  	void channel_impl_v2<_Ty, _Optional>::awake_one_reader_()
  	{
  		state_read_t* state = try_pop_reader_();
  		if (state != nullptr)
  		{
  			assert(!_values.empty());
  
  			if constexpr (USE_RING_QUEUE)
  			{
  				_values.try_pop(*state->_value);
  			}
  			else
  			{
  				*state->_value = std::move(_values.front());
  				_values.pop_front();
  			}
  
  			state->on_notify();
  		}
  	}
  
  	template<class _Ty, bool _Optional>
  	bool channel_impl_v2<_Ty, _Optional>::awake_one_reader_(value_type& val)
  	{
  		state_read_t* state = try_pop_reader_();
  		if (state != nullptr)
  		{
  			*state->_value = std::move(val);
  
  			state->on_notify();
  			return true;
  		}
  		return false;
  	}
  
  	template<class _Ty, bool _Optional>
  	void channel_impl_v2<_Ty, _Optional>::awake_one_writer_()
  	{
  		state_write_t* state = try_pop_writer_();
  		if (state != nullptr)
  		{
  			if constexpr (USE_RING_QUEUE)
  			{
  				bool ret = _values.try_push(std::move(*state->_value));
  				(void)ret;
  				assert(ret);
  			}
  			else
  			{
  				assert(_values.size() < _max_counter);
  				_values.push_back(std::move(*state->_value));
  			}
  
  			state->on_notify();
  		}
  	}
  
  	template<class _Ty, bool _Optional>
  	bool channel_impl_v2<_Ty, _Optional>::awake_one_writer_(optional_type& val)
  	{
  		state_write_t* writer = try_pop_writer_();
  		if (writer != nullptr)
  		{
  			val = std::move(*writer->_value);
  
  			writer->on_notify();
  			return true;
  		}
  		return false;
  	}
  }	//namespace detail
  
  
  
  
  //-----------------------------------------------------------------------------------------------------------------------------------------
  
  	template<class _Ty, bool _Optional, bool _OptimizationThread>
  	struct [[nodiscard]] channel_t<_Ty, _Optional, _OptimizationThread>::read_awaiter
  	{
  		using state_type = typename channel_type::state_read_t;
  
  		read_awaiter(channel_type* ch) noexcept
  			: _channel(ch)
  			, _value()
  		{}
  
  		~read_awaiter()
  		{//为了不在协程中也能正常使用
  			if (_channel != nullptr)
  			{
  				while(!_channel->try_read(_value));
  			}
  		}
  
  		bool await_ready()
  		{
  			//在多线程竞争较为多的时候，先检查是否可用，可以稍微提高点效率
  			if constexpr (optimization_for_multithreading)
  			{
  				scoped_lock<lock_type> lock_(_channel->_lock);
  
  				if (_channel->try_read_nolock(_value))
  				{
  					_channel = nullptr;
  					return true;
  				}
  			}
  			return false;
  		}
  		template<class _PromiseT, typename = std::enable_if_t<traits::is_promise_v<_PromiseT>>>
  		bool await_suspend(coroutine_handle<_PromiseT> handler)
  		{
  			scoped_lock<lock_type> lock_(_channel->_lock);
  
  			if (_channel->try_read_nolock(_value))
  			{
  				_channel = nullptr;
  				return false;
  			}
  
  			_state = new state_type(_channel, _value);
  			_state->on_await_suspend(handler);
  			_channel->add_read_list_nolock(_state.get());
  			_channel = nullptr;
  
  			return true;
  		}
  		value_type await_resume()
  		{
  			if constexpr (use_optional)
  				return std::move(_value).value();
  			else
  				return std::move(_value);
  		}
  	private:
  		channel_type* _channel;
  		counted_ptr<state_type> _state;	//延迟到await_suspend()里创建，减小不必要的内存申请
  		mutable optional_type _value;
  	};
  
  	template<class _Ty, bool _Optional, bool _OptimizationThread>
  	struct [[nodiscard]] channel_t<_Ty, _Optional, _OptimizationThread>::write_awaiter
  	{
  		using state_type = typename channel_type::state_write_t;
  
  		template<class U>
  		write_awaiter(channel_type* ch, U&& val) noexcept(std::is_move_constructible_v<value_type>)
  			: _channel(ch)
  			, _value(std::forward<U>(val))
  		{}
  
  		~write_awaiter()
  		{//为了不在协程中也能正常使用
  			if (_channel != nullptr)
  			{
  				while(!_channel->try_write(_value));
  			}
  		}
  
  		bool await_ready()
  		{
  			//在多线程竞争较为多的时候，先检查是否可用，可以稍微提高点效率
  			if constexpr (optimization_for_multithreading)
  			{
  				scoped_lock<lock_type> lock_(_channel->_lock);
  
  				if (_channel->try_write_nolock(_value))
  				{
  					_channel = nullptr;
  					return true;
  				}
  			}
  			return false;
  		}
  		template<class _PromiseT, typename = std::enable_if_t<traits::is_promise_v<_PromiseT>>>
  		bool await_suspend(coroutine_handle<_PromiseT> handler)
  		{
  			scoped_lock<lock_type> lock_(_channel->_lock);
  
  			if (_channel->try_write_nolock(_value))
  			{
  				_channel = nullptr;
  				return false;
  			}
  
  			_state = new state_type(_channel, _value);
  			_state->on_await_suspend(handler);
  			_channel->add_write_list_nolock(_state.get());
  			_channel = nullptr;
  
  			return true;
  		}
  		void await_resume()
  		{
  		}
  	private:
  		channel_type* _channel;
  		counted_ptr<state_type> _state;	//延迟到await_suspend()里创建，减小不必要的内存申请
  		mutable value_type _value;
  	};
  
  	template<class _Ty, bool _Optional, bool _OptimizationThread>
  	channel_t<_Ty, _Optional, _OptimizationThread>::channel_t(size_t cache_size)
  		:_chan(std::make_shared<channel_type>(cache_size))
  	{
  
  	}
  
  	template<class _Ty, bool _Optional, bool _OptimizationThread>
  	size_t channel_t<_Ty, _Optional, _OptimizationThread>::capacity() const noexcept
  	{
  		return _chan->capacity();
  	}
  
  	template<class _Ty, bool _Optional, bool _OptimizationThread>
  	typename channel_t<_Ty, _Optional, _OptimizationThread>::read_awaiter
  		channel_t<_Ty, _Optional, _OptimizationThread>::operator co_await() const noexcept
  	{
  		return { _chan.get() };
  	}
  
  	template<class _Ty, bool _Optional, bool _OptimizationThread>
  	typename channel_t<_Ty, _Optional, _OptimizationThread>::read_awaiter
  		channel_t<_Ty, _Optional, _OptimizationThread>::read() const noexcept
  	{
  		return { _chan.get() };
  	}
  
  	template<class _Ty, bool _Optional, bool _OptimizationThread>
  	template<class U> requires(std::is_constructible_v<_Ty, U&&>)
  	typename channel_t<_Ty, _Optional, _OptimizationThread>::write_awaiter
  		channel_t<_Ty, _Optional, _OptimizationThread>::write(U&& val) const noexcept(std::is_nothrow_move_constructible_v<U>)
  	{
  		return write_awaiter{ _chan.get(), std::forward<U>(val) };
  	}
  
  	template<class _Ty, bool _Optional, bool _OptimizationThread>
  	template<class U> requires(std::is_constructible_v<_Ty, U&&>)
  	typename channel_t<_Ty, _Optional, _OptimizationThread>::write_awaiter
  		channel_t<_Ty, _Optional, _OptimizationThread>::operator << (U&& val) const noexcept(std::is_nothrow_move_constructible_v<U>)
  	{
  		return write_awaiter{ _chan.get(), std::forward<U>(val) };
  	}
  
  }	//namespace resumef
  