Home Research Papers Posters Manuals Talks Download People Help Internal

conv-core/msgq.h

Go to the documentation of this file.

#ifndef MSG_Q_H
#define MSG_Q_H

#include <deque>
#include <queue>
#include <ostream>
#include <functional>
#include <limits>

#include <unordered_map>

namespace conv {

// Some day, messages may be handled as something other than void* within the runtime.
// Prepare for that day, while enhancing readability today.
typedef void msg_t;

#if !CMK_NO_MSG_PRIOS
template <typename P = int>
class msgQ
{
    public:
        typedef P prio_t;

        msgQ(): qSize(0) {}

        void enq(const msg_t *msg, const prio_t &prio = prio_t(), const bool isFifo = true);

        const msg_t* deq();

        inline size_t size() const { return qSize; }
        inline size_t max_size() const { return std::numeric_limits<size_t>::max(); }

        inline bool empty() const { return (0 == qSize); }

        void enumerate(msg_t **first, msg_t **last) const;

        friend std::ostream& operator<< (std::ostream &out, const msgQ &q)
        {
            out <<"\nmsgQ[" << q.qSize << "]:";
            out<<"\n";
            return out;
        }

    private:
#if CMK_RANDOMIZED_MSGQ
        std::vector<const msg_t*> randQ;
#endif

        typedef std::deque<const msg_t*> bkt_t;
        typedef typename std::pair<prio_t, bkt_t*> prioidx_t;
        typedef typename std::unordered_map<prio_t, bkt_t> bktmap_t;

        size_t qSize;
        bktmap_t msgbuckets;
        std::priority_queue<prioidx_t, std::vector<prioidx_t>, std::greater<prioidx_t> > prioQ;
};



template <typename P>
void msgQ<P>::enq(const msg_t *msg, const prio_t &prio, const bool isFifo)
{
    #if ! CMK_RANDOMIZED_MSGQ
    // Find / create the bucket holding msgs of this priority
    bkt_t &bkt = msgbuckets[prio];
    // If this deq is empty, insert corresponding priority into prioQ
    if (bkt.empty())
        prioQ.push( std::make_pair(prio, &bkt) );
    // Enq msg either at front or back of deq
    isFifo ? bkt.push_back(msg) : bkt.push_front(msg);
    #else
    randQ.push_back(msg);
    #endif

    // Increment the total number of msgs in this container
    qSize++;
}



#if ! CMK_RANDOMIZED_MSGQ // charm NOT built with a randomized queue

// Iterative applications typically have a set of msg priority values that just
// repeat over time. However, the arrival pattern of messages at a given PE is
// unknown. Within a single iteration, the size of a bucket holding msgs of a
// given priority may fluctuate. It may become empty as these messages are
// delivered. Instead of deleting the bucket every time it becomes empty (which
// could be several times in a single iteration), we choose to leave it around
// because it will eventually get filled again. This avoids the costs of
// repeatedly constructing / destroying buckets (dequeues).
//
// However, for non-iterative applications, a given priority value may not
// recur later in the execution. Not deleting buckets may increase container
// sizes, degrading performance over time. Hence, buckets should be deleted
// whenever they become empty (or with a small timeout).
//
// The choice between the two execution modes should be a compile-time
// decision. Ideally, this should be a decision during application compilation
// and be specified programmatically by the application (think template
// parameter). However, since this Q is buried so deep within charm, the
// decision has to be made during charm compilation.
//
// Assume by default that the application is iterative
#define CMK_ITERATIVE_MSG_PRIOS 1

template <typename P>
const msg_t* msgQ<P>::deq()
{
    if (empty())
        return NULL;
    // Find the bucket holding the highest priority msgs
    bkt_t &bkt = * prioQ.top().second;
    // Pop msg from the front of the deque
    const msg_t *msg = bkt.front();
    bkt.pop_front();
    // If all msgs in the bucket have been consumed
    if (bkt.empty())
    {
        #if ! CMK_ITERATIVE_MSG_PRIOS
        // remove the empty bucket from the collection of buckets
        msgbuckets.erase( prioQ.top().first );
        #endif
        // pop corresponding priority from the priority Q
        prioQ.pop();
    }
    // Decrement the total number of msgs in this container
    qSize--;
    return msg;
}

#else // If charm is built with a randomized msg queue

template <typename P>
const msg_t* msgQ<P>::deq()
{
    if (empty())
        return NULL;

#if defined(_WIN64) || defined(_WIN32)
    long rnd = rand() % randQ.size();
#else
    long rnd = lrand48() % randQ.size();
#endif
    const msg_t *msg = randQ[rnd];
    randQ[rnd] = randQ[randQ.size()-1];
    randQ.pop_back();

    // Decrement the total number of msgs in this container
    qSize--;
    return msg;
}

#endif // CMK_RANDOMIZED_MSGQ

template <typename P>
void msgQ<P>::enumerate(msg_t **first, msg_t **last) const
{
    if (first >= last)
        return;

    msg_t **ptr = first;

#if !CMK_RANDOMIZED_MSGQ
    for (typename bktmap_t::const_iterator bktitr = msgbuckets.begin();
            ptr != last && bktitr != msgbuckets.end(); bktitr++)
    {
        bkt_t::const_iterator msgitr = bktitr->second.begin();
        while (ptr != last && msgitr != bktitr->second.end())
            *ptr++ = (msg_t*) *msgitr++;
    }
#else
    for (size_t i = 0; i < randQ.size() && ptr != last; ++i, ++ptr)
      *ptr = (msg_t*) randQ[i];
#endif
}

#else //CMK_NO_MSG_PRIOS
template <typename P = int>
class msgQ
{
    public:
        typedef P prio_t;
        msgQ() {}
        inline size_t size() const { return bkt.size(); }
        inline size_t max_size() const { return bkt.max_size(); }
        inline bool empty() const { return bkt.empty(); }
        void enumerate(msg_t **first, msg_t **last) const {}
        friend std::ostream& operator<< (std::ostream &out, const msgQ &q) { return out; }

        inline void enq(const msg_t *msg, const prio_t &prio = prio_t(), const bool isFifo = true) {
          bkt.push_back(msg);
        }

        inline const msg_t* deq() {
          if (bkt.empty())
            return NULL;
          const msg_t *msg = bkt.front();
          bkt.pop_front();
          return msg;
        }

    private:
        std::deque<const msg_t*> bkt;
};
#endif //CMK_NO_MSG_PRIOS

} // end namespace conv

#endif // MSG_Q_H

---