ScheduledQueue.hxx

Go to the documentation of this file.

 
#ifndef _UTILS_SCHEDULEDQUEUE_HXX_
#define _UTILS_SCHEDULEDQUEUE_HXX_
 
#include "os/OS.hxx"
#include "utils/Fixed16.hxx"
#include "utils/Queue.hxx"
#include "utils/logging.h"
 
class ScheduledQueue
{
public:
    ScheduledQueue(unsigned num_bands, const Fixed16 *strides)
        : numBands_(num_bands)
        , bands_(new Band[num_bands])
    {
        for (unsigned i = 0; i < numBands_; i++)
        {
            bands_[i].stride_ = strides[i];
            bands_[i].currentToken_ -= strides[i];
        }
    }
 
    ~ScheduledQueue()
    {
        delete[] bands_;
    }
 
    Result next()
    {
        OSMutexLock h(lock());
        return next_locked();
    }
 
    Result next_locked()
    {
        if (!numPending_)
        {
            // Empty queue.
            return Result(0, 0);
        }
        // Execute the priority based scheduling algorithm.
        for (unsigned i = 0; i < numBands_; ++i)
        {
            bands_[i].currentToken_ += bands_[i].stride_;
            if (bands_[i].currentToken_.trunc() >= 1)
            {
                Result ret = bands_[i].queue_.next_locked();
                if (ret.item)
                {
                    ret.index = i;
                    --numPending_;
                    bands_[i].currentToken_ -= 1;
                    return ret;
                }
                else
                {
                    // This queue has a token but is empty. We remove
                    // fractional tokens and keep searching onwards in the
                    // priorities.
                    bands_[i].currentToken_ = 1;
                }
            }
        }
        // Fallen off at the end. We go backwards to find any queue with
        // nonempty members.
        for (int i = numBands_ - 1; i >= 0; --i)
        {
            if (!bands_[i].queue_.empty())
            {
                Result ret = bands_[i].queue_.next_locked();
                bands_[i].currentToken_ = 0;
                ret.index = i;
                --numPending_;
                return ret;
            }
        }
        DIE("Unexpected nonempty queue");
        return Result(0, 0);
    }
 
    OSMutex *lock()
    {
        return &lock_;
    }
 
    void insert(QMember *item, unsigned prio)
    {
        OSMutexLock h(lock());
        return insert_locked(item, prio);
    }
 
    void insert_locked(QMember *item, unsigned prio)
    {
        HASSERT(prio < numBands_);
        ++numPending_;
        bands_[prio].queue_.insert_locked(item);
    }
 
    size_t pending(unsigned prio)
    {
        HASSERT(prio < numBands_);
        return bands_[prio].queue_.pending();
    };
 
    size_t pending() const
    {
        return numPending_;
    }
 
    bool empty() const
    {
        return numPending_ == 0;
    }
 
    unsigned num_prio() const
    {
        return numBands_;
    }
 
private:
    struct Band
    {
        Q queue_;
        Fixed16 stride_ {0};
        Fixed16 currentToken_ {1, 0};
    };
 
    OSMutex lock_;
 
    unsigned numBands_;
 
    unsigned numPending_ {0};
 
    Band *bands_;
};
 
#endif // _UTILS_SCHEDULEDQUEUE_HXX_