TaskQueue.cpp

/*
** libutil++
** $Id: TaskQueue.cpp 1653 2016-02-28 19:54:59Z sella $
** Copyright (c) 2011-2016 Digital Genesis, LLC. All Rights Reserved.
** Released under the LGPL Version 2.1 License.
** http://www.digitalgenesis.com
*/

static const char rcsid[] __attribute__((used)) = "$Id: TaskQueue.cpp 1653 2016-02-28 19:54:59Z sella $";

#include "TaskQueue.h"
#include "CommonMacro.h"
#include "Time.h"

#include <stdio.h>
#include <assert.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/time.h>

#include <algorithm>

using namespace sella::util;

class CompareTask {
        public:
                bool operator()(const Task::shared &t1, const Task::shared &t2) const {
                        if (UNLIKELY(t1 == NULL)) {
                                return true;
                        } else if (UNLIKELY(t2 == NULL)) {
                                return false;
                        }

                        const timeval &tv1 = t1->getTimer();
                        const timeval &tv2 = t2->getTimer();

                        if (UNLIKELY(timercmp(&tv1, &tv2, ==))) {
                                return t1->getPriority() < t2->getPriority();
                        } else if (!timerisset(&tv1)) {
                                return false;
                        } else if (!timerisset(&tv2)) {
                                return true;
                        } else if (timercmp(&tv1, &tv2, <)) {
                                return false;
                        }
                        return true;
                }
};

TaskQueue::TaskQueue(const size_t capacity) : libutilxx__log(::libutilxx__log) {
        reserve(capacity);
}

TaskQueue::TaskQueue(Log::shared log, const size_t capacity) : libutilxx__log(log) {
        reserve(capacity);
}

TaskQueue::~TaskQueue() { }

size_t TaskQueue::size(void) const {
        return heap.size();
}

bool TaskQueue::empty(void) const {
        return heap.empty();
}

void TaskQueue::clear(void) {
        heap.clear();
}

Task::shared TaskQueue::top(void) const {
        assert(!heap.empty());

        return heap.front();
}

Task::shared TaskQueue::next(void) const {
        if (empty()) {
                return Task::shared_ptr();
        }

        return heap.front();
}

const timeval TaskQueue::getDelay(timeval fallback) const {
        if (empty()) {
                return fallback;
        }

        return heap.front()->getDelay();
}

uint64_t TaskQueue::getDelayMsec(uint64_t fallback, uint64_t maximum) const {
        if (empty()) {
                return fallback;
        }

        uint64_t delay = heap.front()->getDelayMsec();

        if (maximum > 0 && delay > maximum) {
                return maximum;
        }

        return delay;
}

uint64_t TaskQueue::getDelayUsec(uint64_t fallback, uint64_t maximum) const {
        if (empty()) {
                return fallback;
        }

        uint64_t delay = heap.front()->getDelayUsec();

        if (maximum > 0 && delay > maximum) {
                return maximum;
        }

        return delay;
}

void TaskQueue::push(Task::shared &task) {
        assert(task.get() != NULL);

        heap.push_back(task);
        push_heap(heap.begin(), heap.end(), CompareTask());
}

void TaskQueue::push(Task::shared &task, const std::chrono::system_clock::duration &delay, int priority) {
        assert(task.get() != NULL);

        task->setDelay(delay);
        if (priority) task->setPriority(priority);

        push(task);
}

void TaskQueue::push(Task::shared &task, const std::chrono::system_clock::time_point &time, int priority) {
        assert(task.get() != NULL);

        task->setTimer(time);
        if (priority) task->setPriority(priority);

        push(task);
}

void TaskQueue::push(Task::shared &task, time_t sec, suseconds_t usec, int priority) {
        assert(task.get() != NULL);

        task->setDelay(sec, usec);
        if (priority) task->setPriority(priority);

        push(task);
}

void TaskQueue::push(Task::shared &task, timeval &tv, int priority) {
        assert(task.get() != NULL);

        task->setTimer(tv);
        if (priority) task->setPriority(priority);

        push(task);
}

bool TaskQueue::repush(Task::shared &task, const std::chrono::system_clock::duration &delay, int priority, bool strict) {
        timeval tv = Time::chronototv(delay);

        return repush(task, tv, priority, strict);
}

bool TaskQueue::repush(Task::shared &task, const std::chrono::system_clock::time_point &time, int priority, bool strict) {
        timeval tv = Time::chronototv(time);

        return repush(task, tv, priority, strict);
}

