prt_vsa.c File Reference

PULSAR Runtime (PRT) More...

#include "prt_vsa.h"

Include dependency graph for prt_vsa.c:

Go to the source code of this file.

Functions
int	prt_tuple_equal (void *tuple_a, void *tuple_b)
	tuple equality check Check if tuples are identical in length and content. More...
unsigned int	prt_tuple_hash (void *tuple)
	tuple hash Required by the VSA's tuples hash table. Computes the lenght in characters and calls a string hash function. More...
prt_vsa_t *	prt_vsa_new (int num_threads, void *global_store, int(*vdp_to_core)(int *, void *, int))
	VSA constructor. More...
void	prt_vsa_delete (prt_vsa_t *vsa)
	VSA destructor. More...
void	prt_vsa_vdp_insert (prt_vsa_t *vsa, prt_vdp_t *vdp)
	Inserts a new VDP into a VSA. Puts the VDP in the list of VDPs of the owner thread. Connects corresponding input and output channels of connected VDPs. More...
void	prt_vsa_vdp_insert_local (prt_vsa_t *vsa, prt_vdp_t *vdp, int core, int node_rank)
void	prt_vsa_vdp_merge_channels (prt_vsa_t *vsa, prt_vdp_t *vdp)
void	prt_vsa_vdp_track_tags_local (prt_vsa_t *vsa, prt_vdp_t *vdp, int core, int node_rank)
void	prt_vsa_vdp_track_tags_remote (prt_vsa_t *vsa, prt_vdp_t *vdp, int node_rank)
void	prt_vsa_run (prt_vsa_t *vsa)
	VSA's production cycle Launches worker threads. Sends the master thread in the communication proxy production cycle. Joins the worker threads. More...
void	prt_vsa_config_set (prt_vsa_t *vsa, prt_config_param_t param, prt_config_value_t value)
	Set VSA configuration parameter. More...
void	prt_vsa_dump (prt_vsa_t *vsa)
	Dump the VSA structure. More...

Detailed Description

PULSAR Runtime (PRT)

Virtual Systolic Array (VSA) implementation.

Author

Jakub Kurzak

PULSAR Runtime /pulsar/ Copyright (C) 2012-2013 University of Tennessee.

Definition in file prt_vsa.c.

Function Documentation

int prt_tuple_equal	(	void *	tuple_a,
		void *	tuple_b
	)

tuple equality check Check if tuples are identical in length and content.

Parameters

tuple_a
tuple_b

Return values

0	Tuples differ.
1	Tuples are identical.

Definition at line 161 of file prt_tuple.c.

References prt_tuple_compare().

{
    return (prt_tuple_compare(tuple_a, tuple_b) == 0);
}

Here is the call graph for this function:

Here is the caller graph for this function:

unsigned int prt_tuple_hash

(

void *

tuple

)

tuple hash Required by the VSA's tuples hash table. Computes the lenght in characters and calls a string hash function.

Parameters

tuple

Returns

hash

Definition at line 188 of file prt_tuple.c.

{
    int len = 0;
    int *tup = (int*)tuple;
    while (tup[len] != INT_MAX)
        len++;
    return fnv_hash(tuple, len*sizeof(int));
}

Here is the caller graph for this function:

void prt_vsa_config_set	(	prt_vsa_t *	vsa,
		prt_config_param_t	param,
		prt_config_value_t	value
	)

Set VSA configuration parameter.

Parameters

vsa
param
value

Definition at line 416 of file prt_vsa.c.

References prt_vsa_s::config, prt_error, PRT_SVG_TRACING, PRT_SVG_TRACING_OFF, PRT_SVG_TRACING_ON, PRT_VDP_SCHEDULING, PRT_VDP_SCHEDULING_AGGRESSIVE, PRT_VDP_SCHEDULING_LAZY, prt_config_s::svg_tracing, and prt_config_s::vdp_scheduling.

