4
\$\begingroup\$

I made a task scheduler to practice allocators and type erasure. With my project, you can delay execution of any callable (functions (using std::ref), lambdas...) in time by inserting it in a TaskScheduler, and by giving a std::chrono::time_point (you can also make a task be called multiple times by giving a period (std::chrono::duration) and optionally a number of call)).

This project does not use threads, so to make the tasks called, you have to call TaskScheduler::update (likely in a main loop), which will then call the tasks if needed. I have only basic knowledge on multi-threading so if you have any advice in order to make TaskScheduler thread-safe, I am interested.

Moreover, I have implemented type-erasure (so any callable can fit in the TaskScheduler) so I am interested on comments about the type-safety of this implementation.

Here is the code :

TaskScheduler.h

#ifndef TASK_SCHEDULER_J_VERNAY
#define TASK_SCHEDULER_J_VERNAY
#include <vector>
#include <chrono>
#include <memory_resource>
#include <tuple>

class TaskTraits {
public:
    using clock = std::chrono::system_clock;
    using time_point = clock::time_point;
    using duration = clock::duration;
};

class Task {
    friend class TaskScheduler;
public:
    TaskTraits::time_point when() const noexcept { return _when; }
    void delay(TaskTraits::duration delay_) noexcept { _when += delay_; }
    void set_when(TaskTraits::time_point when_) noexcept { _when = when_; }
    void cancel() noexcept { _fun = nullptr; }
    void execute() { if (_fun) (*_fun)(*this); }

    // for usage in std::sort for TaskScheduler
    static bool before(Task const& a, Task const& b) noexcept { return a._when < b._when; }
private:
    Task(void(*fun_)(Task&), void* args_, std::size_t size_, TaskTraits::time_point when_) noexcept :
        _fun{fun_}, _args{args_}, _size{size_}, _when{when_}
    {}
    TaskTraits::time_point _when;
    void(*_fun)(Task&);
    void* _args;
    std::size_t _size;
};

class TaskScheduler : private std::pmr::vector<Task> {
    using _base = vector;
public:

    using _base::value_type;
    using _base::reference;
    using _base::const_reference;
    using _base::difference_type;
    using _base::size_type;
    using _base::allocator_type;
    using _base::iterator;
    using _base::const_iterator;
    using _base::reverse_iterator;
    using _base::const_reverse_iterator;

    TaskScheduler(allocator_type a_ = {}) : _base(a_) {}
    TaskScheduler(TaskScheduler const& ts_, allocator_type a_ = {}) : _base(ts_, a_) {}
    TaskScheduler(TaskScheduler&& ts_) noexcept : _base(ts_) {}
    TaskScheduler(TaskScheduler&& ts_, allocator_type a_ = {}) noexcept : _base(ts_, a_) {}

    template<typename Fun>
    void insert(TaskTraits::time_point when_, Fun const& fun_);
    template<typename Fun>
    void insert_loop(TaskTraits::time_point when_, TaskTraits::duration loop_, Fun const& fun_);
    template<typename Fun>
    void insert_loop(TaskTraits::time_point when_, TaskTraits::duration loop_, int nbloops_, Fun const& fun_);

    void update();

    // with these iterators, you could first: sort the tasks with std::sort and Task::before,
    // then: iterate by chronological order. But notice the order will likely be modified by update()!
    using _base::begin;
    using _base::end;
    using _base::cbegin;
    using _base::cend;
    using _base::rbegin;
    using _base::rend;
    using _base::crbegin;
    using _base::crend;

    using _base::empty;
    using _base::size;
    using _base::clear;
    void erase(iterator pos) noexcept; // O(1) because order is not preserved

private:
    std::pmr::memory_resource* get_resource() const { return get_allocator().resource(); }
    template<typename Fun>
    void* store_args(Fun const&);
};

