wnd.c

Go to the documentation of this file.

#ifdef HAVE_MPI

#include <communication/communication.h>
#include <communication/wnd.h>
#include <communication/mpi.h>

static outgoing_queue *outgoing_queues;

static int n_queues = 0;


static size_t outgoing_queues_size(void)
{
    int i;
    size_t size = 0;
    for (i = 0; i < n_queues; i++) {
        size += list_sizeof(outgoing_queues[i].queue);
    }
    return size;
}


void outgoing_window_init(void)
{
    n_queues = n_ker;
    outgoing_queues = rsalloc(n_queues * sizeof(outgoing_queue));
    outgoing_queue *oq;
    int i;
    for (i = 0; i < n_queues; i++) {
        oq = outgoing_queues + i;
        spinlock_init(&(oq->lock));
        oq->queue = new_list(outgoing_msg);
    }
}


void outgoing_window_finalize(void)
{
    prune_outgoing_queues();
    size_t pending_out_msgs = outgoing_queues_size();

    if (unlikely(pending_out_msgs > 0)) {
        rootsim_error(true, "Outgoing queues not empty on exit: %zu\n", outgoing_queues_size());
    }

    // TODO: release also lists allocated via new_list in outgoing_window_init()
    rsfree(outgoing_queues);
}


outgoing_msg *allocate_outgoing_msg(void)
{
    return rsalloc(sizeof(outgoing_msg));
}


static inline bool is_msg_delivered(outgoing_msg *msg)
{
    return is_request_completed(&(msg->req));
}


void store_outgoing_msg(outgoing_msg *out_msg, unsigned int dest_kid)
{
    outgoing_queue *oq = &outgoing_queues[dest_kid];

    spin_lock(&(oq->lock));
    list_insert_tail(oq->queue, out_msg);
    spin_unlock(&(oq->lock));
}


static int prune_outgoing_queue(outgoing_queue *oq)
{
    int pruned = 0;

    spin_lock(&(oq->lock));

    outgoing_msg *msg = list_head(oq->queue);

    // check all the outgoing messages in the queue starting from the
    // head (the entry with the minimum timestamp) and delete them
    // if they have been already delivered
    while (msg != NULL && is_msg_delivered(msg)) {
        msg_release(msg->msg);

        list_delete_by_content(oq->queue, msg);
        pruned++;
        msg = list_head(oq->queue);
    }

    spin_unlock(&(oq->lock));

    return pruned;
}


int prune_outgoing_queues(void)
{
    int i;
    int pruned = 0;
    for (i = 0; i < n_queues; i++) {
        pruned += prune_outgoing_queue(outgoing_queues + i);
    }
    return pruned;
}

#endif