prt_proxy.c

Go to the documentation of this file.

#include "prt_proxy.h"


prt_proxy_t* prt_proxy_new(int num_agents)
{
    // 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_agents = num_agents;
    proxy->max_channel_size = 0;
    proxy->num_callbacks = 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");

    int i;
    // Create lists of requested sends.
    proxy->sends_requested =
        (icl_deque_t**)malloc(num_agents*sizeof(icl_deque_t*));
    prt_assert(proxy->sends_requested != NULL, "malloc failed");
    for (i = 0; i < num_agents; i++) {
        proxy->sends_requested[i] = icl_deque_new();
        prt_assert(proxy->sends_requested[i] != NULL, "icl_deque_new failed");
    }

    // Allocate lists of posted sends.
    proxy->sends_posted =
        (icl_list_t**)malloc(num_agents*sizeof(icl_list_t*));
    prt_assert(proxy->sends_posted != NULL, "malloc failed");
    for (i = 0; i < num_agents; i++) {
        proxy->sends_posted[i] = icl_list_new();
        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");

    // Create deque of local transfers.
    proxy->transfers = icl_deque_new();
    prt_assert(proxy->transfers != NULL, "icl_deque_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 lists of sends requested.
    for (i = 0; i < proxy->num_agents; 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");
    }
    free(proxy->sends_requested);

    // Destroy lists of sends posted.
    for (i = 0; i < proxy->num_agents; i++) {
        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(proxy->sends_posted);

    // Destroy the deque of local transfers.
    size = icl_deque_size(proxy->transfers);
    prt_assert(size == 0, "destroying non-empty deque");
    status = icl_deque_destroy(proxy->transfers, NULL);
    prt_assert(status == 0, "icl_deque_destroy failed");

    // Free the proxy.
    free(proxy);
}


void prt_proxy_max_channel_size(prt_proxy_t *proxy, prt_channel_t *channel)
{
    // Quit if no proxy.
    if (proxy == NULL)
        return;

    // Store the largest packet size.
    if (channel->size > proxy->max_channel_size)
        proxy->max_channel_size = channel->size;
}


void prt_proxy_recv(prt_proxy_t *proxy, prt_request_t *request)
{
    int count;
    int retval;
    // Find the message size.
    retval = MPI_Get_count(&request->status,  MPI_BYTE, &count);
    prt_assert(retval == MPI_SUCCESS, "MPI_Get_count failed");
    // Resize the packet if necessary.
    if (count != request->packet->size)
        prt_packet_resize_host(request->packet, count);

    // Locate the destination channel.
    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);

    // IF host VDP.
    if (channel->dst_vdp->location == PRT_LOCATION_HOST) {
        // Place packet in the channel's dequeue.
        icl_deque_append(channel->packets, (void*)request->packet);
    }
    // ELSE if device VDP
    else {
        // Place a host-to-device transfer in the channel's stream.
        prt_packet_host_to_device(request->packet, channel);
    }
}


void prt_proxy_mpi(prt_proxy_t *proxy)
{
    int i;
    // Post another send.
    for (i = 0; i < proxy->num_agents; i++) {
        if (icl_list_size(proxy->sends_posted[i]) <
            PRT_PROXY_MAX_SENDS_PER_AGENT) {
            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_agents; 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_host(request->packet);
                // Remove from the list & destroy the request.
                icl_list_delete(proxy->sends_posted[i], node,
                    (void(*)(void*))prt_request_delete);
            }
        }
    }
    // Post another receive.
    if (icl_list_size(proxy->recvs_posted) <
        PRT_PROXY_MAX_RECVS_PER_AGENT*proxy->num_agents) {
        prt_packet_t *packet =
            prt_packet_new_host(proxy->max_channel_size, NULL);
        prt_request_t *request =
            prt_request_new(
                packet, proxy->max_channel_size,
                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_delete);
        }
    }
}


