prt_thread.c

Go to the documentation of this file.

#include "prt_thread.h"


prt_thread_t* prt_thread_new(int rank, int core, int agent_rank)
{
    // 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->agent_rank = agent_rank;
    thread->finished = 0;

    // Create the list of VDPs.
    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;
    // Set the trace rank, depending on the existence of the proxy.
    int trace_rank = thread->vsa->proxy == NULL ? thread->rank : thread->rank+1;

    // Call the thread warmup function if set.
    if (thread->vsa->thread_warmup_func != NULL)
        thread->vsa->thread_warmup_func();

    // Barrier the threads.
    svg_trace_start_cpu(trace_rank);
    pthread_barrier_wait(&thread->vsa->barrier);
    svg_trace_stop_cpu(trace_rank, -Azure);
    double start = get_time_of_day();

    icl_list_t *node;
    // WHILE the list of VDPs is not empty.
    while ((node = icl_list_first(thread->vdps)) != NULL) {
        // REPEAT.
        do {
            prt_vdp_t *vdp = (prt_vdp_t*)node->data;
            // IF aggressive scheduling.
            if  (thread->vsa->config->vdp_scheduling
                == PRT_VDP_SCHEDULING_AGGRESSIVE) {
                // WHILE the inputs are ready AND the counter is NOT zero.
                while (vdp->counter > 0 && prt_vdp_ready(vdp)) {
                    // Fire the VDP and decrement the counter.
                    svg_trace_start_cpu(trace_rank);
                    vdp->function(vdp);
                    svg_trace_stop_cpu(trace_rank, vdp->color);
                    vdp->counter--;
                }
            }
            else if (thread->vsa->config->vdp_scheduling
                == PRT_VDP_SCHEDULING_LAZY) {
                // IF the inputs are ready AND the counter is NOT zero.
                if (vdp->counter > 0 && prt_vdp_ready(vdp)) {
                    // Fire the VDP and decrement the counter.
                    svg_trace_start_cpu(trace_rank);
                    vdp->function(vdp);
                    svg_trace_stop_cpu(trace_rank, vdp->color);
                    vdp->counter--;
                }
            }
            icl_list_t *prev = node;
            node = icl_list_next(thread->vdps, node);
            // IF the counter is 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 is reached.
    }
    // Signal completion.
    thread->finished = 1;

    // Barrier the threads.
    svg_trace_start_cpu(trace_rank);
    pthread_barrier_wait(&thread->vsa->barrier);
    svg_trace_stop_cpu(trace_rank, -Azure);

    // Return the time between the end of the first thread barrier
    // and the end of the second thread barrier.
    double stop = get_time_of_day();
    thread->time = stop-start;
    return NULL;
}