{
    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_delete

(

prt_vsa_t *

vsa

)

VSA destructor.

Parameters

VSA

Definition at line 83 of file prt_vsa.c.

References prt_vsa_s::channel_lists, prt_vsa_s::channel_tags, prt_vsa_s::config, icl_hash_destroy(), icl_list_destroy(), prt_vsa_s::num_nodes, prt_vsa_s::num_threads, prt_vsa_s::proxy, prt_assert, prt_channel_delete(), prt_config_delete(), prt_proxy_delete(), prt_thread_delete(), prt_vsa_s::thread, prt_vsa_s::thread_attr, and prt_vsa_s::vdps_hash.

{
    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);
}

Here is the call graph for this function:

void prt_vsa_dump

(

prt_vsa_t *

vsa

)

Dump the VSA structure.

Parameters

vsa

Definition at line 454 of file prt_vsa.c.

References icl_list_s::data, prt_channel_s::dst_slot, prt_channel_s::dst_tuple, icl_hash_foreach, icl_list_first(), icl_list_next(), prt_vdp_s::input, prt_vsa_s::node_rank, prt_vdp_s::num_inputs, prt_vsa_s::num_nodes, prt_vdp_s::num_outputs, prt_vsa_s::num_threads, prt_vdp_s::output, prt_vsa_s::proxy, prt_tuple_print(), prt_channel_s::src_slot, prt_channel_s::src_tuple, prt_channel_s::tag, prt_proxy_s::tags_hash, prt_vsa_s::thread, prt_vdp_s::tuple, prt_channel_s::vdp, and prt_thread_s::vdps.

{
    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);
    }
}

Here is the call graph for this function:

prt_vsa_t* prt_vsa_new	(	int	num_threads,
		void *	global_store,
		int(*)(int *, void *, int)	vdp_to_core
	)

VSA constructor.

Parameters

num_threads	number of local worker threads
global_store	VSA's global store accessible to all VDPs
vdp_to_core	function mapping VDP's tuple to a global core number
max_vdps	maximum number of VDP's within a node

Returns

new VSA

Definition at line 27 of file prt_vsa.c.

References prt_vsa_s::channel_lists, prt_vsa_s::channel_tags, prt_vsa_s::config, prt_vsa_s::global_store, icl_hash_create(), prt_vsa_s::node_rank, prt_vsa_s::num_cores, prt_vsa_s::num_nodes, prt_vsa_s::num_threads, prt_vsa_s::proxy, prt_assert, prt_config_new(), prt_proxy_new(), prt_thread_new(), prt_tuple_equal(), prt_tuple_hash(), PRT_VSA_MAX_VDPS_PER_NODE, prt_vsa_s::thread, prt_vsa_s::thread_attr, prt_vsa_s::vdp_to_core, prt_vsa_s::vdps_hash, prt_thread_s::vsa, and prt_proxy_s::vsa.

{
    // 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;
}

Here is the call graph for this function:

void prt_vsa_run

(

prt_vsa_t *

vsa

)

VSA's production cycle Launches worker threads. Sends the master thread in the communication proxy production cycle. Joins the worker threads.

Parameters

vsa

Definition at line 376 of file prt_vsa.c.

References prt_vsa_s::config, prt_thread_s::id, Indigo, prt_vsa_s::num_threads, prt_vsa_s::proxy, prt_assert, prt_proxy_run(), PRT_SVG_TRACING_ON, prt_thread_run(), svg_trace_finish(), svg_trace_init(), svg_trace_start(), svg_trace_stop(), prt_config_s::svg_tracing, prt_vsa_s::thread, and prt_vsa_s::thread_attr.

{
    // 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();
}

Here is the call graph for this function:

void prt_vsa_vdp_insert	(	prt_vsa_t *	vsa,
		prt_vdp_t *	vdp
	)

Inserts a new VDP into a VSA. Puts the VDP in the list of VDPs of the owner thread. Connects corresponding input and output channels of connected VDPs.

Parameters

vsa
vdp

Definition at line 127 of file prt_vsa.c.

