prt_vsa.c

Go to the documentation of this file.

#include "prt_vsa.h"

extern int prt_tuple_equal(void *tuple_a, void *tuple_b);
extern unsigned int prt_tuple_hash(void *tuple);


prt_vsa_t* prt_vsa_new(
    int num_threads, void *global_store, int (*vdp_to_core)(int*, void*, int))
{
    // Allocate the VSA.
    prt_vsa_t *vsa = (prt_vsa_t*)malloc(sizeof(prt_vsa_t));
    prt_assert(vsa != NULL, "malloc failed");

    // Init the VSA.
    MPI_Comm_rank(MPI_COMM_WORLD, &vsa->node_rank);
    MPI_Comm_size(MPI_COMM_WORLD, &vsa->num_nodes);
    vsa->num_threads = num_threads;
    vsa->num_cores = vsa->num_nodes*vsa->num_threads;
    vsa->global_store = global_store;
    vsa->vdp_to_core = vdp_to_core;

    // Init config & proxy.
    vsa->config = prt_config_new();
    vsa->proxy = prt_proxy_new(num_threads);
    vsa->proxy->vsa = vsa;

    // Init pthreads.
    pthread_attr_init(&vsa->thread_attr);
    pthread_attr_setscope(&vsa->thread_attr, PTHREAD_SCOPE_SYSTEM);
    pthread_setconcurrency(vsa->num_threads+1);

    int i;
    // Initialize the threads.
    vsa->thread = (prt_thread_t**)malloc(vsa->num_threads*sizeof(prt_thread_t*));
    prt_assert(vsa->thread != NULL, "malloc failed");
    for (i = 0; i < vsa->num_threads; i++) {
        vsa->thread[i] = prt_thread_new(i, vsa->node_rank*vsa->num_threads+i);
        vsa->thread[i]->vsa = vsa;
    }

    // Initialize the VDPs hash.
    int nbuckets = PRT_VSA_MAX_VDPS_PER_NODE;
    vsa->vdps_hash = icl_hash_create(nbuckets, prt_tuple_hash, prt_tuple_equal);

    // Allocate the array of channel lists.
    vsa->channel_lists = (icl_list_t**)calloc(vsa->num_nodes, sizeof(icl_list_t*));
    prt_assert(vsa->channel_lists != NULL, "malloc failed");

    // Allocate counters for channel tags.
    vsa->channel_tags = (int*)calloc(vsa->num_nodes, sizeof(int));
    prt_assert(vsa->channel_tags != NULL, "malloc failed");

    // Return the VSA.
    return vsa;
}


void prt_vsa_delete(prt_vsa_t *vsa)
{
    int i;
    // Pthreads cleanup.
    pthread_attr_destroy(&vsa->thread_attr);
    for (i = 0; i < vsa->num_threads; i++)
        prt_thread_delete(vsa->thread[i]);
    free(vsa->thread);

    // Delete config & proxy.
    prt_config_delete(vsa->config);
    prt_proxy_delete(vsa->proxy);

    // Destroy the VDPs hash.
//  icl_hash_destroy(vsa->vdps_hash, free, (void(*)(void*))prt_vdp_delete);
    icl_hash_destroy(vsa->vdps_hash, free, (void(*)(void*))NULL);

    // Destroy the channel lists.
    for (i = 0; i < vsa->num_nodes; i++)
        if (vsa->channel_lists[i] != NULL) {
            int status = icl_list_destroy(
                vsa->channel_lists[i], (void(*)(void*))prt_channel_delete);
            prt_assert(status == 0, "icl_list_destroy failed");
        }

    // Free the array of lists.
    free(vsa->channel_lists);
    // Free the tag counters.
    free(vsa->channel_tags);

    // Free the VSA.
    free(vsa);
}


void prt_vsa_vdp_insert(prt_vsa_t *vsa, prt_vdp_t *vdp)
{
    int i;
    // Check arguments.
    prt_assert(vsa != NULL, "NULL VSA");
    prt_assert(vdp != NULL, "NULL VDP");

    // Compute global core number and process rank.
    int core = vsa->vdp_to_core(vdp->tuple, vsa->global_store, vsa->num_cores);
    int node_rank = core / vsa->num_threads;

    // IF VDP in this node.
    if (node_rank == vsa->node_rank)
        prt_vsa_vdp_insert_local(vsa, vdp, core, node_rank);
    else
        prt_vsa_vdp_track_tags_remote(vsa, vdp, node_rank);
}



