prt_vdp.c

Go to the documentation of this file.

#include "prt_vdp.h"


prt_vdp_t* prt_vdp_new(
    int *tuple, int counter,
    prt_vdp_function_t function,
    size_t local_store_size,
    int num_inputs, int num_outputs, int color)
{
    // Check input parameters.
    prt_assert(tuple != NULL, "NULL tuple");
    prt_assert(counter > 0, "counter not larger than zero");
    prt_assert(function != NULL, "NULL VDP function");
    prt_assert(num_inputs >= 0, "negative number of inputs");
    prt_assert(num_outputs >= 0, "negative number of outputs");

    // Allocate the VDP.
    prt_vdp_t *vdp = (prt_vdp_t*)malloc(sizeof(prt_vdp_t));
    prt_assert(vdp != NULL, "malloc failed");

    // Initialize the VDP.
    vdp->vsa = NULL;
    vdp->thread = NULL;
    vdp->device = NULL;
    vdp->tuple = tuple;
    vdp->counter = counter;
    vdp->function = function;
    vdp->color = color;

    // Initialize input channels.
    vdp->num_inputs = num_inputs;
    if (vdp->num_inputs > 0) {
        vdp->input =
            (prt_channel_t**)calloc(vdp->num_inputs, sizeof(prt_channel_t*));
        prt_assert(vdp->input != NULL, "malloc failed");
    }
    // Initialize output channels.
    vdp->num_outputs = num_outputs;
    if (vdp->num_outputs > 0) {
        vdp->output =
            (prt_channel_t**)calloc(vdp->num_outputs, sizeof(prt_channel_t*));
        prt_assert(vdp->output != NULL, "malloc failed");
    }
    // Allocate local store.
    if (local_store_size > 0) {
        vdp->local_store = (void*)malloc(local_store_size);
        prt_assert(vdp->local_store != NULL, "malloc failed");
    }
    // Return the VDP.
    return vdp;
}


void prt_vdp_delete(prt_vdp_t *vdp)
{
    // Check for a NULL VDP.
    prt_assert(vdp != NULL, "NULL VDP");

    // Delete the tuple.
    prt_assert(vdp->tuple != NULL, "NULL tuple");
    prt_tuple_delete(vdp->tuple);

    int i;
    // Delete all input channels.
    for (i = 0; i < vdp->num_inputs; i++) {
        prt_channel_t *channel = vdp->input[i];
        if (channel != NULL)
            prt_channel_delete(channel);
    }
    // Delete dangling output channels.
    for (i = 0; i < vdp->num_outputs; i++) {
        prt_channel_t *channel = vdp->output[i];
        if (channel != NULL)
            // IF a dangling thread channel.
            if (vdp->location == PRT_LOCATION_HOST &&
                channel->dst_node != vdp->thread->vsa->node_rank)
                // Delete the channel.
                prt_channel_delete(channel);
            // ELSE IF a dangling device channel.
            else if (vdp->location == PRT_LOCATION_DEVICE &&
                channel->dst_node != vdp->device->vsa->node_rank)
                // Delete the channel.
                prt_channel_delete(channel);
    }
    // Free array of inputs.
    if (vdp->num_inputs > 0)
        free(vdp->input);

    // Free array of outputs.
    if (vdp->num_outputs > 0)
        free(vdp->output);

    // Free local store.
    if (vdp->local_store != NULL)
        free(vdp->local_store);

    if (vdp->location == PRT_LOCATION_DEVICE) {
        cudaError_t error;
        // Set the VDP's device.
        error = cudaSetDevice(vdp->device->rank);
        prt_assert(error == cudaSuccess, cudaGetErrorString(error));
        // Destroy the VDP's stream.
        error = cudaStreamDestroy(vdp->stream);
        prt_assert(error == cudaSuccess, cudaGetErrorString(error));
    }
    // Free the VDP.
    free(vdp);
}


