prt_proxy.c

Go to the documentation of this file.

#include "prt_proxy.h"


prt_proxy_t* prt_proxy_new(int num_threads)
{
    // 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;
}


void prt_proxy_delete(prt_proxy_t *proxy)
{
    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);
}


void prt_proxy_max_packet_size(prt_proxy_t *proxy, prt_channel_t *channel)
{
    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;
}


void prt_proxy_send(prt_proxy_t *proxy, int thread_rank, prt_channel_t *channel)
{
    // 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);
}


void prt_proxy_recv(prt_proxy_t *proxy, prt_request_t *request)
{
    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);
}


void prt_proxy_run(prt_proxy_t *proxy)
{
    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);
}