Part (2/2): Active Objects for Asynchronous Logging.

Contra-auguste Active Object

TL;DR It's just an slightly better (imho) implementation of Sutter's Dr. Dobbs 2010 articles [1][2]

As per Lavender & Schmidt's article[3] Sutter's implementation is a further refinement of the integrated scheduler variant of their original pattern removing, as it does, altogether the proxy and scheduler relying on direct message passing between client and scheduler. 

Indeed Sutter's scheduler (his Active class) contains the message queue as a private composition -which is too much integration. Too much because it forces the client to be aware of the scheduler which as an unnecessary, and avoidable, level of dependancy that binds the scheduler to the client.

The client need only be aware of the tail of the activation queue and, hence, the has-a-relationship with the message queue can be broken out of the scheduler for a more naturally component orientated approach that facilitates scheduler fungibility. Further, the scheduler need only be aware of the head of the activation queue.

All of which leads to the message request scheduler:

	#include <thread>
	#include <stdexcept>
	namespace aob {
	template<typename T>
	class scheduler {
	public:
	scheduler(T& dq): dq(dq) {
	worker = std::unique_ptr<std::thread>(new std::thread([this] { this->run(); }));
	}
	~scheduler() {
	worker->join();
	}
	private:
	void run() {
	try {
	while(true) {
	dq.dequeue().operator()();
	}
	}
	catch (std::runtime_error& e) {
	std::cout << e.what() << std::endl;
	}
	}
	scheduler(const scheduler&) = delete;
	void operator=( const scheduler& ) = delete;
	T& dq;
	std::unique_ptr<std::thread> worker;
	};
	}

view raw scheduler.cpp hosted with ❤ by GitHub

The usage is as below and demonstrates that anything can be passed as message request function but an anonymous inner function expressed as a lambda function is an intuitive approach:

	using message_queue = que::shared_queue<que::gpcg_concurrent_queue<aob::message>>;
	message_queue q;
	//instantiate a dequeuer for the message queue and give it the scheduler
	que::dequeuer<message_queue> dq{q};
	aob::scheduler<que::dequeuer<message_queue>> s{dq};
	//instantiate and enqueuer for the message queue
	que::enqueuer<message_queue> nq{q};
	//enqueue a lamda function that prints hellow and waits a second
	nq.enqueue([=]{
	std::cout << std::this_thread::get_id() << " hello" << std::endl;
	std::this_thread::sleep_for(std::chrono::seconds(1));
	});
	//given the scheduler chance
	std::this_thread::sleep_for(std::chrono::milliseconds(500));
	//then kill it
	s.kill();
	//the scheduler is dead and won't print world
	nq.enqueue( [=]{std::cout << std::this_thread::get_id() << " world" << std::endl;} );
	//but this thread can
	auto o = dq.dequeue();
	o();
	return approve();

view raw test.cpp hosted with ❤ by GitHub

All of which makes for a very simple asynchronous stream serializor:

	namespace log {
	template<typename T>
	class async_stream {
	public:
	async_stream(T& nq, std::ostream& os): nq(nq), os(os) {}
	void operator << (std::string s) {
	nq.enqueue([this,s]{
	os << s;
	});
	}
	private:
	std::ostream& os;
	T& nq;
	};
	}

view raw async_stream.cpp hosted with ❤ by GitHub

The usage of async_stream within the logger is straightforward:

	std::ofstream file("test.log");
	log::async_stream<que::enqueuer<message_queue>> aout{nq, file};
	log::logger<decltype(aout)> flog(aout, {new log::line_number_stamp(), new log::line_number_stamp(10), new log::unix_time_stamp()});
	.
	.
	.
	flog("hello ", "file " "world ", 1,2,3);

view raw usage.cpp hosted with ❤ by GitHub

This little sojourn into utilising C++11 language features and idioms for a general purpose logger with an asynchronous stream serializor has (hopefully) been interesting, but does it bring us any nearer to an actual libfbp?

Perhaps the next blog entry will elicudate...?

References:

[1] Prefer Using Active Objects Instead of Naked Threads

[2] Prefer Using Futures or Callbacks to Communicate Asynchronous Results

[3] Active Object - Lavender & Schmidt 2010