prt_channel.c

Go to the documentation of this file.

#include "prt_channel.h"


prt_channel_t* prt_channel_new(
    size_t size,
    int *src_tuple, int src_slot,
    int *dst_tuple, int dst_slot)
{
    // Check input parameters.
    prt_assert(size > 0, "channel size equals zero");
    prt_assert(size <= INT_MAX, "channel size larger than INT_MAX");
    prt_assert(src_tuple != NULL, "NULL source tuple");
    prt_assert(src_slot >= 0, "negative source slot");
    prt_assert(dst_tuple != NULL, "NULL destination tuple");
    prt_assert(dst_slot >= 0, "negative destination slot");

    // Allocate the channel.
    prt_channel_t *channel = (prt_channel_t*)malloc(sizeof(prt_channel_t));
    prt_assert(channel != NULL, "malloc failed");

    // Initialize the channel.
    channel->dst_vdp = NULL;
    channel->src_vdp = NULL;
    channel->proxy = NULL;
    channel->size = size;
    channel->src_tuple = src_tuple;
    channel->dst_tuple = dst_tuple;
    channel->src_slot = src_slot;
    channel->dst_slot = dst_slot;
    channel->active = 1;

    // Create the list of packets.
    channel->packets = icl_deque_new();
    prt_assert(channel->packets != NULL, "icl_deque_new() failed");

    // Return the channel.
    return channel;
}


void prt_channel_delete(prt_channel_t *channel)
{
    // Check for a NULL channel.
    prt_assert(channel != NULL, "NULL channel");

    // Free the source tuple.
    prt_assert(channel->src_tuple != NULL, "NULL tuple");
    free(channel->src_tuple);

    // Free the destination tuple.
    prt_assert(channel->dst_tuple != NULL, "NULL tuple");
    free(channel->dst_tuple);

    // Destroy the list of packets.
    // Only an empty list can be destroyed.
    // Therefore NULL given as the packet destructor.
    prt_assert(channel->packets != NULL, "NULL packets list");
    int size = icl_deque_size(channel->packets);

    prt_assert(size == 0, "non-epty packet list");
    icl_deque_destroy(channel->packets, NULL);

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


void prt_channel_push_host(
    prt_vdp_t *vdp, prt_channel_t *channel, prt_packet_t *packet)
{
    // Increment references.
    // Now the packet is in the VDP and in the channel.
    // The VDP's reference is active until the VDP releases the packet.
    __sync_fetch_and_add(&packet->num_refs, 1);

    // IF going to a remote destination.
    if (channel->src_node != channel->dst_node) {
        // Create the request.
        prt_request_t *request =
            prt_request_new(
                packet, packet->size, channel->dst_node, channel->tag);
        // Queue for send.
        icl_deque_append(
            channel->proxy->sends_requested[vdp->thread->agent_rank], request);
    }
    // ELSE IF a local destination.
    else { // IF a host destination.
        if (channel->dst_vdp->location == PRT_LOCATION_HOST) {
            // host -> host
            icl_deque_append(channel->packets, (void*)packet);
        }
        // ELSE if a device destination.
        else { // host -> device
            prt_transfer_t *transfer =
                prt_transfer_new(packet, channel, PRT_HOST_TO_DEVICE, -1);
            icl_deque_append(channel->proxy->transfers, (void*)transfer);
        }
    }
}


void prt_channel_push_device(
    prt_vdp_t *vdp, prt_channel_t *channel, prt_packet_t *packet)
{
    // Increment references.
    // Now the packet is in the channel and in the VDP.
    // The VDP's reference is active until the VDP releases the packet.
    __sync_fetch_and_add(&packet->num_refs, 1);

    // IF remote destination.
    if (channel->src_node != channel->dst_node) {
        // Device -> MPI.
        // Add a callback that queues the transfer request.
        prt_callback_queue_t *callback =
            prt_callback_queue_new(
                NULL, packet, channel,
                PRT_DEVICE_MPI_TO_HOST,
                vdp->device->agent_rank);
        __sync_fetch_and_add(&channel->proxy->num_callbacks, 1);
        cudaStreamAddCallback(
            vdp->stream, prt_callback_queue_handler, (void*)callback, 0);
    }
    // ELSE IF local destination.
    else { // IF host destination.
        if (channel->dst_vdp->location == PRT_LOCATION_HOST) {
            // Device -> host.
            // Add a callback that queues the transfer request.
            prt_callback_queue_t *callback =
                prt_callback_queue_new(
                    NULL, packet, channel, PRT_DEVICE_TO_HOST, -1);
            __sync_fetch_and_add(&channel->proxy->num_callbacks, 1);
            cudaStreamAddCallback(
                vdp->stream, prt_callback_queue_handler, (void*)callback, 0);
        }
        // ELSE IF device destination.
        else { // IF the same device.
            if (packet->device_rank == channel->dst_vdp->device->rank) {
                // Device -> the same device.
                // Add a callback that puts the packet in the channel.
                prt_callback_finish_t *callback =
                    prt_callback_finish_new(NULL, packet, channel);
                __sync_fetch_and_add(&channel->proxy->num_callbacks, 1);
                cudaStreamAddCallback(
                    vdp->stream, prt_callback_finish_handler, (void*)callback, 0);
            }
            // ELSE IF a different device.
            else { // dev -> different dev
                // Add a callback that queues the transfer request.
                prt_callback_queue_t *callback =
                    prt_callback_queue_new(
                        NULL, packet, channel, PRT_DEVICE_TO_DEVICE, -1);
                __sync_fetch_and_add(&channel->proxy->num_callbacks, 1);
                cudaStreamAddCallback(
                    vdp->stream, prt_callback_queue_handler, (void*)callback, 0);
            }
        }
    }
}


prt_packet_t *prt_channel_pop(prt_channel_t *channel)
{
    // Defensive error check.
    prt_assert(channel->packets != NULL, "NULL list of packets");

    icl_node_t *node = icl_deque_first(channel->packets);
    prt_assert(node != NULL, "empty list of packets");

    prt_packet_t *packet = (prt_packet_t*)node->data;
    prt_assert(packet != NULL, "NULL packet");

    icl_deque_delete(channel->packets, node, NULL);
    return packet;
}


int prt_channel_empty(prt_channel_t *channel)
{
    // Check for a NULL input params.
    prt_assert(channel != NULL, "NULL channel");
    prt_assert(channel->packets != NULL, "NULL list of packets");

    // Return the status.
    return (icl_deque_first(channel->packets) == NULL);
}


int prt_channel_compare(void *channel1, void *channel2)
{
    prt_channel_t *c1 = (prt_channel_t*)channel1;
    prt_channel_t *c2 = (prt_channel_t*)channel2;

    if (prt_tuple_compare(c1->src_tuple, c2->src_tuple) != 0)
        return prt_tuple_compare(c1->src_tuple, c2->src_tuple);

    if (c1->src_slot < c2->src_slot) return -1;
    if (c1->src_slot > c2->src_slot) return 1;

    if (prt_tuple_compare(c1->dst_tuple, c2->dst_tuple) != 0)
        return prt_tuple_compare(c1->dst_tuple, c2->dst_tuple);

    if (c1->dst_slot < c2->dst_slot) return -1;
    if (c1->dst_slot > c2->dst_slot) return 1;

    return 0;
}


void prt_channel_off(prt_channel_t *channel)
{
    // Switch the channel off.
    channel->active = 0;
}


void prt_channel_on(prt_channel_t *channel)
{
    // Switch the channel on.
    channel->active = 1;
}