SerializedTaskArena.h

#include "tbb/tbb.h"
#include "tbb/concurrent_queue.h"
#include "tbb/task_arena.h"

// Based on LocalSerializer pattern which enforces execution order based on submission order
// https://www.threadingbuildingblocks.org/docs/help/tbb_userguide/Design_Patterns/Local_Serializer.html

class SerializedTaskArena
{
public:

	struct WorkItem
	{
		virtual void run() = 0;
	};

private:

	tbb::task_arena taskArena_;
	tbb::concurrent_queue<WorkItem*> readyQueue_;
	tbb::concurrent_queue<WorkItem*> waitingQueue_;
	std::atomic<int> waitingCount_;
	std::condition_variable allCompleteCV_;
	std::mutex allCompleteMutex_;
	int totalCount_;

	// Transfer item from waiting queue to ready queue
	void moveOneItemToReadyQueue()
	{
		WorkItem* item = nullptr;
		waitingQueue_.try_pop( item );
		readyQueue_.push( item );
		taskArena_.enqueue([this]() {
			runNextReadyItem();
			{
				std::lock_guard<std::mutex> lock(allCompleteMutex_);
				--totalCount_;
			}
			allCompleteCV_.notify_one();
		});
	}

	void runNextReadyItem()
	{
		WorkItem* item = nullptr;
		readyQueue_.try_pop( item );
		item->run();
		if (--waitingCount_ != 0)
		{
			moveOneItemToReadyQueue();
		}
		// TODO: recycle
		delete item;
	}

public:

	SerializedTaskArena()
		: taskArena_( 1 )
		, totalCount_( 0 )
	{
		waitingCount_ = 0;
	}

	~SerializedTaskArena()
	{
		waitOnAll();
	}

	void enqueue( WorkItem* item )
	{
		{
			std::lock_guard<std::mutex> lock(allCompleteMutex_);
			++totalCount_;
		}
		waitingQueue_.push( item );
		if (++waitingCount_ == 1)
		{
			moveOneItemToReadyQueue();
		}
	}
	void waitOnAll()
	{
		std::unique_lock<std::mutex> lock(allCompleteMutex_);
		allCompleteCV_.wait(lock, [this] {return totalCount_ == 0;});
	}
};