<commit_before>/* <Actor.h>
 *
 * This file is part of the x0 web server project and is released under GPL-3.
 * http://www.xzero.io/
 *
 * (c) 2009-2013 <NAME> <<EMAIL>>
 */

#ifndef sw_x0_Actor_h
#define sw_x0_Actor_h (1)

#include <x0/Api.h>

#include <deque>
#include <vector>
#include <thread>
#include <future>
#include <mutex>
#include <condition_variable>
#include <memory>
#include <functional>

namespace x0 {

template<typename Message>
class X0_API Actor
{
private:
	bool shutdown_;
	std::deque<Message> messages_;
	std::vector<std::future<void>> threads_;
	std::mutex mutex_;
	std::unique_lock<std::mutex> lock_;
	std::condition_variable cond_;

public:
	explicit Actor(size_t scalability = 1);
	virtual ~Actor();

	bool empty() const;
	size_t size() const;
	int scalability() const { return threads_.size(); }

	void send(const Message& message);

	void push_back(const Message& message) { send(message); }
	Actor<Message>& operator<<(const Message& message) { send(message); return *this; }

	void start();
	void stop();
	void join();

protected:
	virtual void process(Message message) = 0;

private:
	void main();
};

// {{{ impl
template<typename Message>
inline Actor<Message>::Actor(size_t scalability) :
	shutdown_(false),
	messages_(),
	threads_(scalability),
	mutex_(),
	lock_(mutex_),
	cond_()
{
}

template<typename Message>
inline Actor<Message>::~Actor()
{
}

template<typename Message>
bool Actor<Message>::empty() const
{
	return size() == 0;
}

template<typename Message>
size_t Actor<Message>::size() const
{
	std::lock_guard<decltype(mutex_)> l(mutex_);
	size_t result = messages_.size();
	return result;
}

template<typename Message>
inline void Actor<Message>::send(const Message& message)
{
	std::lock_guard<decltype(mutex_)> l(mutex_);
	messages_.push_back(message);
	cond_.notify_one();
}

template<typename Message>
void Actor<Message>::start()
{
	shutdown_ = false;

	for (auto& thread: threads_) {
		thread = std::move(std::async(std::bind(&Actor<Message>::main, this)));
	}
}

template<typename Message>
inline void Actor<Message>::stop()
{
	shutdown_ = true;
	cond_.notify_all();
}

template<typename Message>
inline void Actor<Message>::join()
{
	for (auto& thread: threads_) {
		thread.wait();
	}
}

template<typename Message>
void Actor<Message>::main()
{
	std::lock_guard<decltype(mutex_)> l(mutex_);

	for (;;) {
		cond_.wait(lock_);

		if (shutdown_)
			break;

		Message message = messages_.front();
		messages_.pop_front();

			process(message);
		/*lock_.unlock();
		try {
		} catch (...) {
			lock_.lock();
			throw;
		}
		lock_.lock();*/
	}
}
// }}}

} // namespace x0

#endif
<commit_msg>[base] Actor: use lock-free Queue over std::deque.<commit_after>/* <Actor.h>
 *
 * This file is part of the x0 web server project and is released under GPL-3.
 * http://www.xzero.io/
 *
 * (c) 2009-2013 <NAME> <<EMAIL>>
 */

#ifndef sw_x0_Actor_h
#define sw_x0_Actor_h (1)

#include <x0/Api.h>
#include <x0/Queue.h>

#include <vector>
#include <thread>
#include <future>
#include <mutex>
#include <condition_variable>
#include <memory>
#include <functional>

namespace x0 {

template<typename Message>
class X0_API Actor
{
private:
	bool shutdown_;
	Queue<Message> messages_;
	std::vector<std::future<void>> threads_;

	std::mutex mutex_;
	std::unique_lock<std::mutex> lock_;
	std::condition_variable cond_;

public:
	explicit Actor(size_t scalability = 1);
	virtual ~Actor();

	bool empty() const;
	int scalability() const { return threads_.size(); }

	void send(const Message& message);

	void push_back(const Message& message) { send(message); }
	Actor<Message>& operator<<(const Message& message) { send(message); return *this; }

	void start();
	void stop();
	void join();

protected:
	virtual void process(Message message) = 0;

private:
	void main();
};

// {{{ impl
template<typename Message>
inline Actor<Message>::Actor(size_t scalability) :
	shutdown_(false),
	messages_(),
	threads_(scalability),
	mutex_(),
	lock_(mutex_),
	cond_()
{
}

template<typename Message>
inline Actor<Message>::~Actor()
{
}

template<typename Message>
bool Actor<Message>::empty() const
{
	return messages_.empty();
}

template<typename Message>
inline void Actor<Message>::send(const Message& message)
{
	messages_.enqueue(message);
	cond_.notify_one();
}

template<typename Message>
void Actor<Message>::start()
{
	shutdown_ = false;

	for (auto& thread: threads_) {
		thread = std::move(std::async(std::bind(&Actor<Message>::main, this)));
	}
}

template<typename Message>
inline void Actor<Message>::stop()
{
	shutdown_ = true;
	cond_.notify_all();
}

template<typename Message>
inline void Actor<Message>::join()
{
	for (auto& thread: threads_) {
		thread.wait();
	}
}

template<typename Message>
void Actor<Message>::main()
{
	std::lock_guard<decltype(mutex_)> l(mutex_);

	for (;;) {
		cond_.wait(lock_);

		if (shutdown_)
			break;

		Message message;
		while (messages_.dequeue(&message)) {
			process(message);
		}
	}
}
// }}}

} // namespace x0

#endif

{'fraction_non_alphanumeric': 0.1619943932719263, 'fraction_numerical': 0.008009611533840609, 'mean_word_length': 5.275125628140704, 'pattern_counts': {'":': 0, '<': 85, '<?xml version=': 0, '>': 81, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '4581314', 'n_tokens_mistral': 1806, 'n_tokens_neox': 1709, 'n_words': 416}