msg_queue.c

Go to the documentation of this file.

 
#include <datatypes/msg_queue.h>
 
#include <core/core.h>
#include <core/sync.h>
#include <datatypes/heap.h>
#include <lp/lp.h>
#include <mm/msg_allocator.h>
 
#include <stdalign.h>
 
struct msg_queue {
        alignas(CACHE_LINE_SIZE) spinlock_t lck;
        binary_heap(struct lp_msg *) q;
};
 
static struct msg_queue *queues;
 
#define mqueue(from, to) (&queues[to * n_threads + from])
 
void msg_queue_global_init(void)
{
        queues = mm_alloc(n_threads * n_threads * sizeof(*queues));
}
 
void msg_queue_init(void)
{
        rid_t i = n_threads;
        while(i--) {
                heap_init(mqueue(i, rid)->q);
                spin_init(&(mqueue(i, rid)->lck));
        }
}
 
void msg_queue_fini(void)
{
        rid_t i = n_threads;
        while(i--) {
                struct msg_queue *this_q = mqueue(i, rid);
                array_count_t j = heap_count(this_q->q);
                while(j--) {
                        struct lp_msg *msg = heap_items(this_q->q)[j];
                        if(!(atomic_load_explicit(&msg->flags,
                                memory_order_relaxed) & MSG_FLAG_PROCESSED))
                                msg_allocator_free(msg);
                }
                heap_fini(this_q->q);
        }
}
 
void msg_queue_global_fini(void)
{
        mm_free(queues);
}
 
struct lp_msg *msg_queue_extract(void)
{
        rid_t i = n_threads;
        struct msg_queue *bid_q = mqueue(rid, rid);
        struct lp_msg *msg = heap_count(bid_q->q) ? heap_min(bid_q->q) : NULL;
 
        while(i--) {
                struct msg_queue *this_q = mqueue(i, rid);
                if(!spin_trylock(&this_q->lck))
                        continue;
 
                if (heap_count(this_q->q) &&
                    (!msg || msg_is_before(heap_min(this_q->q), msg))) {
                        msg = heap_min(this_q->q);
                        bid_q = this_q;
                }
                spin_unlock(&this_q->lck);
        }
 
        spin_lock(&bid_q->lck);
 
        if(likely(heap_count(bid_q->q)))
                msg = heap_extract(bid_q->q, msg_is_before);
        else
                msg = NULL;
 
        spin_unlock(&bid_q->lck);
        return msg;
}
 
simtime_t msg_queue_time_peek(void)
{
        const rid_t t_cnt = n_threads;
        simtime_t t_min = SIMTIME_MAX;
        bool done[t_cnt];
        memset(done, 0, sizeof(done));
 
        for(rid_t i = 0, r = t_cnt; r; i = (i + 1) % t_cnt){
                if(done[i])
                        continue;
 
                struct msg_queue *this_q = mqueue(i, rid);
                if(!spin_trylock(&this_q->lck))
                        continue;
 
                done[i] = true;
                --r;
                if (heap_count(this_q->q) && t_min >
                                             heap_min(this_q->q)->dest_t) {
                        t_min = heap_min(this_q->q)->dest_t;
                }
                spin_unlock(&this_q->lck);
        }
 
        return t_min;
}
 
void msg_queue_insert(struct lp_msg *msg)
{
        rid_t dest_rid = lid_to_rid(msg->dest);
        struct msg_queue *this_q = mqueue(rid, dest_rid);
        spin_lock(&this_q->lck);
        heap_insert(this_q->q, msg_is_before, msg);
        spin_unlock(&this_q->lck);
}