Scheduler.h

Go to the documentation of this file.

/*
* i6engine
* Copyright (2016) Daniel Bonrath, Michael Baer, All rights reserved.
*
* This file is part of i6engine; i6engine is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
*/

#ifndef __I6ENGINE_CORE_SCHEDULER_H__
#define __I6ENGINE_CORE_SCHEDULER_H__

#include <queue>
#include <set>

#include "i6engine/utils/Clock.h"
#include "i6engine/utils/Exceptions.h"

#include "i6engine/core/configs/JobPriorities.h"
#include "i6engine/core/configs/SchedulerConfig.h"

#include "boost/thread.hpp"

namespace i6e {
namespace core {

        template<class ClockUpdater>
        class Scheduler {
        public:
                typedef struct Job {
                        Job(const boost::function<bool(void)> & f, uint64_t t, JobPriorities p, uint64_t d, uint64_t i = UINT64_MAX) : func(f), time(t), priority(p), interval(i), id(d) {
                        }

                        boost::function<bool(void)> func;

                        uint64_t time;

                        JobPriorities priority; // the lower, the better

                        uint64_t interval;

                        uint64_t id;

                        bool operator<(const Job & other) const {
                                if (time != other.time) {
                                        return time > other.time;
                                }
                                if (priority != other.priority) {
                                        return priority > other.priority;
                                }
                                return interval > other.interval;
                        }
                } Job;

                explicit Scheduler(utils::Clock<ClockUpdater> & c) : _running(true), _clock(c), _queue(), _lock(), _id(), _workerThreads(), _removeIDs() {
                        for (uint16_t i = 0; i < SCHEDULER_THREAD_AMOUNT; i++) {
                                uint64_t tid = _clock.registerTimer();
                                _workerThreads.push_back(std::make_pair(tid, new boost::thread(boost::bind(&Scheduler<ClockUpdater>::worker, this, tid))));
                        }
                }

                ~Scheduler() {
                        _running = false;
                        _lock.lock();
                        while (!_queue.empty()) {
                                _queue.pop();
                        }
                        _lock.unlock();
                        for (uint16_t i = 0; i < SCHEDULER_THREAD_AMOUNT; i++) {
                                _clock.updateWaitTime(_workerThreads[i].first, 0);
                                _workerThreads[i].second->join();
                                delete _workerThreads[i].second;
                        }
                }

                uint64_t runOnce(uint64_t time, const boost::function<bool(void)> & f, JobPriorities priority) {
                        if (time <= 0) {
                                ISIXE_THROW_API("Scheduler", "time need to be > 0");
                        }

                        Job j(f, _clock.getTime() + time, priority, _id++);
                        boost::mutex::scoped_lock sl(_lock);
                        _queue.push(j);
                        if (_queue.top().time == j.time) {
                                for (uint16_t i = 0; i < SCHEDULER_THREAD_AMOUNT; i++) {
                                        _clock.updateWaitTime(_workerThreads[i].first, j.time);
                                }
                        }

                        return j.id;
                }

                uint64_t runRepeated(uint64_t interval, const boost::function<bool(void)> & f, JobPriorities priority) {
                        if (interval <= 0) {
                                ISIXE_THROW_API("Scheduler", "interval has to be greater than 0, otherwise there would be an instant call");
                        }

                        Job j(f, _clock.getTime() + interval, priority, _id++, interval);
                        boost::mutex::scoped_lock sl(_lock);
                        _queue.push(j);
                        if (_queue.top().time == j.time) {
                                for (uint16_t i = 0; i < SCHEDULER_THREAD_AMOUNT; i++) {
                                        _clock.updateWaitTime(_workerThreads[i].first, j.time);
                                }
                        }
                        return j.id;
                }

                uint64_t getTimeLeft(uint64_t id) const {
                        _lock.lock();
                        std::priority_queue<Job> copy = _queue;
                        _lock.unlock();

                        while (!copy.empty()) {
                                Job j = copy.top();
                                copy.pop();

                                if (j.id == id) {
                                        return j.time - _clock.getTime();
                                }
                        }

                        return UINT64_MAX;
                }

                void removeTimer(JobPriorities priority) {
                        boost::mutex::scoped_lock sl(_lock);
                        std::priority_queue<Job> copy = _queue;

                        while (!_queue.empty()) {
                                _queue.pop();
                        }

                        while (!copy.empty()) {
                                Job j = copy.top();
                                copy.pop();

                                if (j.priority != priority) {
                                        _queue.push(j);
                                }
                        }
                }

                bool stop(uint64_t id) {
                        boost::mutex::scoped_lock sl(_lock);
                        _removeIDs.insert(id);
                        return true;
                }

        private:
                // TODO: (Michael) convert to a threadpool if needed
                void worker(uint64_t tid) {
                        while (_running) {
                                _lock.lock();
                                while (!_queue.empty() && _queue.top().time <= _clock.getTime()) {
                                        Job j = _queue.top();
                                        _queue.pop();
                                        if (_removeIDs.find(j.id) != _removeIDs.end()) {
                                                _removeIDs.erase(j.id);
                                                continue;
                                        }
                                        _lock.unlock();
                                        bool b = j.func();
                                        _lock.lock();
                                        if (j.interval != UINT64_MAX && b) {
                                                j.time = _clock.getTime() + j.interval;
                                                if (_removeIDs.find(j.id) != _removeIDs.end()) {
                                                        _removeIDs.erase(j.id);
                                                } else {
                                                        _queue.push(j);
                                                }
                                        }
                                }
                                uint64_t t = _clock.getTime() + 1000000; // sleep 1 second if no task is there
                                if (!_queue.empty()) {
                                        t = _queue.top().time;
                                }
                                _lock.unlock();
                                if (!_clock.waitForTime(tid, t)) {
                                        break;
                                }
                        }
                }

                std::atomic<bool> _running;

                utils::Clock<ClockUpdater> & _clock;

                std::priority_queue<Job> _queue;

                mutable boost::mutex _lock;

                std::atomic<uint64_t> _id;

                std::vector<std::pair<uint64_t, boost::thread *>> _workerThreads;

                std::set<uint64_t> _removeIDs;

                Scheduler(const Scheduler &) = delete;
                Scheduler & operator=(const Scheduler &) = delete;
        };

} /* namespace core */
} /* namespace i6e */

#endif /* __I6ENGINE_CORE_SCHEDULER_H__ */