bool TaskQueue::repush(Task::shared &task, time_t sec, suseconds_t usec, int priority, bool strict) {
        assert(task.get() != NULL);

        if (task.unique()) { /* A unique pointer won't be in the list. */
                if (strict) {
                        return false;
                }

                push(task, sec, usec, priority);

                return true;
        }

        for (auto it = heap.cbegin(); it != heap.cend(); ++it) {
                if (*it == task) {
                        (*it)->setDelay(sec, usec);
                        if (priority) (*it)->setPriority(priority);
                        make_heap(heap.begin(), heap.end(), CompareTask());

                        return true;
                }
        }

        if (strict) {
                return false;
        }

        push(task, sec, usec, priority);

        return true;
}

bool TaskQueue::repush(Task::shared &task, timeval &tv, int priority, bool strict) {
        assert(task.get() != NULL);

        if (task.unique()) { /* A unique pointer won't be in the list. */
                if (strict) {
                        return false;
                }

                push(task, tv, priority);

                return true;
        }

        for (auto it = heap.cbegin(); it != heap.cend(); ++it) {
                if (*it == task) {
                        (*it)->setTimer(tv);
                        if (priority) (*it)->setPriority(priority);
                        make_heap(heap.begin(), heap.end(), CompareTask());

                        return true;
                }
        }

        if (strict) {
                return false;
        }

        push(task, tv, priority);

        return true;
}

void TaskQueue::pop(void) {
        assert(!heap.empty());

        pop_heap(heap.begin(), heap.end(), CompareTask());
        heap.pop_back();
}

bool TaskQueue::erase(Task::shared &task, bool firstonly) {
        bool found = false;

        if (task.unique()) { /* A unique pointer won't be in the list. */
                return false;
        }

        auto it = heap.begin();
        while (it != heap.end()) {
                if (*it == task) {
                        it = heap.erase(it);

                        found = true;
                        if (firstonly) break;
                } else {
                        ++it; /* Faster than it++. */
                }
        }

        make_heap(heap.begin(), heap.end(), CompareTask());

        return found;
}

bool TaskQueue::erase(void *object) {
        bool found = false;

        auto it = heap.begin();
        while (it != heap.end()) {
                if ((*it)->getObject() == object) {
                        it = heap.erase(it);

                        found = true;
                } else {
                        ++it; /* Faster than it++. */
                }
        }

        make_heap(heap.begin(), heap.end(), CompareTask());

        return found;
}

size_t TaskQueue::capacity(void) const {
        return heap.capacity();
}

void TaskQueue::reserve(size_t capacity) {
        heap.reserve(capacity);
}

void TaskQueue::setLog(Log::shared log) {
        this->libutilxx__log = log;
}

bool TaskQueue::isReady(void) const {
        if (empty()) {
                return false;
        }

        return heap.front()->isReady();
}

void TaskQueue::print(void) const {
        Task::shared task;
        timeval now, timer, delay;

        if (libutilxx__log == NULL) {
                return;
        }

        gettimeofday(&now, NULL);
        INFO("[TaskQueue Heap] - now %10lu.%06lu, %zd queued", now.tv_sec, now.tv_usec, heap.size());

        if (heap.empty()) {
                INFO(" heap => empty");

                return;
        }

        for (auto it = heap.begin(); it != heap.end(); ++it) {
                timer = (*it)->getTimer();
                if (timerisset(&timer)) {
                        /* The output comes out very wrong when timersub() is used. This output is odd, but accurate. */
                        delay.tv_sec = timer.tv_sec - now.tv_sec;
                        delay.tv_usec = timer.tv_usec - now.tv_usec;
                        if (delay.tv_sec > 0 && delay.tv_usec < 0) { /* Only preform the subtraction when tv_sec is > 0. */
                                --delay.tv_sec;
                                delay.tv_usec += 1000000;
                        }

                        INFO(" heap => task(%10p) obj: %10p pri: %3d timer: %10lu.%06lu delay: %4d.%d", (*it).get(), (*it)->getObject(), (*it)->getPriority(), timer.tv_sec, timer.tv_usec, (signed) delay.tv_sec, (signed) delay.tv_usec);
                } else {
                        INFO(" heap => task(%10p) obj: %10p pri: %3d timer: %10lu.%06lu delay: -inf", (*it).get(), (*it)->getObject(), (*it)->getPriority(), timer.tv_sec, timer.tv_usec);
                }
        }
}

/*
** vim: noet ts=3 sw=3
*/