prt_channel.c

Go to the documentation of this file.

#include "prt_channel.h"


prt_channel_t* prt_channel_new(
    int count, MPI_Datatype datatype,
    int *src_tuple, int src_slot,
    int *dst_tuple, int dst_slot)
{
    // Allocate the channel.
    prt_channel_t *channel = (prt_channel_t*)malloc(sizeof(prt_channel_t));
    prt_assert(channel != NULL, "malloc failed");

    // Check parameters.
    prt_assert(count > 0, "count less or equal zero");
    prt_assert(src_tuple != NULL, "NULL source tuple");
    prt_assert(src_tuple != NULL, "NULL destination tuple");

    // Initialize the channel.
    channel->vdp = NULL;
    channel->count = count;
    channel->datatype = datatype;
    channel->src_tuple = src_tuple;
    channel->dst_tuple = dst_tuple;
    channel->src_slot = src_slot;
    channel->dst_slot = dst_slot;

    // Create the list of packets.
    channel->packets = icl_deque_new();
    prt_assert(channel->packets != NULL, "icl_deque_new() failed");
//printf("new\n"); fflush(stdout);
    // 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);
//printf("delete\n"); fflush(stdout);
    // Free the channel.
    free(channel);
}


void prt_channel_push(prt_channel_t *channel, prt_packet_t *packet)
{
    // Check for a NULL input params.
    prt_assert(packet != NULL, "NULL packet");
    prt_assert(channel != NULL, "NULL channel");
    prt_assert(channel->packets != NULL, "NULL packet list");

    __sync_fetch_and_add(&packet->num_refs, 1);
    icl_deque_append(channel->packets, (void*)packet);

    // IF a dangling channel.
    if (channel->dst_node != channel->vdp->thread->vsa->node_rank)
        // Notify the proxy.
        prt_proxy_send(
            channel->vdp->thread->vsa->proxy,
            channel->vdp->thread->rank,
            channel);
}


prt_packet_t *prt_channel_pop(prt_channel_t *channel)
{
    // Check for a NULL input params.
    prt_assert(channel != NULL, "popping from a NULL channel");
    prt_assert(channel->packets != NULL, "popping from a NULL packet list");

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

    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 packet list");

    // 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 1;
    if (prt_tuple_compare(c1->dst_tuple, c2->dst_tuple) != 0) return 1;
    if (c1->src_slot != c2->src_slot) return 1;
    if (c1->dst_slot != c2->dst_slot) return 1;
    return 0;
}