1
\$\begingroup\$

I'm practicing C++11/14/17 and am teaching myself the STL multithreading library by writing an asynchronous dispatch queue for my simple use case. It seems to "work ok" but I want to know if it's actually correct and if there are any improvements that I can make for any aspect of my code, including performance.

The interface has just 3 methods:

  • enqueue a function/lambda (task) for a worker thread to call.
  • cancel any pending tasks by flushing the pending queue, but leave workers running.
  • wait block the parent/main thread until the workers have completed all the pending tasks. Workers remain running.

Questions:

  • If I'm enqueuing tasks in a loop, is it inefficient to be locking the mutex each iteration, even if there's no contention yet?

  • I'm not sure how to approach exceptions thrown from a task function. My use case is to do 'pure' data-parallel work (e.g. divide a large raster into many sub-rects for workers to crunch). Should it be the responsibility of my dispatch_queue_t::thread_handler to catch exceptions or is it the responsibility of the caller/client to catch them in their task function that they pass to enqueue? Can I somehow express/enforce a constraint stating that a task shouldn't throw?

dispatch_queue.h

#if !defined(_DISPATCH_QUEUE_H_)
#define _DISPATCH_QUEUE_H_

#include <thread>
#include <mutex>
#include <atomic>

#include <deque>
#include <vector>

#include <cstdint>


class dispatch_queue_t {
public:

    typedef std::function<void()> task_t;

    dispatch_queue_t();
    ~dispatch_queue_t();

    template<typename F>
    void enqueue(F &&task) {
        std::unique_lock<std::mutex> lock(_pending_tasks_mutex);
        _pending_tasks.emplace_back(std::forward<F>(task));
        lock.unlock();

        _task_pending.notify_one();
    }

    void cancel();
    void wait();

private:

    void thread_handler();

    // neither copyable nor movable
    dispatch_queue_t(const dispatch_queue_t &other) = delete;
    dispatch_queue_t(dispatch_queue_t &&other) = delete;
    dispatch_queue_t &operator=(const dispatch_queue_t &other) = delete;
    dispatch_queue_t &operator=(dispatch_queue_t &&other) = delete;

private:

    std::vector<std::thread> _worker_threads;

    std::deque<task_t> _pending_tasks;
    std::mutex _pending_tasks_mutex;

    std::condition_variable _task_pending;
    std::condition_variable _task_completion;

    std::atomic_bool _done = false;
    std::atomic_uint8_t _num_busy = 0;
};

#endif // _DISPATCH_QUEUE_H_

dispatch_queue.cpp

#include <dispatch_queue.h>


dispatch_queue_t::dispatch_queue_t() {
    const std::size_t n = std::thread::hardware_concurrency();
    for (auto i = 0U; i < n; ++i) {
        _worker_threads.emplace_back(&dispatch_queue_t::thread_handler, this);
    }
}


dispatch_queue_t::~dispatch_queue_t() {
    _done = true;
    _task_pending.notify_all(); // wake threads to check for `done` condition and exit`

    for (auto &t : _worker_threads) {
        if (t.joinable()) {
            t.join();
        }
    }
}


void dispatch_queue_t::thread_handler() {
    while (true) {
        std::unique_lock<std::mutex> lock(_pending_tasks_mutex);

        // wait unless exiting or there's pending tasks
        _task_pending.wait(lock, [this](){
            return _done || _pending_tasks.size() > 0;
        });

        if (_done) {
            return;
        }

        // at this point there must be work
        _num_busy++;

        try {
            auto task = std::move(_pending_tasks.front());
            _pending_tasks.pop_front();
            lock.unlock();

            task();
        } catch(...) {
            // TODO: I'm not sure what to do here
        }

        _num_busy--;
        _task_completion.notify_one();
    }
}


void dispatch_queue_t::cancel() {
    std::unique_lock<std::mutex> lock(_pending_tasks_mutex);
    _pending_tasks.clear();
    lock.unlock();
}


void dispatch_queue_t::wait() {
    std::unique_lock<std::mutex> lock(_pending_tasks_mutex);
    // block until no pending tasks remain and all workers are idle
    _task_completion.wait(lock, [this]() { 
        return _pending_tasks.empty() && _num_busy == 0; 
    });
}

My test case (in main.cpp)

// [...]
const uint32_t num_chunks = 1 * 1000 * 1000;;

dispatch_queue_t task_queue;

std::atomic<uint64_t> num_tasks_processed = 0; // expected to be 1M
std::vector<uint32_t> chunk_results(num_chunks, 0);

for (auto i = 0; i < num_chunks; ++i) {
    task_queue.enqueue([&, i]() {
        chunk_results[i] = (i * i); // pretend this is an expensive op
        ++num_tasks_processed;
    });
}

task_queue.wait();
// [...]

```
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.