PRT communication proxy. More...

#include "prt.h"

Include dependency graph for prt_proxy.h:

This graph shows which files directly or indirectly include this file:

Data Structures
struct	prt_proxy_s
	VSA's communication proxy Serves communication requests from the worker threads. Contains a list of recv requests from each worker thread. Contains a list of send requests from each worker thread. More...

Macros
#define	PRT_PROXY_MAX_TAGS_PER_NODE 1009
	maximum tags per node Size of the proxy's hash table of tags. Should be a prime number. More...

#define	PRT_PROXY_MAX_SENDS_PER_THREAD 1

#define	PRT_PROXY_MAX_RECVS_PER_THREAD 1

Typedefs
typedef struct prt_proxy_s	prt_proxy_t
	VSA's communication proxy Serves communication requests from the worker threads. Contains a list of recv requests from each worker thread. Contains a list of send requests from each worker thread. More...

Functions
prt_proxy_t *	prt_proxy_new (int num_threads)
	communication proxy constructor More...

void	prt_proxy_delete (prt_proxy_t *proxy)
	communication proxy destructor Checking if all the lists are empty at the time of destruction. Not destroying the list of receives (destroyed at the end of prt_proxy_run). More...

void	prt_proxy_max_packet_size (prt_proxy_t proxy, struct prt_channel_s channel)
	Look for maximum channel/packet size. More...

void	prt_proxy_send (prt_proxy_t proxy, int thread_rank, struct prt_channel_s channel)
	send from a channel More...

void	prt_proxy_recv (prt_proxy_t proxy, struct prt_request_s request)
	recv to a channel More...

void	prt_proxy_run (prt_proxy_t *proxy)
	communication proxy production cycle Serves communication requests of local worker threads until shut down. More...

Detailed Description

PRT communication proxy.

Author: Jakub Kurzak

Definition in file prt_proxy.h.

Macro Definition Documentation

#define PRT_PROXY_MAX_RECVS_PER_THREAD 1

Definition at line 31 of file prt_proxy.h.

#define PRT_PROXY_MAX_SENDS_PER_THREAD 1

Definition at line 30 of file prt_proxy.h.

#define PRT_PROXY_MAX_TAGS_PER_NODE 1009

maximum tags per node Size of the proxy's hash table of tags. Should be a prime number.

Definition at line 29 of file prt_proxy.h.

Typedef Documentation

typedef struct prt_proxy_s prt_proxy_t

VSA's communication proxy Serves communication requests from the worker threads. Contains a list of recv requests from each worker thread. Contains a list of send requests from each worker thread.

Function Documentation

void prt_proxy_delete ( prt_proxy_t * proxy )

communication proxy destructor Checking if all the lists are empty at the time of destruction. Not destroying the list of receives (destroyed at the end of prt_proxy_run).

Parameters

proxy

Definition at line 66 of file prt_proxy.c.

