dispatch_queue.hpp

Go to the documentation of this file.

#pragma once
 
#include <functional>
#include <utility>
 
#include "detail/function_result.hpp"
#include "detail/pending_task_queue.hpp"
#include "detail/promise.hpp"
#include "detail/ranges.hpp"
#include "detail/worker_pool.hpp"
#include "task_tag.hpp"
#include "task.hpp"
#include "when_all.hpp"
 
#ifndef DISPATCH_QUEUE_DEFAULT_BATCH_SIZE
    #define DISPATCH_QUEUE_DEFAULT_BATCH_SIZE 64
#endif
 
namespace dispatch_queue {
 
class dispatch_queue {
public:
    dispatch_queue();

    dispatch_queue(int thread_count);

    template<typename Fn>
    dispatch_queue(int thread_count, Fn&& worker_init) {
        if (thread_count < 0) {
            thread_count = std::thread::hardware_concurrency();
        }
        if (thread_count > 0) {
            worker_pool = std::make_unique<detail::worker_pool>(task_queue, thread_count, std::move(worker_init));
        }
    }
 
    dispatch_queue(const dispatch_queue&) = delete;
    dispatch_queue& operator=(const dispatch_queue&) = delete;

    ~dispatch_queue();

    template<typename F, typename... Args, typename Ret = detail::function_result<F, Args...>>
    task<Ret> dispatch(F&& f, Args&&... args) {
        return dispatch_internal(detail::task_type::background, NULL_TAG, std::forward<F>(f), std::forward<Args>(args)...);
    }

    template<typename F, typename... Args, typename Ret = detail::function_result<F, Args...>>
    task<Ret> dispatch_main(F&& f, Args&&... args) {
        return dispatch_internal(detail::task_type::main, NULL_TAG, std::forward<F>(f), std::forward<Args>(args)...);
    }

    template<typename F, typename... Args, typename Ret = detail::function_result<F, Args...>>
    task<Ret> dispatch_tagged(task_tag tag, F&& f, Args&&... args) {
        return dispatch_internal(detail::task_type::tagged, tag, std::forward<F>(f), std::forward<Args>(args)...);
    }

    template<typename F, typename It>
    task<void> parallel_for(F&& f, const It& begin, const It& end, size_t batch_size = DISPATCH_QUEUE_DEFAULT_BATCH_SIZE) {
        std::vector<task<void>> tasks;
        detail::apply_batches([&](auto&& batch_begin, auto&& batch_end) {
            tasks.emplace_back(dispatch([=] {
                for (auto it = batch_begin; it != batch_end; ++it) {
                    f(*it);
                }
            }));
        }, begin, end, batch_size);
        return when_all(tasks);
    }

    template<typename F, typename R>
    task<void> parallel_for(F&& f, R&& range, size_t batch_size = DISPATCH_QUEUE_DEFAULT_BATCH_SIZE) {
        std::vector<task<void>> tasks;
        detail::apply_batches([&](auto&& batch_begin, auto&& batch_end) {
            tasks.emplace_back(dispatch([=] {
                for (auto it = batch_begin; it != batch_end; ++it) {
                    f(*it);
                }
            }));
        }, range, batch_size);
        return when_all(tasks);
    }

    bool is_threaded() const;

    int thread_count() const;

    size_t size() const;

    bool empty() const;

    void clear();

    void main_loop();

    void wait();

    template<class Rep, class Period>
    bool wait_for(const std::chrono::duration<Rep, Period>& timeout_duration) {
        if (worker_pool) {
            return worker_pool->wait_for(timeout_duration);
        }
        else {
            return true;
        }
    }

    template<class Clock, class Duration>
    bool wait_until(const std::chrono::time_point<Clock, Duration>& timeout_time) {
        if (worker_pool) {
            return worker_pool->wait_until(timeout_time);
        }
        else {
            return true;
        }
    }

    void shutdown();
 
#ifdef __cpp_lib_coroutine
private:
    struct dispatch_awaiter {
        dispatch_queue& dispatch_queue;
 
        bool await_ready() const noexcept { return false; }
        void await_suspend(std::coroutine_handle<> cont) const {
            dispatch_queue.dispatch([cont]{
                cont();
                if (cont.done()) {
                    cont.destroy();
                }
            });
        }
        void await_resume() {}
    };
 
    struct dispatch_main_awaiter {
        dispatch_queue& dispatch_queue;
 
        bool await_ready() const noexcept { return false; }
        void await_suspend(std::coroutine_handle<> cont) const {
            dispatch_queue.dispatch_main([cont]{
                cont();
                if (cont.done()) {
                    cont.destroy();
                }
            });
        }
        void await_resume() {}
    };
 
    struct dispatch_tagged_awaiter {
        dispatch_queue& dispatch_queue;
        task_tag tag;
 
        bool await_ready() const noexcept { return false; }
        void await_suspend(std::coroutine_handle<> cont) const {
            dispatch_queue.dispatch_tagged(tag, [cont]{
                cont();
                if (cont.done()) {
                    cont.destroy();
                }
            });
        }
        void await_resume() {}
    };
public:
    dispatch_awaiter dispatch() {
        return dispatch_awaiter(*this);
    }

    dispatch_main_awaiter dispatch_main() {
        return dispatch_main_awaiter(*this);
    }

    dispatch_tagged_awaiter dispatch_tagged(task_tag tag) {
        return dispatch_tagged_awaiter(*this, tag);
    }
#endif
 
private:
    std::unique_ptr<detail::worker_pool> worker_pool;
    detail::pending_task_queue task_queue;
 
    template<typename F, typename... Args, typename Ret = detail::function_result<F, Args...>>
    task<Ret> dispatch_internal(detail::task_type type, task_tag tag, F&& f, Args&&... args) {
        auto work = std::bind(std::move(f), std::forward<Args>(args)...);
        if (worker_pool) {
            auto future = detail::task_future<Ret>::create_pending();
            worker_pool->enqueue_task(type, { future->wrap(work) }, tag);
            return task<Ret>(future);
        }
        else if (type == detail::task_type::main) {
            auto future = detail::task_future<Ret>::create_pending();
            task_queue.push(type, { future->wrap(work) });
            return task<Ret>(future);
        }
        else {
            auto future = detail::task_future<Ret>::create(work);
            return task<Ret>(future);
        }
    }
};
 
} // end namespace dispatch_queue