void prt_vsa_vdp_insert_local(
    prt_vsa_t *vsa, prt_vdp_t *vdp, int core, int node_rank)
{
    // Insert in the VSA's VDP hash.
    icl_entry_t *entry = icl_hash_insert(
        vsa->vdps_hash, (void*)vdp->tuple, (void*)vdp);
    prt_assert(entry != NULL, "icl_hash_insert failed");

    // Insert in the thread's VDPs list.
    int thread_rank = core % vsa->num_threads;
    icl_list_t *node = icl_list_append(vsa->thread[thread_rank]->vdps, vdp);
    prt_assert(node != NULL, "icl_list_append failed");
    vdp->thread = vsa->thread[thread_rank];

    // Merge intra-node channels.
    prt_vsa_vdp_merge_channels(vsa, vdp);
    // Track channel tags for inter-node communication.
    prt_vsa_vdp_track_tags_local(vsa, vdp, core, node_rank);
}


void prt_vsa_vdp_merge_channels(prt_vsa_t *vsa, prt_vdp_t *vdp)
{
    int i;
    // FOR each input channel.
    for (i = 0; i < vdp->num_inputs; i++) {
        prt_channel_t *channel = vdp->input[i];
        if (channel != NULL) {
            // Look for maximum channel size.
            prt_proxy_max_packet_size(vsa->proxy, channel);
            // Look up the source VDP.
            prt_vdp_t *src_vdp = icl_hash_find(vsa->vdps_hash, (void*)channel->src_tuple);
            // IF source VDP found.
            if (src_vdp != NULL) {
                // Check if the channel on the source VDP points to this VDP.
                int *src_vdp_dst_tuple = src_vdp->output[channel->src_slot]->dst_tuple;
                int retval = prt_tuple_equal(src_vdp_dst_tuple, vdp->tuple);
                prt_assert(retval == 1, "VDP channel tuple mismatch");
                // Swap this channel to the existing channel.
                vdp->input[i] = src_vdp->output[channel->src_slot];
                prt_channel_delete(channel);
                vdp->input[i]->tag = -1;
            }
        }
    }
    // FOR each output channel.
    for (i = 0; i < vdp->num_outputs; i++) {
        prt_channel_t *channel = vdp->output[i];
        if (channel != NULL) {
            // Look for maximum channel size.
            prt_proxy_max_packet_size(vsa->proxy, channel);
            // Look up the destination VDP.
            prt_vdp_t *dst_vdp = icl_hash_find(vsa->vdps_hash, (void*)channel->dst_tuple);
            // IF destination VDP found.
            if (dst_vdp != NULL) {
                // Check if the channel on the destination VDP points to this VDP.
                int *dst_vdp_src_tuple = dst_vdp->input[channel->dst_slot]->src_tuple;
                int retval = prt_tuple_equal(dst_vdp_src_tuple, vdp->tuple);
                prt_assert(retval == 1, "VDP channel tuple mismatch");
                // Swap this channel for the existing channel.
                vdp->output[i] = dst_vdp->input[channel->dst_slot];
                prt_channel_delete(channel);
                vdp->output[i]->tag = -1;
            }
        }
    }
}