References icl_deque_destroy(), icl_deque_size(), icl_hash_destroy(), icl_list_destroy(), icl_list_size(), prt_proxy_s::num_threads, prt_assert, prt_proxy_s::sends_posted, prt_proxy_s::sends_requested, and prt_proxy_s::tags_hash.

 {
     int status;
     // Destroy the hash.
     // Free the keys (tuples).
     // Leave the channel references alone.
     status = icl_hash_destroy(proxy->tags_hash, free, NULL);
     prt_assert(status == 0, "icl_hash_destroy failed");
 
     int i;
     int size;
     // Destroy send lists.
     // Check if empty at destruction.
     for (i = 0; i < proxy->num_threads; i++) {
         size = icl_deque_size(proxy->sends_requested[i]);
         prt_assert(size == 0, "destroying non-empty deque");
         status = icl_deque_destroy(proxy->sends_requested[i], NULL);
         prt_assert(status == 0, "icl_deque_destroy failed");
 
         size = icl_list_size(proxy->sends_posted[i]);
         prt_assert(size == 0, "destroying non-empty list");
         status = icl_list_destroy(proxy->sends_posted[i], NULL);
         prt_assert(status == 0, "icl_list_destroy failed");
     }
     // Free arrays.
     free(proxy->sends_requested);
     free(proxy->sends_posted);
 
     // Free the proxy.
     free(proxy);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void prt_proxy_max_packet_size	(	prt_proxy_t *	proxy,
		prt_channel_t *	channel
	)

Look for maximum channel/packet size.

Parameters

proxy
channel

Definition at line 105 of file prt_proxy.c.

References prt_channel_s::count, prt_channel_s::datatype, and prt_proxy_s::max_packet_size.

 {
     int type_size;
     MPI_Type_size(channel->datatype, &type_size);
     size_t packet_size = (size_t)channel->count*type_size;
     if (packet_size > proxy->max_packet_size)
         proxy->max_packet_size = packet_size;
 }

Here is the caller graph for this function:

prt_proxy_t* prt_proxy_new ( int num_threads )

communication proxy constructor

Parameters

num_threads number of local worker threads

Returns: new communication proxy

Definition at line 21 of file prt_proxy.c.

References icl_deque_new(), icl_hash_create(), icl_list_new(), prt_proxy_s::max_packet_size, prt_proxy_s::num_threads, prt_assert, PRT_PROXY_MAX_TAGS_PER_NODE, prt_tuple_equal(), prt_tuple_hash(), prt_proxy_s::recvs_posted, prt_proxy_s::sends_posted, prt_proxy_s::sends_requested, and prt_proxy_s::tags_hash.

 {
     // Allocate the proxy.
     prt_proxy_t *proxy = (prt_proxy_t*)malloc(sizeof(prt_proxy_t));
     prt_assert(proxy != NULL, "malloc failed");
 
     // Init the proxy.
     proxy->num_threads = num_threads;
     proxy->max_packet_size = 0;
 
     // Create the hash table of tags.
     int nbuckets = PRT_PROXY_MAX_TAGS_PER_NODE;
     proxy->tags_hash = icl_hash_create(nbuckets, prt_tuple_hash, prt_tuple_equal);
     prt_assert(proxy->tags_hash != NULL, "icl_hash_create failed");
 
     // Allocate arrays of list.
     proxy->sends_requested = (icl_deque_t**)malloc(num_threads*sizeof(icl_deque_t*));
     proxy->sends_posted = (icl_list_t**)malloc(num_threads*sizeof(icl_list_t*));
     prt_assert(proxy->sends_requested != NULL, "malloc failed");
     prt_assert(proxy->sends_posted != NULL, "malloc failed");
 
     int i;
     // Create send lists.
     for (i = 0; i < num_threads; i++) {
         proxy->sends_requested[i] = icl_deque_new();
         proxy->sends_posted[i] = icl_list_new();
         prt_assert(proxy->sends_requested[i] != NULL, "icl_deque_new failed");
         prt_assert(proxy->sends_posted[i] != NULL, "icl_list_new failed");
     }
     // Create list of receives.
     proxy->recvs_posted = icl_list_new();
     prt_assert(proxy->recvs_posted != NULL, "icl_list_new failed");
 
     // Return the proxy.
     return proxy;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void prt_proxy_recv	(	prt_proxy_t *	proxy,
		prt_request_t *	request
	)

recv to a channel

Parameters

proxy
request

Definition at line 143 of file prt_proxy.c.

References prt_packet_s::data, icl_hash_find(), prt_packet_s::num_refs, prt_request_s::packet, prt_assert, prt_channel_push(), prt_tuple_new2, prt_request_s::status, and prt_proxy_s::tags_hash.

 {
     int count;
     int retval;
     // Find the message size and realloc the packet data.
     retval = MPI_Get_count(&request->status,  MPI_BYTE, &count);
     prt_assert(retval == MPI_SUCCESS, "MPI_Get_count failed");
     request->packet->data = realloc(request->packet->data, count);
 
     // Find the channel and push.
     int source = request->status.MPI_SOURCE;
     int tag = request->status.MPI_TAG;
     int *source_tag = prt_tuple_new2(source, tag);
     prt_channel_t *channel = icl_hash_find(proxy->tags_hash, source_tag);
     free(source_tag);
 
     // Reset num_refs and push.
     request->packet->num_refs = 0;
     prt_channel_push(channel, request->packet);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void prt_proxy_run ( prt_proxy_t * proxy )

communication proxy production cycle Serves communication requests of local worker threads until shut down.

Parameters

proxy

Definition at line 171 of file prt_proxy.c.

References icl_list_s::data, prt_thread_s::finished, icl_deque_delete(), icl_deque_first(), icl_deque_size(), icl_list_append(), icl_list_delete(), icl_list_destroy(), icl_list_first(), icl_list_size(), prt_proxy_s::max_packet_size, prt_proxy_s::num_threads, prt_request_s::packet, prt_packet_new(), prt_packet_release(), PRT_PROXY_MAX_RECVS_PER_THREAD, PRT_PROXY_MAX_SENDS_PER_THREAD, prt_proxy_recv(), prt_request_cancel(), prt_request_destroy(), prt_request_new(), prt_request_recv(), prt_request_send(), prt_request_test(), prt_proxy_s::recvs_posted, prt_proxy_s::sends_posted, prt_proxy_s::sends_requested, prt_vsa_s::thread, and prt_proxy_s::vsa.

 {
     int sends_posted;
     int sends_requested;
     int threads_finished;
     do {
         int i;
         // Post another send.
         for (i = 0; i < proxy->num_threads; i++) {
             if (icl_list_size(proxy->sends_posted[i]) <
                 PRT_PROXY_MAX_SENDS_PER_THREAD) {
                 icl_node_t *node = icl_deque_first(proxy->sends_requested[i]);
                 if (node != NULL) {
                     prt_request_t *request = (prt_request_t*)node->data;
                     // Post the send request.
                     prt_request_send(request);
                     // Move from the list of requested to the list of posted.
                     // Only moving the request, not destroying the request.
                     icl_deque_delete(proxy->sends_requested[i], node, NULL);
                     icl_list_append(proxy->sends_posted[i], request);
                 }
             }
         }
         // Complete another send.
         for (i = 0; i < proxy->num_threads; i++) {
             icl_node_t *node = icl_list_first(proxy->sends_posted[i]);
             if (node != NULL) {
                 prt_request_t *request = (prt_request_t*)node->data;
                 // If send request completed.
                 if (prt_request_test(request)) {
                     // Release the packet.
                     prt_packet_release(request->packet);
                     // Remove from the list & destroy the request.
                     icl_list_delete(proxy->sends_posted[i], node,
                         (void(*)(void*))prt_request_destroy);
                 }
             }
         }
         // Post another receive.
         if (icl_list_size(proxy->recvs_posted) <
             PRT_PROXY_MAX_RECVS_PER_THREAD*proxy->num_threads) {
             prt_packet_t *packet = prt_packet_new(proxy->max_packet_size);
             prt_request_t *request =
                 prt_request_new(
                     packet, proxy->max_packet_size,
                     MPI_BYTE, MPI_ANY_SOURCE, MPI_ANY_TAG);
             prt_request_recv(request);
             icl_list_append(proxy->recvs_posted, request);
         }
         // Complete another receive.
         icl_node_t *node = icl_list_first(proxy->recvs_posted);
         if (node != NULL) {
             prt_request_t *request = (prt_request_t*)node->data;
             // IF recv request completed.
             if (prt_request_test(request)) {
                 // Put the packet in a channel.
                 prt_proxy_recv(proxy, request);
                 // Remove from the list & destroy the request.
                 icl_list_delete(proxy->recvs_posted, node,
                     (void(*)(void*))prt_request_destroy);
             }
         }
         threads_finished = 1;
         // Check if threads finished.
         for (i = 0; i < proxy->num_threads; i++)
             if (!proxy->vsa->thread[i]->finished)
                 threads_finished = 0;
 
         sends_requested = 0;
         // Check if sends pending.
         for (i = 0; i < proxy->num_threads; i++)
             if (icl_deque_size(proxy->sends_requested[i]) > 0)
                 sends_requested = 1;
 
         sends_posted = 0;
         // Check if sends pending.
         for (i = 0; i < proxy->num_threads; i++)
             if (icl_list_size(proxy->sends_posted[i]) > 0)
                 sends_posted = 1;
     }
     while (!threads_finished || sends_requested || sends_posted);
     // WHILE threads not finished OR sends pending.
 
     // Destroy the list of pending receives.
     // Cancel and free all requests. Release all packets.
     icl_list_destroy(proxy->recvs_posted, (void(*))(void*)prt_request_cancel);
 
     // Barrier before any other MPI call
     // picks up one of those pending recvs.
     MPI_Barrier(MPI_COMM_WORLD);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void prt_proxy_send	(	prt_proxy_t *	proxy,
		int	thread_rank,
		prt_channel_t *	channel
	)

send from a channel

Parameters

proxy
thread_rank
channel	channel requesting a send

Definition at line 122 of file prt_proxy.c.

References prt_channel_s::count, prt_channel_s::datatype, prt_channel_s::dst_node, icl_deque_append(), prt_channel_pop(), prt_request_new(), prt_proxy_s::sends_requested, and prt_channel_s::tag.

 {
     // Pop the packet from the channel.
     prt_packet_t *packet = prt_channel_pop(channel);
     // Create a request.
     prt_request_t *request =
         prt_request_new(
             packet,
             channel->count, channel->datatype,
             channel->dst_node, channel->tag);
     // Queue for send.
     icl_deque_append(proxy->sends_requested[thread_rank], request);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

Data Structures

Macros

Typedefs

Functions

Detailed Description

Macro Definition Documentation

Typedef Documentation

Function Documentation