void prt_proxy_cuda(prt_proxy_t *proxy)
{
    int dev;
    // Cycle devices.
    for (dev = 0; dev < proxy->vsa->num_devices; dev++)
        prt_device_cycle(proxy->vsa->device[dev]);

    // Issue all local comms.
    icl_node_t *node;
    while ((node = icl_deque_first(proxy->transfers)) != NULL) {
        prt_transfer_t *transfer =
            (prt_transfer_t*)node->data;
        switch (transfer->direction) {
            case PRT_HOST_TO_DEVICE:
                prt_packet_host_to_device(
                    transfer->packet, transfer->channel);
                break;
            case PRT_DEVICE_TO_HOST:
                prt_packet_device_to_host(
                    transfer->packet, transfer->channel);
                break;
            case PRT_DEVICE_TO_DEVICE:
                prt_packet_device_to_device(
                    transfer->packet, transfer->channel);
                break;
            case PRT_DEVICE_MPI_TO_HOST:
                prt_packet_device_mpi_to_host(
                    transfer->packet,
                    transfer->channel,
                    transfer->agent);
                break;
            case PRT_DEVICE_MPI_FROM_HOST: {
                prt_request_t *request =
                    prt_request_new(
                        transfer->packet,
                        transfer->packet->size,
                        transfer->channel->dst_node,
                        transfer->channel->tag);
                icl_deque_append(
                    proxy->sends_requested[transfer->agent], request);
                break;
            }
            case PRT_DEVICE_PACKET_RELEASE:
                prt_packet_release_device(transfer->packet);
                break;
        }
        icl_deque_delete(proxy->transfers, node, NULL);
        prt_transfer_delete(transfer);
    }
}


double prt_proxy_run(prt_proxy_t *proxy)
{
    // MPI barrier
    svg_trace_start_cpu(0);
    MPI_Barrier(MPI_COMM_WORLD);
    svg_trace_stop_cpu(0, -Honeydew);

    // Barrier threads.
    svg_trace_start_cpu(0);
    pthread_barrier_wait(&proxy->vsa->barrier);
    svg_trace_stop_cpu(0, -Azure);
    double start = get_time_of_day();

    int dev;
    // Register initial device and DMA events.
    for (dev = 0; dev < proxy->vsa->num_devices; dev++) {
        cudaSetDevice(dev);
        svg_trace_start_gpu(0);
        svg_trace_start_dma(0);
        svg_trace_stop_gpu(0, Black);
        svg_trace_stop_dma(0, Black);
    }
    int sends_posted;
    int sends_requested;
    int threads_finished;
    int devices_finished;
    do {
        // Cycle MPI.
        if (proxy->vsa->num_nodes > 1)
            prt_proxy_mpi(proxy);

        // Cycle CUDA.
        if (proxy->vsa->num_devices > 0)
            prt_proxy_cuda(proxy);

        int i;
        devices_finished = 1;
        // Check if devices finished.
        for (i = 0; i < proxy->vsa->num_devices; i++)
            if (!proxy->vsa->device[i]->finished)
                devices_finished = 0;

        threads_finished = 1;
        // Check if threads finished.
        for (i = 0; i < proxy->vsa->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_agents; 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_agents; i++)
            if (icl_list_size(proxy->sends_posted[i]) > 0)
                sends_posted = 1;
    }
    while (
        !threads_finished || !devices_finished ||
        sends_requested || sends_posted ||
        icl_deque_size(proxy->transfers) > 0 || proxy->num_callbacks > 0);
    // 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);

    // Synchronize each device.
    for (dev = 0; dev < proxy->vsa->num_devices; dev++) {
        cudaSetDevice(dev);
        cudaDeviceSynchronize();
    }
    // thread barrier
    svg_trace_start_cpu(0);
    pthread_barrier_wait(&proxy->vsa->barrier);
    svg_trace_stop_cpu(0, -Azure);

    // Barrier before any other MPI call
    // picks up one of those pending recvs.
    svg_trace_start_cpu(0);
    MPI_Barrier(MPI_COMM_WORLD);
    svg_trace_stop_cpu(0, -Honeydew);

    // Return the time between the end of the first MPI barrier
    // and the end of the second MPI barrier.
    double stop = get_time_of_day();
    return (stop - start);
}