void prt_vsa_vdp_track_tags_local(
    prt_vsa_t *vsa, prt_vdp_t *vdp, int core, int node_rank)
{
    int i;
    // FOR each input channel.
    for (i = 0; i < vdp->num_inputs; i++) {
        prt_channel_t *channel = vdp->input[i];
        if (channel != NULL) {
            channel->dst_node = node_rank;
            int src_core = vsa->vdp_to_core(
                channel->src_tuple, vsa->global_store, vsa->num_cores);
            int src_node = src_core / vsa->num_threads;
            // IF another node is the source.
            if (src_node != vsa->node_rank) {
                // Create the list if empty.
                if (vsa->channel_lists[src_node] == NULL)
                    vsa->channel_lists[src_node] = icl_list_new();
                // Search the list for the channel.
                icl_list_t *node = icl_list_search(
                    vsa->channel_lists[src_node], channel, prt_channel_compare);
                // IF channel already on the list.
                if (node != NULL)
                    // Assign the existing tag.
                    channel->tag = ((prt_channel_t*)(node->data))->tag;
                // ELSE
                else {
                    // Add the channel to the list.
                    icl_list_append(vsa->channel_lists[src_node], channel);
                    // Assign the next value and increment.
                    channel->tag = vsa->channel_tags[src_node];
                    vsa->channel_tags[src_node]++;
                }
                // Add the channel to the proxy's tag hash.
                int *src_node_tag = prt_tuple_new2(src_node, channel->tag);
                icl_entry_t *entry = icl_hash_insert(
                    vsa->proxy->tags_hash, (void*)src_node_tag, (void*)channel);
                prt_assert(entry != NULL, "icl_hash_insert failed");
            }
        }
    }
    // FOR each output channel.
    for (i = 0; i < vdp->num_outputs; i++) {
        prt_channel_t *channel = vdp->output[i];
        if (channel != NULL) {
            int dst_core = vsa->vdp_to_core(
                channel->dst_tuple, vsa->global_store, vsa->num_cores);
            int dst_node = dst_core / vsa->num_threads;
            channel->dst_node = dst_node;
            // IF another node is the destination.
            if (dst_node != vsa->node_rank) {
                // Create the list if empty.
                if (vsa->channel_lists[dst_node] == NULL)
                    vsa->channel_lists[dst_node] = icl_list_new();
                // Search the list for the channel.
                icl_list_t *node = icl_list_search(
                    vsa->channel_lists[dst_node], channel, prt_channel_compare);
                // IF channel already on the list.
                if (node != NULL)
                    // Assign the existing tag.
                    channel->tag = ((prt_channel_t*)(node->data))->tag;
                // ELSE
                else {
                    // Add the channel to the list.
                    icl_list_append(vsa->channel_lists[dst_node], channel);
                    // Assign the next value and increment.
                    channel->tag = vsa->channel_tags[dst_node];
                    vsa->channel_tags[dst_node]++;
                }
                // Add the channel to the proxy's tag hash.
                int *dst_node_tag = prt_tuple_new2(dst_node, channel->tag);
                icl_entry_t *entry = icl_hash_insert(
                    vsa->proxy->tags_hash, (void*)dst_node_tag, (void*)channel);
                prt_assert(entry != NULL, "icl_hash_insert failed");
            }
        }
    }



}



void prt_vsa_vdp_track_tags_remote(prt_vsa_t *vsa, prt_vdp_t *vdp, int node_rank)
{
    int i;
    // FOR each input channel.
    for (i = 0; i < vdp->num_inputs; i++) {
        prt_channel_t *channel = vdp->input[i];
        if (channel != NULL) {
            int src_core = vsa->vdp_to_core(channel->src_tuple, vsa->global_store, vsa->num_cores);
            int src_node = src_core / vsa->num_threads;
            channel->dst_node = node_rank;
            // IF this node is the source.
            if (src_node == vsa->node_rank) {
                // Create the list if empty.
                if (vsa->channel_lists[node_rank] == NULL)
                    vsa->channel_lists[node_rank] = icl_list_new();
                // Search the list for the channel.
                icl_list_t *node = icl_list_search(
                    vsa->channel_lists[node_rank], channel, prt_channel_compare);
                // IF channel already on the list.
                if (node != NULL) {
                    // Free this channel.
                    prt_channel_delete(channel);
                }
                else {
                    // Add the channel to the list.
                    icl_list_append(vsa->channel_lists[node_rank], channel);
                    // Assign the next value and increment.
                    channel->tag = vsa->channel_tags[node_rank];
                    vsa->channel_tags[node_rank]++;
                }
            }
            else
                prt_channel_delete(channel);
        }
    }
    // FOR each output channel.
    for (i = 0; i < vdp->num_outputs; i++) {
        prt_channel_t *channel = vdp->output[i];
        if (channel != NULL) {
            int dst_core = vsa->vdp_to_core(channel->dst_tuple, vsa->global_store, vsa->num_cores);
            int dst_node = dst_core / vsa->num_threads;
            channel->dst_node = dst_node;
            // IF this node is the destination.
            if (dst_node == vsa->node_rank) {
                // Create the list if empty.
                if (vsa->channel_lists[node_rank] == NULL)
                    vsa->channel_lists[node_rank] = icl_list_new();
                // Search the list for the channel.
                icl_list_t *node = icl_list_search(
                    vsa->channel_lists[node_rank], channel, prt_channel_compare);
                // IF channel already on the list.
                if (node != NULL) {
                    // Free this channel.
                    prt_channel_delete(channel);
                }
                else {
                    // Add the channel to the list.
                    icl_list_append(vsa->channel_lists[node_rank], channel);
                    // Assign the next value and increment.
                    channel->tag = vsa->channel_tags[node_rank];
                    vsa->channel_tags[node_rank]++;
                }
            }
            else
                prt_channel_delete(channel);
        }
    }
}


