tw-kp.c

Go to the documentation of this file.

#include <ross.h>

void
tw_kp_onpe(tw_kpid id, tw_pe * pe)
{
        if(id >= g_tw_nkp)
                tw_error(TW_LOC, "ID %d exceeded MAX KPs", id);

        if(g_tw_kp[id])
                tw_error(TW_LOC, "KP already allocated: %lld\n", id);

        g_tw_kp[id] = (tw_kp *) tw_calloc(TW_LOC, "Local KP", sizeof(tw_kp), 1);

        g_tw_kp[id]->id = id;
        g_tw_kp[id]->pe = pe;

#ifdef ROSS_QUEUE_kp_splay
        g_tw_kp[id]->pq = tw_eventpq_create();
#endif
}

void
tw_kp_rollback_to(tw_kp * kp, tw_stime to)
{
        tw_event       *e;
        tw_clock pq_start;

        kp->s_rb_total++;
        // instrumentation
        kp->kp_stats->s_rb_total++;

#if VERIFY_ROLLBACK
        printf("%d %d: rb_to %f, now = %f \n",
               kp->pe->id, kp->id, TW_STIME_DBL(to), TW_STIME_DBL(kp->last_time));
#endif

        while(kp->pevent_q.size && TW_STIME_CMP(kp->pevent_q.head->recv_ts, to) >= 0)
        {
                e = tw_eventq_shift(&kp->pevent_q);

                /*
                 * rollback first
                 */
                tw_event_rollback(e);

                /*
                 * reset kp pointers
                 */
                if (kp->pevent_q.size == 0)
                {
                        kp->last_time = kp->pe->GVT;
                } else
                {
                        kp->last_time = kp->pevent_q.head->recv_ts;
                }

                /*
                 * place event back into priority queue
                 */
                pq_start = tw_clock_read();
                tw_pq_enqueue(kp->pe->pq, e);
                kp->pe->stats.s_pq += tw_clock_read() - pq_start;
        }
}

void
tw_kp_rollback_event(tw_event * event)
{
    tw_event       *e = NULL;
    tw_kp          *kp;
    tw_pe          *pe;
    tw_clock pq_start;

    kp = event->dest_lp->kp;
    pe = kp->pe;

    kp->s_rb_total++;
        kp->s_rb_secondary++;
    // instrumentation
    kp->kp_stats->s_rb_total++;
        kp->kp_stats->s_rb_secondary++;

#if VERIFY_ROLLBACK
        printf("%d %d: rb_event: %f \n", pe->id, kp->id, event->recv_ts);

        if(!kp->pevent_q.size)
                tw_error(TW_LOC, "Attempting to rollback empty pevent_q!");
#endif

        e = tw_eventq_shift(&kp->pevent_q);
        while(e != event)
        {
                kp->last_time = kp->pevent_q.head->recv_ts;
                tw_event_rollback(e);
                pq_start = tw_clock_read();
                tw_pq_enqueue(pe->pq, e);
                pe->stats.s_pq += tw_clock_read() - pq_start;

                e = tw_eventq_shift(&kp->pevent_q);
        }

        tw_event_rollback(e);

        if (0 == kp->pevent_q.size)
                kp->last_time = kp->pe->GVT;
        else
                kp->last_time = kp->pevent_q.head->recv_ts;
}

#ifndef NUM_OUT_MESG
#define NUM_OUT_MESG 2000
#endif
static tw_out*
init_output_messages(tw_kp *kp)
{
    int i;

    tw_out *ret = (tw_out *) tw_calloc(TW_LOC, "tw_out", sizeof(struct tw_out), NUM_OUT_MESG);

    for (i = 0; i < NUM_OUT_MESG - 1; i++) {
        ret[i].next = &ret[i + 1];
        ret[i].owner = kp;
    }
    ret[i].next = NULL;
    ret[i].owner = kp;

    return ret;
}

void
tw_init_kps(tw_pe * me)
{
        tw_kpid i;
    int j;

        for (i = 0; i < g_tw_nkp; i++)
        {
                tw_kp *kp = tw_getkp(i);

                if (kp->pe != me)
                        continue;

                kp->id = i;
        kp->s_nevent_processed = 0;
                kp->s_e_rbs = 0;
                kp->s_rb_total = 0;
                kp->s_rb_secondary = 0;
        if (g_tw_synchronization_protocol == OPTIMISTIC ||
            g_tw_synchronization_protocol == OPTIMISTIC_DEBUG ||
            g_tw_synchronization_protocol == OPTIMISTIC_REALTIME) {
            kp->output = init_output_messages(kp);
        }

        // instrumentation setup
        kp->kp_stats = (st_kp_stats*) tw_calloc(TW_LOC, "KP instrumentation", sizeof(st_kp_stats), 1);
        for (j = 0; j < 3; j++)
            kp->last_stats[j] = (st_kp_stats*) tw_calloc(TW_LOC, "KP instrumentation", sizeof(st_kp_stats), 1);
        }
}

tw_out *
tw_kp_grab_output_buffer(tw_kp *kp)
{
    if (kp->output) {
        tw_out *ret = kp->output;
        kp->output = kp->output->next;
        ret->next = 0;
        return ret;
    }

    return NULL;
}

void
tw_kp_put_back_output_buffer(tw_out *out)
{
    tw_kp *kp = out->owner;

    if (kp->output) {
        out->next = kp->output;
        kp->output = out;
    }
    else {
        kp->output = out;
        kp->output->next = NULL;
    }
}