template<typename Fun>
void* TaskScheduler::store_args(Fun const& fun_) {
    void* args = nullptr;
    if constexpr (sizeof(fun_) <= sizeof(void*)) {
        get_allocator().construct(reinterpret_cast<Fun*>(&args), fun_);
    } else {
        args = get_resource()->allocate(sizeof(fun_));
        get_allocator().construct(reinterpret_cast<Fun*>(args), fun_);
    }
    return args;
}

template<typename Fun>
void TaskScheduler::insert(TaskTraits::time_point when_, Fun const& fun_) {
    void* args = store_args(fun_);

    auto fun = static_cast<void(*)(Task&)>(
        [](Task& t) {
            if constexpr (sizeof(Fun) <= sizeof(void*))
                reinterpret_cast<Fun&>(t._args)();
            else
                (*reinterpret_cast<Fun*>(t._args))();
            t.cancel();
        });


    push_back(Task(fun, args, sizeof(fun_), when_));
}


template<typename Fun>
void TaskScheduler::insert_loop(TaskTraits::time_point when_, TaskTraits::duration loop_, Fun const& fun_) {
    auto data = std::make_tuple(fun_, loop_);
    using data_t = decltype(data);
    void* args = store_args(data);

    auto fun = static_cast<void(*)(Task&)>(
        [](Task& t) {
            data_t* data = nullptr;
            if constexpr (sizeof(data_t) <= sizeof(void*))
                data = reinterpret_cast<data_t*>(&t._args); 
            else
                data = reinterpret_cast<data_t*>(t._args);
            std::get<Fun>(*data)();
            t.delay(std::get<TaskTraits::duration>(*data));
        });
    push_back(Task(fun, args, sizeof(data_t), when_));
}
template<typename Fun>
void TaskScheduler::insert_loop(TaskTraits::time_point when_, TaskTraits::duration loop_, int nbloops_, Fun const& fun_) {
    if (nbloops_ <= 0) return;
    auto data = std::make_tuple(fun_, loop_, nbloops_);
    using data_t = decltype(data);
    void* args = store_args(data);

    auto fun = static_cast<void(*)(Task&)>(
        [](Task& t) {
            data_t* data = nullptr;
            if constexpr (sizeof(Fun) <= sizeof(void*))
                data = reinterpret_cast<data_t*>(&t._args); 
            else
                data = reinterpret_cast<data_t*>(t._args);
            std::get<Fun>(*data)();
            if (--std::get<int>(*data) <= 0)
                t.cancel();
            else
                t.delay(std::get<TaskTraits::duration>(*data));
        });
    push_back(Task(fun, args, sizeof(data_t), when_));
}

#endif //TASK_SCHEDULER_J_VERNAY

TaskScheduler.cpp

#include "TaskScheduler.h"

void TaskScheduler::update() {
    TaskTraits::time_point now = TaskTraits::clock::now();
    std::size_t userdata_size = 0;
    for (std::size_t i = 0, c = size(); i < c; ++i) {
        auto task = begin() + i;
        while (task->when() < now) {
            task->execute();
            if (!task->_fun) {
                erase(task);
                --c;
                if (i >= c) return;
                task = begin() + i;
            }
        }
    }
}

void TaskScheduler::erase(TaskScheduler::iterator pos) noexcept {
    if (pos->_size > sizeof(void*))
        get_resource()->deallocate(pos->_args, pos->_size);
    *pos = back();
    pop_back();
}

main.cpp

#include <thread> // std::this_thread::sleep_for
#include <iostream>
#include <string>
#include "TaskScheduler.h"

using namespace std::chrono_literals;
void print_the_end() { std::printf("==== THE END ====\n"); }


int main() {
    std::string a = "Hello", b = "World";

    auto begin = TaskTraits::clock::now();
    TaskScheduler ts;
    ts.insert_loop(begin, 200ms, 25, [begin, &a, i = 0]() mutable { // looping during 200ms * 25 = 5s
        auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(TaskTraits::clock::now() - begin);
        std::printf("Iter %2d, t=%4dms : a = %s\n", ++i, (int)ms.count(), a.c_str());
    });

    for (std::size_t i = 0; i < 5; ++i) // changing a every 1 second
        ts.insert(begin + 500ms + i * 1s, [&a, b, i] {
            a[i] = b[i];
            std::printf("== a has changed! ==\n");
        });

    ts.insert(begin + 5s, std::ref(print_the_end));

    while (!ts.empty()) { // main loop
        ts.update();
        std::this_thread::sleep_for(5ms);
    }
    return 0;
}