void prt_vdp_annihilate(prt_vdp_t *vdp)
{
    // Check for a NULL VDP.
    prt_assert(vdp != NULL, "NULL VDP");

    // Delete the tuple.
    prt_assert(vdp->tuple != NULL, "NULL tuple");
    prt_tuple_delete(vdp->tuple);

    int i;
    // Delete all input channels.
    for (i = 0; i < vdp->num_inputs; i++) {
        prt_channel_t *channel = vdp->input[i];
        if (channel != NULL)
            prt_channel_delete(channel);
    }
    // Delete all output channels.
    for (i = 0; i < vdp->num_outputs; i++) {
        prt_channel_t *channel = vdp->output[i];
        if (channel != NULL)
            prt_channel_delete(channel);
    }
    // Free array of inputs.
    if (vdp->num_inputs > 0)
        free(vdp->input);

    // Free array of outputs.
    if (vdp->num_outputs > 0)
        free(vdp->output);

    // Free local store.
    if (vdp->local_store != NULL)
        free(vdp->local_store);

    // Free the VDP.
    free(vdp);
}


void prt_vdp_channel_insert(
    prt_vdp_t *vdp, prt_channel_t *channel,
    prt_channel_direction_t direction, int slot)
{
    // Check input parameters.
    prt_assert(vdp != NULL, "NULL VDP");
    prt_assert(channel != NULL, "NULL channel");
    prt_assert(direction == PRT_INPUT_CHANNEL || direction == PRT_OUTPUT_CHANNEL,
        "undefined direction");
    prt_assert(slot >= 0, "negative slot");

    // IF inserting input channel.
    if (direction == PRT_INPUT_CHANNEL) {
        // Check if channel destination tuple equals VDP's tuple.
        prt_assert(prt_tuple_equal(channel->dst_tuple, vdp->tuple),
            "input channel destination tuple does not match VDP tuple");
        // Check if channel slot empty.
        prt_assert(vdp->input[slot] == NULL,
            "inserting channel in occupied input slot");
        // Check the slot.
        // Link the channel.
        prt_assert(slot >= 0 && slot < vdp->num_inputs, "slot out of range");
        vdp->input[slot] = channel;
        // Set the destination VDP.
        channel->dst_vdp = vdp;
    }
    // ELSE IF inserting output channel.
    else if (direction == PRT_OUTPUT_CHANNEL) {
        // Check if channel source tuple equals VDP's tuple.
        prt_assert(prt_tuple_equal(channel->src_tuple, vdp->tuple),
            "output channel source tuple does not match VDP tuple");
        // Check if channel slot empty.
        prt_assert(vdp->output[slot] == NULL,
            "inserting channel in occupied output slot");
        // Check the slot.
        // Link the channel.
        prt_assert(slot >= 0 && slot < vdp->num_outputs, "slot out of range");
        vdp->output[slot] = channel;
        // Set the source VDP.
        channel->src_vdp = vdp;
    }
}


prt_packet_t *prt_vdp_packet_new(prt_vdp_t *vdp, size_t size, void *data)
{
    // Check input parameters.
    prt_assert(vdp != NULL, "NULL VDP");
    prt_assert(size > 0, "packet size equals zero");
    prt_assert(size <= INT_MAX, "packet size larger than INT_MAX");

    // Test for NULL VDP.
    prt_assert(vdp != NULL, "NULL VDP");

    // IF host VDP.
    if (vdp->location == PRT_LOCATION_HOST)
        // Create host packet.
        return prt_packet_new_host(size, data);

    // IF device VDP.
    if (vdp->location == PRT_LOCATION_DEVICE)
        // Create device packet.
        return prt_packet_new_device(size, data, vdp);

    prt_error("NULL thread and device");
}


