queues.c

Go to the documentation of this file.

#include <stdlib.h>
#include <string.h>
#include <assert.h>

#include <scheduler/process.h>
#include <core/core.h>
#include <arch/atomic.h>
#include <arch/thread.h>
#include <datatypes/list.h>
#include <datatypes/msgchannel.h>
#include <queues/queues.h>
#include <mm/state.h>
#include <mm/mm.h>
#include <scheduler/scheduler.h>
#include <communication/communication.h>
#include <communication/gvt.h>
#include <statistics/statistics.h>
#include <gvt/gvt.h>

simtime_t next_event_timestamp(struct lp_struct *lp)
{
    msg_t *evt;

    // The bound can be NULL in the first execution or if it has gone back
    if (unlikely(lp->bound == NULL && !list_empty(lp->queue_in))) {
        return list_head(lp->queue_in)->timestamp;
    } else {
        evt = list_next(lp->bound);
        if (likely(evt != NULL)) {
            return evt->timestamp;
        }
    }

    return INFTY;

}

msg_t *advance_to_next_event(struct lp_struct *lp)
{
    if (likely(list_next(lp->bound) != NULL)) {
        lp->bound = list_next(lp->bound);
    } else {
        return NULL;
    }

    return lp->bound;
}

void insert_bottom_half(msg_t * msg)
{
    struct lp_struct *lp = find_lp_by_gid(msg->receiver);

    validate_msg(msg);

    insert_msg(lp->bottom_halves, msg);
#ifdef HAVE_PREEMPTION
    update_min_in_transit(lp->worker_thread, msg->timestamp);
#endif
}

void process_bottom_halves(void)
{
    struct lp_struct *receiver;

    msg_t *msg_to_process;
    msg_t *matched_msg;

    foreach_bound_lp(lp) {

        while ((msg_to_process = get_msg(lp->bottom_halves)) != NULL) {
            receiver = find_lp_by_gid(msg_to_process->receiver);

            // Sanity check
            if (unlikely
                (msg_to_process->timestamp < get_last_gvt()))
                rootsim_error(true,
                          "The impossible happened: I'm receiving a message before the GVT\n");

            // Handle control messages
            if (unlikely(!receive_control_msg(msg_to_process))) {
                msg_release(msg_to_process);
                continue;
            }

            switch (msg_to_process->message_kind) {

                // It's an antimessage
            case negative:

                statistics_post_data(receiver, STAT_ANTIMESSAGE, 1.0);

                // Find the message matching the antimessage
                matched_msg = list_tail(receiver->queue_in);
                while (matched_msg != NULL
                       && matched_msg->mark !=
                       msg_to_process->mark) {
                    matched_msg = list_prev(matched_msg);
                }

                // Sanity check
                if (unlikely(matched_msg == NULL)) {
                    rootsim_error(false,
                              "LP %d Received an antimessage, but no such mark has been found!\n",
                              receiver->gid.to_int);
                    dump_msg_content(msg_to_process);
                    rootsim_error(true, "Aborting...\n");
                }
                // If the matched message is in the past, we have to rollback
                if (matched_msg->timestamp <= lvt(receiver)) {

                    receiver->bound = list_prev(matched_msg);
                    while ((receiver->bound != NULL)
                        && D_EQUAL(receiver->bound->timestamp, msg_to_process->timestamp)) {
                        receiver->bound = list_prev(receiver->bound);
                    }
                    
                    receiver->state = LP_STATE_ROLLBACK;
                }
#ifdef HAVE_MPI
                register_incoming_msg(msg_to_process);
#endif

                // Delete the matched message
                list_delete_by_content(receiver->queue_in,
                               matched_msg);
                msg_release(matched_msg);

                break;

                // It's a positive message
            case positive:

                // A positive message is directly placed in the queue
                list_insert(receiver->queue_in, timestamp,
                        msg_to_process);

                // Check if we've just inserted an out-of-order event.
                // Here we check for a strictly minor timestamp since
                // the queue is FIFO for same-timestamp events. Therefore,
                // A contemporaneous event does not cause a causal violation.
                if (msg_to_process->timestamp < lvt(receiver)) {

                    receiver->bound = list_prev(msg_to_process);
                    while ((receiver->bound != NULL)
                           && D_EQUAL(receiver->bound->timestamp, msg_to_process->timestamp)) {
                        receiver->bound = list_prev(receiver->bound);
                    }

                    receiver->state = LP_STATE_ROLLBACK;
                }
#ifdef HAVE_MPI
                register_incoming_msg(msg_to_process);
#endif
                break;

                // It's a control message
            case control:

                // Check if it is an anti control message
                if (!anti_control_message(msg_to_process)) {
                    msg_release(msg_to_process);
                    continue;
                }

                break;

            default:
                rootsim_error(true, "Received a message which is neither positive nor negative. Aborting...\n");
            }
        }
    }

    // We have processed all in transit messages.
    // Actually, during this operation, some new in transit messages could
    // be placed by other threads. In this case, we loose their presence.
    // This is not a correctness error. The only issue could be that the
    // preemptive scheme will not detect this, and some events could
    // be in fact executed out of order.
#ifdef HAVE_PREEMPTION
    reset_min_in_transit(local_tid);
#endif
}

unsigned long long generate_mark(struct lp_struct *lp)
{
    unsigned long long k1 = (unsigned long long)lp->gid.to_int;
    unsigned long long k2 = lp->mark++;

    return (unsigned long long)(((k1 + k2) * (k1 + k2 + 1) / 2) + k2);
}