Possible output

Iter  1, t=   0ms : a = Hello
Iter  2, t= 200ms : a = Hello
Iter  3, t= 402ms : a = Hello
== a has changed! ==
Iter  4, t= 605ms : a = Wello
Iter  5, t= 802ms : a = Wello
Iter  6, t=1004ms : a = Wello
Iter  7, t=1200ms : a = Wello
Iter  8, t=1402ms : a = Wello
== a has changed! ==
Iter  9, t=1605ms : a = Wollo
Iter 10, t=1801ms : a = Wollo
Iter 11, t=2001ms : a = Wollo
Iter 12, t=2202ms : a = Wollo
Iter 13, t=2404ms : a = Wollo
== a has changed! ==
Iter 14, t=2602ms : a = Worlo
Iter 15, t=2804ms : a = Worlo
Iter 16, t=3002ms : a = Worlo
Iter 17, t=3202ms : a = Worlo
Iter 18, t=3401ms : a = Worlo
== a has changed! ==
Iter 19, t=3601ms : a = Worlo
Iter 20, t=3801ms : a = Worlo
Iter 21, t=4002ms : a = Worlo
Iter 22, t=4201ms : a = Worlo
Iter 23, t=4402ms : a = Worlo
== a has changed! ==
Iter 24, t=4602ms : a = World
Iter 25, t=4801ms : a = World
==== THE END ====

Thank you for any review! =)

\$\endgroup\$
2
\$\begingroup\$

I like:

  • good include-guards, unlikely to collide with others
  • careful use of scoped names
  • efficient storage of small values (though the repeated access code could perhaps be refactored to a utility method in Task)

I dislike:

  • identifiers that differ only in _ prefix or suffix
  • pointless parens around value arguments to sizeof that make them look like type arguments
  • our types are in the global namespace

Specific issues:

  • Task(void(*fun_)(Task&), void* args_, std::size_t size_, TaskTraits::time_point when_) noexcept :
        _fun{fun_}, _args{args_}, _size{size_}, _when{when_}
    {}
    

    It's misleading to write the _when initializer last, as it will be initialized first. In this specific case, it doesn't matter, because there's no interdependencies, but I recommend always writing the initializers in the same order as the members.

  • void TaskScheduler::update() {
        std::size_t userdata_size = 0;
    

    This local variable is never used.

  • I don't see any reason why Task needs to be a publicly-visible class. Since we can't insert or obtain one, it should probably be a private struct within TaskScheduler.

  • When scheduling new tasks, we ought to accept a forwarding reference to a Fun, and use std::fwd() as appropriate, rather than copying from a const reference.

  • We're over-constraining the allocator by letting the alignment default to alignof(std::max_align_t) rather than using the actual alignment requirements like this:

    args = get_resource()->allocate(sizeof fun_, alignof(Fun));
    
  • The linear search for tasks to run might be a performance drag when there are lots of tasks. This is a problem for which the heap structure is a perfect fit (std::priority_queue).

\$\endgroup\$
  • \$\begingroup\$ Thank you for your review. Task is publicly-visible because you can iterate using begin(), end(), etc, so that you can cancel tasks or delay some. Because of this, something like a heap (or sorting using std::sort) should be recreated each update call \$\endgroup\$ – Julien Vernay Jul 30 at 7:48
  • \$\begingroup\$ If you're adding tasks one at a time, then don't sort each time - just use std::binary_search to insert each item in the right place. You'll still pay for moving the subsequent elements, but that's less than sorting the whole container. Or change to a std::multiset. \$\endgroup\$ – Toby Speight Jul 30 at 8:16

Your Answer

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

Not the answer you're looking for? Browse other questions tagged or ask your own question.