void prt_vsa_run(prt_vsa_t *vsa)
{
    // Init tracing.
    svg_trace_init(vsa->num_threads);

    int i;
    // Launch threads.
    for (i = 0; i < vsa->num_threads; i++) {
        int status =
            pthread_create(
                &vsa->thread[i]->id, &vsa->thread_attr, prt_thread_run, vsa->thread[i]);
        prt_assert(status == 0, "pthread_create failed");
    }

    // Barrier for trace alignment.
    svg_trace_start(0);
    MPI_Barrier(MPI_COMM_WORLD);
    svg_trace_stop(0, Indigo);

    // Service the communication proxy.
    prt_proxy_run(vsa->proxy);

    // Join threads.
    for (i = 0; i < vsa->num_threads; i++) {
        int status = pthread_join(vsa->thread[i]->id, NULL);
        prt_assert(status == 0, "pthread_join failed");
    }
    // Finish tracing.
    if (vsa->config->svg_tracing == PRT_SVG_TRACING_ON)
        svg_trace_finish();
}


void prt_vsa_config_set(
    prt_vsa_t *vsa, prt_config_param_t param, prt_config_value_t value)
{
    switch (param) {
        case PRT_VDP_SCHEDULING:
            switch (value) {
                case PRT_VDP_SCHEDULING_LAZY:
                case PRT_VDP_SCHEDULING_AGGRESSIVE:
                    vsa->config->vdp_scheduling = value;
                    break;
                default:
                    prt_error("invalid value");
                    break;
            }
            break;
        case PRT_SVG_TRACING:
            switch (value) {
                case PRT_SVG_TRACING_ON:
                case PRT_SVG_TRACING_OFF:
                    vsa->config->svg_tracing = value;
                    break;
                default:
                    prt_error("invalid value");
                    break;
            }
            break;
        default:
            prt_error("invalid param");
            break;
    }
}


void prt_vsa_dump(prt_vsa_t *vsa)
{
    int rank;
    for (rank = 0; rank < vsa->num_nodes; rank++)
    {
        if (rank == vsa->node_rank)
        {
            printf("    NODE:%3d\n", vsa->node_rank);
            int thread_rank;
            for (thread_rank = 0; thread_rank < vsa->num_threads; thread_rank++)
            {
                printf("        THREAD:%3d\n", thread_rank);
                prt_thread_t *thread = vsa->thread[thread_rank];
                icl_list_t *ptr;
                for (ptr = icl_list_first(thread->vdps);
                    ptr != NULL;
                    ptr = icl_list_next(thread->vdps, ptr))
                {
                    prt_vdp_t *vdp = (prt_vdp_t*)ptr->data;
                    printf("            VDP:"); prt_tuple_print(vdp->tuple); printf("\n");
                    int i;
                    for (i = 0; i < vdp->num_inputs; i++)
                    {
                        prt_channel_t *channel = vdp->input[i];
                        if (channel != NULL) {
                            printf("                input from VDP"); prt_tuple_print(channel->src_tuple);
                            if (channel->tag == -1)
                                printf(" slot%3d, channel tag:   \n", channel->src_slot);
                            else
                                printf(" slot%3d, channel tag:%3d\n", channel->src_slot, channel->tag);
                        }
                        else {
                            printf("                input NULL\n");
                        }
                    }
                    for (i = 0; i < vdp->num_outputs; i++)
                    {
                        if (i == 0)
                            printf("\n");
                        prt_channel_t *channel = vdp->output[i];
                        if (channel != NULL) {
                            printf("                output to VDP"); prt_tuple_print(channel->dst_tuple);
                            if (channel->tag == -1)
                                printf(" slot%3d, channel tag:   \n", channel->dst_slot);
                            else
                                printf(" slot%3d, channel tag:%3d\n", channel->dst_slot, channel->tag);
                        }
                        else {
                            printf("                output NULL\n");
                        }
                    }
                }
            }
        }
        MPI_Barrier(MPI_COMM_WORLD);
        fflush(stdout);
        sleep(1);
    }

    for (rank = 0; rank < vsa->num_nodes; rank++)
    {
        if (rank == vsa->node_rank)
        {
            printf("    NODE:%3d\n", vsa->node_rank);
            int tmpint;
            icl_entry_t *tmpent;
            int *pk;
            prt_channel_t *pd;
            icl_hash_foreach(vsa->proxy->tags_hash, tmpint, tmpent, pk, pd)
            {
                prt_channel_t *channel = pd;
                printf("      %d %d: ", pk[0], pk[1]);
                prt_tuple_print(channel->src_tuple);
                printf(" -> ");
                prt_tuple_print(channel->dst_tuple);
                printf("\n");
            }
        }
        MPI_Barrier(MPI_COMM_WORLD);
        fflush(stdout);
        sleep(1);
    }
}