A worker thread in C++

Introduction

Sometimes we need an extra thread to do a task for us in parallel or to just unblock main thread. For better performance we would not like to create a new thread every time, Also we need little more control over the thread. In this article, we will implement a thread with below abilities.

• Submit a task to be executed asynchronously
• Submit a task to be executed synchronously ( but on worker thread)
• Automatically wait for task to complete 
• Implementation uses C++ 11 extensively ( thread , lambda , condition variables)

Using the code

Below sample code list usages of worker thread

Hide   Shrink    Copy Code

#include <iostream>
#include <chrono>
#include <thread>
#include "workerthread.h"

int main()
{
    std::cout << "Hi, Welcome to demo of worker thread" << std::endl;

    {
        // Create two worker threads
        WorkerThread thread;
        WorkerThread thread2;

        // Example of a synchronous task
        thread.doSync([]{    std::cout << "First - blocking call" << std::endl;    });

        for (int i = 1; i < 100; i++)
        {
            auto& t = i % 2 == 0 ? thread : thread2;
            if (i == 10) // Another synchronous task in between
            {
                thread.doSync([]{    std::cout << "Second - blocking call" << std::endl;    });
            }
            // Multiple asynchronous tasks queued
            t.doAsync([i]
            {
                std::this_thread::sleep_for(std::chrono::duration<int, std::milli>(200));
                std::cout << (i % 2 == 0 ? "thread-" : "thread2-") << "iteration number: " << i <<                std::endl;
            });
        }
    
        thread.doSync([]{    std::cout << "Last - blocking call";    });
    }// Added extra scope to demo destruction of worker thread
    std::cout << "This must be last line\n";

}

Implementation of worker thread

• Use C++ function construct to store submitted task
• Finish the present queue and wait for signal for new work

Hide   Copy Code

void WorkerThread::startThread()
{
    std::unique_lock<std::mutex> l(mutex);
    do
    {
        while (isRunning && tasks.empty())
            itemInQueue.wait(l);

        while (!tasks.empty())
        {
            auto size = tasks.size();
            printf("Number of pending task are %d\n", size);
            const std::function<void()> t = tasks.front();
            tasks.pop_front();
            l.unlock();
            t();
            l.lock();
        }
        itemInQueue.notify_all();

    } while (isRunning);
    itemInQueue.notify_all();
}

An asynchronous task is just queued and calling thread not blocked

Hide   Copy Code

void WorkerThread::doAsync(const std::function<void()>& t)
{
    std::lock_guard<std::mutex> _(mutex);
    tasks.push_back(t);
    itemInQueue.notify_one();

}

 Synchronous task is little trickier, We need to block calling thread until all task queued before the submitted task and submitted task completes.

Hide   Copy Code

void WorkerThread::doSync(const std::function<void()>& t)
{
    std::condition_variable event;
    bool finished = false;

    std::unique_lock<std::mutex> l(mutex);
    auto lambda = [this, &t, &finished, &event]
    {
        t();
        std::lock_guard<std::mutex> l(mutex);

        finished = true;
        event.notify_one();
    };
    tasks.push_back(lambda);
    itemInQueue.notify_one();

    while (!finished)
        event.wait(l);

}

Next update

- Implement priority of task

- Allow cancellation of task