Queue.hxx

Go to the documentation of this file.

 
#ifndef _UTILS_QUEUE_HXX_
#define _UTILS_QUEUE_HXX_
 
#include <cstdint>
#include <cstdlib>
#include <cstdarg>
 
#include "openmrn_features.h"
#include "executor/Executable.hxx"
#include "executor/Notifiable.hxx"
#include "os/OS.hxx"
#include "utils/Atomic.hxx"
#include "utils/MultiMap.hxx"
#include "utils/QMember.hxx"
#include "utils/macros.h"
 
namespace openlcb
{
class AsyncIfTest;
}
 
 
struct Result
{
    Result()
        : item(NULL)
        , index(0)
        {
        }
 
    Result(QMember *item, unsigned index)
        : item(item)
        , index(index)
        {
        }
 
    ~Result()
        {
        }
 
    QMember *item;  
    unsigned index; 
};
 
class QResultHolder
{
public:
};
 
class Q : private Atomic
{
public:
    Q()
        : head(NULL)
        , tail(NULL)
        , count(0)
    {
    }
 
    ~Q()
    {
    }
 
    Atomic *lock()
    {
        return this;
    }
 
    void insert(QMember *item, unsigned index = 0)
    {
        AtomicHolder h(this);
        HASSERT(item->next == nullptr);
        HASSERT(item != tail);
        if (head == NULL)
        {
            head = tail = item;
        }
        else
        {
            tail->next = item;
            tail = item;
        }
        item->next = NULL;
        ++count;
    }
 
    void insert_locked(QMember *item, unsigned index = 0)
    {
        HASSERT(item->next == nullptr);
        HASSERT(item != tail);
        if (head == NULL)
        {
            head = tail = item;
        }
        else
        {
            tail->next = item;
            tail = item;
        }
        item->next = NULL;
        ++count;
    }
 
    QMember *next(unsigned index)
    {
        return next().item;
    }
 
    Result next()
    {
        AtomicHolder h(this);
        return next_locked();
    }
 
    Result next_locked()
    {
        if (head == NULL)
        {
            return Result();
        }
        --count;
        QMember *qm = head;
        if (head == tail)
        {
            tail = NULL;
        }
        head = (qm->next);
        qm->next = NULL;
 
        return Result(qm, 0);
    }
    
    size_t pending(unsigned index)
    {
        return pending();
    };
 
    size_t pending()
    {
        return count;
    }
 
    bool empty(unsigned index)
    {
        return empty();
    }
 
    bool empty()
    {
        return (head == NULL);
    }
 
private:
    QMember *head;
 
    QMember *tail;
 
    size_t count;
 
    DISALLOW_COPY_AND_ASSIGN(Q);
};
 
class QAsync
{
public:
    QAsync()
        : waiting(true)
    {
    }
 
    ~QAsync()
    {
    }
 
    void insert(QMember *item, unsigned index = 0)
    {
        Executable *executable = NULL;
        {
            AtomicHolder h(impl_.lock());
            if (waiting)
            {
                if (impl_.empty())
                {
                    waiting = false;
                    impl_.insert(item);
                }
                else
                {
                    executable = static_cast<Executable *>(impl_.next().item);
                }
            }
            else
            {
                impl_.insert(item);
            }
        }
        if (executable)
        {
            executable->alloc_result(item);
        }
    }
 
    void next_async(Executable *flow)
    {
        QMember *qm = NULL;
        {
            AtomicHolder h(impl_.lock());
            if (waiting)
            {
                impl_.insert(flow);
            }
            else
            {
                qm = impl_.next(0);
                if (qm == NULL)
                {
                    impl_.insert(flow);
                    waiting = true;
                }
            }
        }
        if (qm)
        {
            flow->alloc_result(qm);
        }
    }
 
    QMember *next(unsigned index)
    {
        return next().item;
    }
 
