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();
// [...]

```
New contributor
yorztif is a new contributor to this site. Take care in asking for clarification, commenting, and answering. Check out our Code of Conduct.
\$\endgroup\$

Your Answer

yorztif is a new contributor. Be nice, and check out our Code of Conduct.

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.