Path: csiph.com!eternal-september.org!reader02.eternal-september.org!.POSTED!not-for-mail
From: Bonita Montero <Bonita.Montero@gmail.com>
Newsgroups: comp.programming.threads
Subject: A thread-queue
Date: Wed, 1 Jun 2022 20:42:36 +0200
Organization: A noiseless patient Spider
Lines: 207
Message-ID: <t78bug$laf$1@dont-email.me>
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8; format=flowed
Content-Transfer-Encoding: 7bit
Injection-Date: Wed, 1 Jun 2022 18:42:24 -0000 (UTC)
Injection-Info: reader02.eternal-september.org; posting-host="bb8f0ac5e94ee4d40cf7ef27c23d2aeb"; logging-data="21839"; mail-complaints-to="abuse@eternal-september.org";	posting-account="U2FsdGVkX1/zvdz2v1edzkqsZDwHBMuMeacDmezADSo="
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:91.0) Gecko/20100101 Thunderbird/91.10.0
Cancel-Lock: sha1:esiMp4gSgcx1zvhtembo62Xl8KM=
Content-Language: de-DE
Xref: csiph.com comp.programming.threads:7746

A thread-queue in C++20, with a flexibility (especially because of
the functional programming parts) that would never be possible with
FreePascal or Delphi; these languages simply suck.

#pragma once
#include <deque>
#include <mutex>
#include <condition_variable>
#include <utility>
#include <concepts>
#include <list>

template<typename QueueType>
concept thread_queue_concept =
	std::same_as<QueueType, std::deque<typename QueueType::value_type, 
typename QueueType::allocator_type>>
	|| std::same_as<QueueType, std::list<typename QueueType::value_type, 
typename QueueType::allocator_type>>;

template<typename QueueType>
	requires thread_queue_concept<QueueType>
struct thread_queue
{
	using value_type = typename QueueType::value_type;
	thread_queue();
	explicit thread_queue( typename QueueType::allocator_type const &alloc );
	thread_queue( thread_queue &&other );
	thread_queue &operator =( thread_queue const & ) = delete;
	thread_queue &operator =( thread_queue &&other );
	bool empty() const;
	std::size_t size() const;
	void shrink_to_fit();
	void clear();
	template<typename ... Args>
		requires std::is_constructible_v<typename QueueType::value_type, Args ...>
	void enque( Args &&... args );
	template<typename Producer>
		requires requires( Producer producer ) { { producer() } -> 
std::same_as<std::pair<bool, typename QueueType::value_type>>; }
	void enqueue_multiple( Producer producer );
	template<typename Consumer>
		requires requires( Consumer consumer, typename QueueType::value_type 
value ) { { consumer( std::move( value ) ) } -> std::same_as<bool>; }
	void dequeue_multiple( Consumer consumer );
	typename QueueType::value_type dequeue();
	QueueType container() noexcept;
	void swap( thread_queue &other );
private:
	mutable std::mutex m_mtx;
	mutable std::condition_variable m_cv;
	QueueType m_queue;
};

template<typename QueueType>
	requires thread_queue_concept<QueueType>
thread_queue<QueueType>::thread_queue()
{
}

template<typename QueueType>
	requires thread_queue_concept<QueueType>
thread_queue<QueueType>::thread_queue( typename 
QueueType::allocator_type const &alloc ) :
	m_queue( alloc )
{
}

template<typename QueueType>
	requires thread_queue_concept<QueueType>
thread_queue<QueueType>::thread_queue( thread_queue &&other )
{
	using namespace std;
	lock_guard lock( other.m_mtx );
	m_queue = move( other.m_queue );
}

template<typename QueueType>
	requires thread_queue_concept<QueueType>
thread_queue<QueueType> &thread_queue<QueueType>::thread_queue::operator 
=( thread_queue &&other )
{
	using namespace std;
	lock_guard
		ourLock( m_mtx ),
		otherLock( other.m_mtx );
	m_queue = move( other.m_queue );
	return *this;
}

template<typename QueueType>
	requires thread_queue_concept<QueueType>
bool thread_queue<QueueType>::thread_queue::empty() const
{
	std::lock_guard lock( m_mtx );
	return m_queue.empty();
}

template<typename QueueType>
	requires thread_queue_concept<QueueType>
std::size_t thread_queue<QueueType>::thread_queue::size() const
{
	std::lock_guard lock( m_mtx );
	return m_queue.size();
}

template<typename QueueType>
	requires thread_queue_concept<QueueType>
void thread_queue<QueueType>::thread_queue::shrink_to_fit()
{
	std::lock_guard lock( m_mtx );
	return m_queue.shrink_to_fit();
}

template<typename QueueType>
	requires thread_queue_concept<QueueType>
void thread_queue<QueueType>::thread_queue::clear()
{
	std::lock_guard lock( m_mtx );
	m_queue.clear();
}

template<typename QueueType>
	requires thread_queue_concept<QueueType>
template<typename ... Args>
	requires std::is_constructible_v<typename QueueType::value_type, Args ...>
void thread_queue<QueueType>::thread_queue::enque( Args &&... args )
{
	using namespace std;
	unique_lock lock( m_mtx );
	m_queue.emplace_front( forward<Args>( args ) ... );
	m_cv.notify_one();
}

template<typename QueueType>
	requires thread_queue_concept<QueueType>
typename QueueType::value_type 
thread_queue<QueueType>::thread_queue::dequeue()
{
	using namespace std;
	unique_lock lock( m_mtx );
	while( m_queue.empty() )
		m_cv.wait( lock );
	value_type value = move( m_queue.back() );
	m_queue.pop_back();
	return value;
}

template<typename QueueType>
	requires thread_queue_concept<QueueType>
template<typename Producer>
	requires requires( Producer producer ) { { producer() } -> 
std::same_as<std::pair<bool, typename QueueType::value_type>>; }
void thread_queue<QueueType>::enqueue_multiple( Producer producer )
{
	using namespace std;
	lock_guard lock( m_mtx );
	for( std::pair<bool, value_type> ret; (ret = move( producer() )).first; )
		m_queue.emplace_front( move( ret.second ) ),
		m_cv.notify_one();
}

template<typename QueueType>
	requires thread_queue_concept<QueueType>
template<typename Consumer>
	requires requires( Consumer consumer, typename QueueType::value_type 
value ) { { consumer( std::move( value ) ) } -> std::same_as<bool>; }
void thread_queue<QueueType>::dequeue_multiple( Consumer consumer )
{
	using namespace std;
	unique_lock lock( m_mtx );
	for( ; ; )
	{
		while( m_queue.empty() )
			m_cv.wait( lock );
		try
		{
			bool cont = consumer( move( m_queue.back() ) );
			m_queue.pop_back();
			if( !cont )
				return;
		}
		catch( ... )
		{
			m_queue.pop_back();
			throw;
		}
	}
}

template<typename QueueType>
	requires thread_queue_concept<QueueType>
QueueType thread_queue<QueueType>::container() noexcept
{
	using namespace std;
	unique_lock lock( m_mtx );
	return move( m_queue );
}

template<typename QueueType>
	requires thread_queue_concept<QueueType>
void thread_queue<QueueType>::thread_queue::swap( thread_queue &other )
{
	std::lock_guard
		ourLock( m_mtx ),
		otherLock( other.m_mtx );
	m_queue.swap( other.m_queue );
}