    Result next()
    {
        AtomicHolder h(impl_.lock());
        return next_locked();
    }
 
    Result next_locked()
    {
        return waiting ? Result() : impl_.next();
    }
 
    size_t pending(unsigned index)
    {
        return pending();
    };
 
    size_t pending()
    {
        AtomicHolder h(impl_.lock());
        return waiting ? 0 : impl_.pending();
    }
 
    bool empty(unsigned index)
    {
        return empty();
    }
 
    bool empty()
    {
        AtomicHolder h(impl_.lock());
        return waiting ? true : impl_.empty();
    }
 
private:
    bool waiting;
 
    Q impl_;
 
    DISALLOW_COPY_AND_ASSIGN(QAsync);
};
 
template <class T> class TypedQAsync : public QAsync
{
public:
    T *next_blocking()
    {
        return BlockingWait(this).result();
    }
 
    void typed_insert(T* entry) {
        insert(entry);
    }
 
 
private:
    class BlockingWait : public Executable
    {
    public:
        BlockingWait(TypedQAsync<T> *parent)
        {
            parent->next_async(this);
            n_.wait_for_notification();
        }
 
        T *result()
        {
            return result_;
        }
 
    private:
        void alloc_result(QMember *item) OVERRIDE
        {
            result_ = static_cast<T *>(item);
            n_.notify();
        }
 
        void run() OVERRIDE
        {
            DIE("Unexpected call to Run() in BlockingWait");
        }
 
    private:
        SyncNotifiable n_;
        T *result_;
    };
};
 
template <unsigned ITEMS> class QList
{
public:
    QList()
        : list()
    {
    }
 
    ~QList()
    {
    }
 
    typedef ::Result Result;
 
    Atomic *lock()
    {
        return list[0].lock();
    }
 
    void insert(QMember *item, unsigned index)
    {
        AtomicHolder h(lock());
        insert_locked(item, index);
    }
 
    void insert_locked(QMember *item, unsigned index)
    {
        if (index >= ITEMS)
        {
            index = ITEMS - 1;
        }
        list[index].insert_locked(item);
    }
 
    QMember *next(unsigned index)
    {
        AtomicHolder h(lock());
        return list[index].next_locked().item;
    }
 
    Result next()
    {
        AtomicHolder h(lock());
        return next_locked();
    }
 
    Result next_locked()
    {
        for (unsigned i = 0; i < ITEMS; ++i)
        {
            QMember *result = list[i].next_locked().item;
            if (result)
            {
                return Result(result, i);
            }
        }
        return Result();
    }
 
    size_t pending(unsigned index)
    {
        return list[index].pending();
    }
 
    size_t pending()
    {
        // @todo we probably don't want to keep locking and unlocking all of
        // the queues here. There should probably be only one lock for this
        // entire iteration.
        size_t result = 0;
        for (unsigned i = 0; i < ITEMS; ++i)
        {
            result += list[i].pending();
        }
        return result;
    }
 
    size_t size()
    {
        return pending();
    }
 
    bool empty(unsigned index)
    {
        return list[index].empty();
    }
 
    bool empty()
    {
        AtomicHolder h(lock());
        for (unsigned i = 0; i < ITEMS; ++i)
        {
            if (!list[i].empty())
            {
                return false;
            }
        }
        return true;
    }
 
private:
    Q list[ITEMS];
 
    DISALLOW_COPY_AND_ASSIGN(QList);
};
 