References prt_vsa_s::global_store, prt_vsa_s::node_rank, prt_vsa_s::num_cores, prt_vsa_s::num_threads, prt_assert, prt_vsa_vdp_insert_local(), prt_vsa_vdp_track_tags_remote(), prt_vdp_s::tuple, and prt_vsa_s::vdp_to_core.

{
    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);
}

Here is the call graph for this function:

void prt_vsa_vdp_insert_local	(	prt_vsa_t *	vsa,
		prt_vdp_t *	vdp,
		int	core,
		int	node_rank
	)

Definition at line 147 of file prt_vsa.c.

References icl_hash_insert(), icl_list_append(), prt_vsa_s::num_threads, prt_assert, prt_vsa_vdp_merge_channels(), prt_vsa_vdp_track_tags_local(), prt_vdp_s::thread, prt_vsa_s::thread, prt_vdp_s::tuple, prt_thread_s::vdps, and prt_vsa_s::vdps_hash.

{
    // 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);
}

Here is the call graph for this function:

Here is the caller graph for this function:

void prt_vsa_vdp_merge_channels	(	prt_vsa_t *	vsa,
		prt_vdp_t *	vdp
	)

Definition at line 168 of file prt_vsa.c.

References prt_channel_s::dst_slot, prt_channel_s::dst_tuple, icl_hash_find(), prt_vdp_s::input, prt_vdp_s::num_inputs, prt_vdp_s::num_outputs, prt_vdp_s::output, prt_vsa_s::proxy, prt_assert, prt_channel_delete(), prt_proxy_max_packet_size(), prt_tuple_equal(), prt_channel_s::src_slot, prt_channel_s::src_tuple, prt_channel_s::tag, prt_vdp_s::tuple, and prt_vsa_s::vdps_hash.

{
    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;
            }
        }
    }
}

Here is the call graph for this function:

Here is the caller graph for this function:

void prt_vsa_vdp_track_tags_local	(	prt_vsa_t *	vsa,
		prt_vdp_t *	vdp,
		int	core,
		int	node_rank
	)

Definition at line 215 of file prt_vsa.c.

References prt_vsa_s::channel_lists, prt_vsa_s::channel_tags, icl_list_s::data, prt_channel_s::dst_node, prt_channel_s::dst_tuple, prt_vsa_s::global_store, icl_hash_insert(), icl_list_append(), icl_list_new(), icl_list_search(), prt_vdp_s::input, prt_vsa_s::node_rank, prt_vsa_s::num_cores, prt_vdp_s::num_inputs, prt_vdp_s::num_outputs, prt_vsa_s::num_threads, prt_vdp_s::output, prt_vsa_s::proxy, prt_assert, prt_channel_compare(), prt_tuple_new2, prt_channel_s::src_tuple, prt_channel_s::tag, prt_proxy_s::tags_hash, and prt_vsa_s::vdp_to_core.

{
    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");
            }
        }
    }



}

Here is the call graph for this function:

Here is the caller graph for this function:

void prt_vsa_vdp_track_tags_remote	(	prt_vsa_t *	vsa,
		prt_vdp_t *	vdp,
		int	node_rank
	)

Definition at line 298 of file prt_vsa.c.

References prt_vsa_s::channel_lists, prt_vsa_s::channel_tags, prt_channel_s::dst_node, prt_channel_s::dst_tuple, prt_vsa_s::global_store, icl_list_append(), icl_list_new(), icl_list_search(), prt_vdp_s::input, prt_vsa_s::node_rank, prt_vsa_s::num_cores, prt_vdp_s::num_inputs, prt_vdp_s::num_outputs, prt_vsa_s::num_threads, prt_vdp_s::output, prt_channel_compare(), prt_channel_delete(), prt_channel_s::src_tuple, prt_channel_s::tag, and prt_vsa_s::vdp_to_core.

{
    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);
        }
    }
}

Here is the call graph for this function:

Here is the caller graph for this function: