prt_thread.c

Go to the documentation of this file.

#include "prt_thread.h"


prt_thread_t* prt_thread_new(int rank, int core)
{
    // Allocate the thread.
    prt_thread_t *thread = (prt_thread_t*)malloc(sizeof(prt_thread_t));
    prt_assert(thread != NULL, "malloc failed");

    // Initialize the thread.
    thread->rank = rank;
    thread->core = core;
    thread->finished = 0;

    // Create the list of VDPs.
    // No error checking.
    thread->vdps = icl_list_new();
    prt_assert(thread->vdps != NULL, "icl_list_new failed");

    // Return the thread.
    return thread;
}


void prt_thread_delete(prt_thread_t *thread)
{
    // Destroy the list of VDPs.
    int size = icl_list_size(thread->vdps);
    prt_assert(size == 0, "destroying non-empty list");
    int status = icl_list_destroy(thread->vdps, NULL);
    prt_assert(status == 0, "icl_list_destroy failed");

    // Free the thread.
    free(thread);
}


void* prt_thread_run(void *thrd)
{
    prt_thread_t *thread = (prt_thread_t*)thrd;
    icl_list_t *prev;
    icl_list_t *node;
    icl_list_t *next;
    // WHILE the list of VDPs not empty.
    while ((node = icl_list_first(thread->vdps)) != NULL) {
        // REPEAT.
        do {
            prt_vdp_t *vdp = (prt_vdp_t*)node->data;
            if  (thread->vsa->config->vdp_scheduling == PRT_VDP_SCHEDULING_AGGRESSIVE) {
                // WHILE inputs ready AND counter NOT zero.
                while (vdp->counter > 0 && prt_vdp_ready(vdp)) {
                    // Fire and decrement.
                    svg_trace_start(thread->rank+1);
                    vdp->function(
                        vdp->tuple, vdp->counter,
                        vdp->input, vdp->output,
                        vdp->local_store, thread->vsa->global_store);
                    svg_trace_stop(thread->rank+1, vdp->color);
                    vdp->counter--;
                }
            }
            else if (thread->vsa->config->vdp_scheduling == PRT_VDP_SCHEDULING_LAZY) {
                // IF inputs ready AND counter NOT zero.
                if (vdp->counter > 0 && prt_vdp_ready(vdp)) {
                    // Fire and decrement.
                    svg_trace_start(thread->rank+1);
                    vdp->function(
                        vdp->tuple, vdp->counter,
                        vdp->input, vdp->output,
                        vdp->local_store, thread->vsa->global_store);
                    svg_trace_stop(thread->rank+1, vdp->color);
                    vdp->counter--;
                }
            }
            prev = node;
            node = icl_list_next(thread->vdps, node);
            // IF counter zeroed.
            if (vdp->counter == 0)
                // Remove the VDP from the list
                // Do not destroy the VDP itself.
                icl_list_delete(thread->vdps, prev, NULL);
        }
        while (node != NULL);
        // UNTIL the end of the list.
    }
    // Signal completion.
    thread->finished = 1;
}