prt_packet_t *prt_vdp_packet_new_host_to_device(
    prt_vdp_t *vdp, size_t size, void *data)
{
    // Check input parameters.
    prt_assert(vdp != NULL, "NULL VDP");
    prt_assert(size > 0, "packet size equals zero");
    prt_assert(size <= INT_MAX, "packet size larger than INT_MAX");
    prt_assert(data != NULL, "NULL data pointer");

    // Create a new device packet.
    prt_packet_t *packet = prt_packet_new_device(size, NULL, vdp);

    // Put host to device copy in the VDP's stream.
    svg_trace_start_dma(vdp->stream);
    cudaMemcpyAsync(
        packet->data, data, size, cudaMemcpyHostToDevice, vdp->stream);
    svg_trace_stop_dma(vdp->stream, Silver);

    // Return the packet.
    return packet;
}


void prt_vdp_packet_release(prt_vdp_t *vdp, prt_packet_t *packet)
{
    // Check input parameters.
    prt_assert(vdp != NULL, "NULL VDP");
    prt_assert(packet != NULL, "NULL packet");

    // IF a host packet.
    if (vdp->location == PRT_LOCATION_HOST) {
        // Release immediately.
        prt_packet_release_host(packet);
    }
    else {
        // Put a callback in the VDP's stream.
        prt_callback_release_t *callback =
            prt_callback_release_new(vdp, packet);
        __sync_fetch_and_add(&vdp->vsa->proxy->num_callbacks, 1);
        cudaStreamAddCallback(
            vdp->stream, prt_callback_release_handler, (void*)callback, 0);
    }
}


void prt_vdp_channel_push(prt_vdp_t *vdp, int channel_num, prt_packet_t *packet)
{
    // Check input params.
    prt_assert(vdp != NULL, "NULL VDP");
    prt_assert(channel_num >= 0 && channel_num < vdp->num_outputs,
        "channel number out of range");
    prt_assert(packet != NULL, "NULL packet");
    prt_assert(packet->size <= vdp->output[channel_num]->size,
        "packet size larger than channel size");

    // IF thread VDP.
    if (vdp->location == PRT_LOCATION_HOST)
        // Push from host.
        prt_channel_push_host(vdp, vdp->output[channel_num], packet);
    // ELSE IF device VPD.
    else if (vdp->location == PRT_LOCATION_DEVICE)
        // Push from device.
        prt_channel_push_device(vdp, vdp->output[channel_num], packet);
    // ELSE report error.
    else
        prt_error("NULL thread and device");
}


prt_packet_t *prt_vdp_channel_pop(prt_vdp_t *vdp, int channel_num)
{
    // Check input parameters.
    prt_assert(vdp != NULL, "NULL VDP");
    prt_assert(channel_num >= 0 && channel_num < vdp->num_inputs,
        "channel number out of range");
    prt_assert(vdp->input[channel_num]->active,
        "popping from an inactive channel");

    // Call the channel pop function.
    return prt_channel_pop(vdp->input[channel_num]);
}


void prt_vdp_channel_off(prt_vdp_t *vdp, int channel_num)
{
    // Check input parameters.
    prt_assert(vdp != NULL, "NULL VDP");
    prt_assert(channel_num >= 0 && channel_num < vdp->num_inputs,
        "channel number out of range");

    // Switch off the channel.
    prt_channel_off(vdp->input[channel_num]);
}


void prt_vdp_channel_on(prt_vdp_t *vdp, int channel_num)
{
    // Check input parameters.
    prt_assert(vdp != NULL, "NULL VDP");
    prt_assert(channel_num >= 0 && channel_num < vdp->num_inputs,
        "channel number out of range");

    // Switch in the channel.
    prt_channel_on(vdp->input[channel_num]);
}


int prt_vdp_ready(prt_vdp_t *vdp)
{
    // Check input parameters.
    prt_assert(vdp != NULL, "NULL VDP");

    int i;
    // FOR each input channel.
    for (i = 0; i < vdp->num_inputs; i++)
        // IF the channel was established
        // AND the channel is active
        // AND the list of packets is empty.
        if (vdp->input[i] != NULL &&
            vdp->input[i]->active &&
            prt_channel_empty(vdp->input[i]))
            // Return "not ready".
            return 0;

    // Return "ready".
    return 1;
}