template<unsigned items> using QListProtected = QList<items>;
 
 
#if 0
template <class T> class QueueWait : public Q <T>, public OSSem
{
public:
    QueueWait()
        : Q<T>(),
          OSSem(0)
    {
    }
 
    ~QueueWait()
    {
    }
 
    void insert(T *item)
    {
        Q<T>::insert(item);
        post();
    }
 
    T *next()
    {
        T *result = Q<T>::next();
        if (result != NULL)
        {
            /* decrement semaphore */
            OSSem::wait();
        }
        return result;
    }
 
    T *wait()
    {
        OSSem::wait();
        T *result = Q<T>::next();
        if(result == NULL)
        {
            errno = EINTR;
        }
        return result;
    }
 
    T *timedwait(long long timeout)
    {
        if (OSSem::timedwait(timeout) != 0)
        {
            errno = ETIMEDOUT;
            return NULL;
        }
 
        T *result = Q<T>::next();
        if (result == NULL)
        {
            errno = EINTR;
        }
        return result;
    }
 
    void wakeup()
    {
        post();
    }
 
private:
 
    DISALLOW_COPY_AND_ASSIGN(QueueWait);
};
 
template <class T> class QueueProtectedWait : public QProtected <T>, public OSSem
{
public:
    QueueProtectedWait()
        : QProtected<T>(),
          OSSem(0)
    {
    }
 
    ~QueueProtectedWait()
    {
    }
 
    void insert(T *item)
    {
        QProtected<T>::insert(item);
        post();
    }
 
#if OPENMRN_FEATURE_RTOS_FROM_ISR
    void insert_from_isr(T *item)
    {
        QProtected<T>::insert_locked(item);
        post_from_isr();
    }
#endif // OPENMRN_FEATURE_RTOS_FROM_ISR
 
    T *next()
    {
        T *result = QProtected<T>::next();
        if (result != NULL)
        {
            /* decrement semaphore */
            OSSem::wait();
        }
        return result;
    }
 
    T *wait()
    {
        OSSem::wait();
        T *result = QProtected<T>::next();
        if(result == NULL)
        {
            errno = EINTR;
        }
        return result;
    }
 
    T *timedwait(long long timeout)
    {
        if (OSSem::timedwait(timeout) != 0)
        {
            errno = ETIMEDOUT;
            return NULL;
        }
 
        T *result = QProtected<T>::next();
        if (result == NULL)
        {
            errno = EINTR;
        }
        return result;
    }
 
    void wakeup()
    {
        post();
    }
 
private:
    DISALLOW_COPY_AND_ASSIGN(QueueProtectedWait);
};
 
#endif
 
template <unsigned items>
class QListProtectedWait : public QListProtected<items>, public OSSem
{
public:
    QListProtectedWait()
        : QListProtected<items>()
        , OSSem(0)
    {
    }
 
    ~QListProtectedWait()
    {
    }
 
    void insert(QMember *item, unsigned index)
    {
        QListProtected<items>::insert(item, index);
        post();
    }
 
#if OPENMRN_FEATURE_RTOS_FROM_ISR
    void insert_from_isr(QMember *item, unsigned index)
    {
        int woken = 0;
        QListProtected<items>::insert_locked(item, index);
        this->post_from_isr(&woken);
    }
#endif // OPENMRN_FEATURE_RTOS_FROM_ISR
 
    typedef typename QListProtected<items>::Result Result;
 
    Result next()
    {
        Result result = QListProtected<items>::next();
        if (result.item != NULL)
        {
            /* decrement semaphore */
            OSSem::wait();
        }
        return result;
    }
 
    Result wait()
    {
        OSSem::wait();
        Result result = QListProtected<items>::next();
        if (result.item == NULL)
        {
            errno = EINTR;
        }
        return result;
    }
 
#if OPENMRN_FEATURE_SEM_TIMEDWAIT
    Result timedwait(long long timeout)
    {
        if (OSSem::timedwait(timeout) != 0)
        {
            errno = ETIMEDOUT;
            return {NULL, 0};
        }
 
        Result result = QListProtected<items>::next();
        if (result.item == NULL)
        {
            errno = EINTR;
        }
        return result;
    }
#endif
 
    void wakeup()
    {
        post();
    }
 
private:
    DISALLOW_COPY_AND_ASSIGN(QListProtectedWait);
};
 
#endif /* _UTILS_QUEUE_HXX_ */