mpi.c

Go to the documentation of this file.

 
#include <distributed/mpi.h>
 
#ifdef ROOTSIM_MPI
 
#include <core/core.h>
#include <core/sync.h>
#include <datatypes/array.h>
#include <datatypes/msg_queue.h>
#include <datatypes/remote_msg_map.h>
#include <gvt/gvt.h>
#include <gvt/termination.h>
#include <mm/msg_allocator.h>
 
#include <mpi.h>
 
#define MSG_CTRL_RAW (MSG_CTRL_TERMINATION + 1)
 
#ifdef ROOTSIM_MPI_SERIALIZABLE
 
static bool mpi_serialize;
static spinlock_t mpi_spinlock;
 
#define mpi_lock()      if (mpi_serialize) spin_lock(&mpi_spinlock)
#define mpi_unlock()    if (mpi_serialize) spin_unlock(&mpi_spinlock)
#define mpi_trylock()   (!mpi_serialize || spin_trylock(&mpi_spinlock))
 
#else
 
#define mpi_lock()
#define mpi_unlock()
#define mpi_trylock()   (true)
 
#endif
 
static void comm_error_handler(MPI_Comm *comm, int *err_code_p, ...)
{
        (void) comm;
        log_log(LOG_FATAL, "MPI error with code %d!", *err_code_p);
 
        int err_len;
        char err_str[MPI_MAX_ERROR_STRING];
        MPI_Error_string(*err_code_p, err_str, &err_len);
        log_log(LOG_FATAL, "MPI error msg is %s ", err_str);
 
        exit(-1);
}
 
void mpi_global_init(int *argc_p, char ***argv_p)
{
        int thread_lvl = MPI_THREAD_SINGLE;
        MPI_Init_thread(argc_p, argv_p, MPI_THREAD_MULTIPLE, &thread_lvl);
 
        if (thread_lvl < MPI_THREAD_MULTIPLE) {
                if (thread_lvl < MPI_THREAD_SERIALIZED) {
                        log_log(LOG_FATAL,
                                "This MPI implementation does not support threaded access");
                        abort();
                } else {
#ifdef ROOTSIM_MPI_SERIALIZABLE
                        mpi_serialize = true;
                        spin_init(&mpi_spinlock);
#else
                        log_log(LOG_FATAL,
                                "This MPI implementation only supports serialized calls: you need to build ROOT-Sim with -Dserialized_mpi=true");
                        abort();
#endif
                }
        }
 
        MPI_Errhandler err_handler;
        if (MPI_Comm_create_errhandler(comm_error_handler, &err_handler)) {
                log_log(LOG_FATAL, "Unable to create MPI error handler");
                abort();
        }
 
        MPI_Comm_set_errhandler(MPI_COMM_WORLD, err_handler);
 
        int helper;
        MPI_Comm_rank(MPI_COMM_WORLD, &helper);
        nid = helper;
        MPI_Comm_size(MPI_COMM_WORLD, &helper);
        n_nodes = helper;
}
 
void mpi_global_fini(void)
{
        MPI_Errhandler err_handler;
        MPI_Comm_get_errhandler(MPI_COMM_WORLD, &err_handler);
        MPI_Errhandler_free(&err_handler);
 
        MPI_Finalize();
}
 
void mpi_remote_msg_send(struct lp_msg *msg, nid_t dest_nid)
{
        msg->msg_id = msg_id_get(msg, gvt_phase_get());
        gvt_on_remote_msg_send(dest_nid);
 
        mpi_lock();
        MPI_Request req;
        MPI_Isend(msg, msg_bare_size(msg), MPI_BYTE, dest_nid, 0,
                MPI_COMM_WORLD, &req);
        MPI_Request_free(&req);
        mpi_unlock();
}
 
void mpi_remote_anti_msg_send(struct lp_msg *msg, nid_t dest_nid)
{
        msg_id_anti_phase_set(msg->msg_id, gvt_phase_get());
        gvt_on_remote_msg_send(dest_nid);
 
        mpi_lock();
        MPI_Request req;
        MPI_Isend(&msg->msg_id, sizeof(msg->msg_id), MPI_BYTE, dest_nid, 0,
                MPI_COMM_WORLD, &req);
        MPI_Request_free(&req);
        mpi_unlock();
}
 
void mpi_control_msg_broadcast(enum msg_ctrl_tag ctrl)
{
        MPI_Request req;
        nid_t i = n_nodes;
        mpi_lock();
        while (i--) {
                if(i == nid)
                        continue;
                MPI_Isend(NULL, 0, MPI_BYTE, i, ctrl, MPI_COMM_WORLD, &req);
                MPI_Request_free(&req);
        }
        mpi_unlock();
}
 
void mpi_control_msg_send_to(enum msg_ctrl_tag ctrl, nid_t dest)
{
        MPI_Request req;
        mpi_lock();
        MPI_Isend(NULL, 0, MPI_BYTE, dest, ctrl, MPI_COMM_WORLD, &req);
        MPI_Request_free(&req);
        mpi_unlock();
}
 
void mpi_remote_msg_handle(void)
{
        int pending;
        MPI_Message mpi_msg;
        MPI_Status status;
 
        while (1) {
                if (!mpi_trylock())
                        return;
 
                MPI_Improbe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD,
                        &pending, &mpi_msg, &status);
 
                if (!pending) {
                        mpi_unlock();
                        return;
                }
 
                if (unlikely(status.MPI_TAG)) {
                        MPI_Mrecv(NULL, 0, MPI_BYTE, &mpi_msg, MPI_STATUS_IGNORE);
                        mpi_unlock();
                        switch(status.MPI_TAG){
                        case MSG_CTRL_GVT_START:
                                gvt_on_start_ctrl_msg();
                                break;
                        case MSG_CTRL_GVT_DONE:
                                gvt_on_done_ctrl_msg();
                                break;
                        case MSG_CTRL_TERMINATION:
                                termination_on_ctrl_msg();
                                break;
                        }
                } else {
                        int size;
                        MPI_Get_count(&status, MPI_BYTE, &size);
 
                        if (unlikely(size == sizeof(uintptr_t))) {
                                uintptr_t anti_id;
                                MPI_Mrecv(&anti_id, size, MPI_BYTE, &mpi_msg,
                                        MPI_STATUS_IGNORE);
                                mpi_unlock();
 
                                remote_msg_map_match(anti_id,
                                        status.MPI_SOURCE, NULL);
                                gvt_on_remote_msg_receive(
                                        msg_id_anti_phase_get(anti_id));
                        } else {
                                mpi_unlock();
 
                                struct lp_msg *msg = msg_allocator_alloc(size -
                                        offsetof(struct lp_msg, pl));
 
                                mpi_lock();
                                MPI_Mrecv(msg, size, MPI_BYTE, &mpi_msg,
                                        MPI_STATUS_IGNORE);
                                mpi_unlock();
 
                                uintptr_t msg_id = msg->msg_id;
                                atomic_store_explicit(&msg->flags, 0U,
                                        memory_order_relaxed);
                                remote_msg_map_match(msg_id,
                                        status.MPI_SOURCE, msg);
                                msg_queue_insert(msg);
                                gvt_on_remote_msg_receive(
                                        msg_id_phase_get(msg_id));
                        }
                }
        }
}
 
static MPI_Request reduce_sum_scatter_req = MPI_REQUEST_NULL;
 
void mpi_reduce_sum_scatter(const unsigned node_vals[n_nodes], unsigned *result)
{
        mpi_lock();
        MPI_Ireduce_scatter_block(node_vals, result, 1, MPI_UNSIGNED, MPI_SUM,
                MPI_COMM_WORLD, &reduce_sum_scatter_req);
        mpi_unlock();
}
 
bool mpi_reduce_sum_scatter_done(void)
{
        int flag = 0;
        mpi_lock();
        MPI_Test(&reduce_sum_scatter_req, &flag, MPI_STATUS_IGNORE);
        mpi_unlock();
        return flag;
}
 
static MPI_Request reduce_min_req = MPI_REQUEST_NULL;
 
void mpi_reduce_min(simtime_t *node_min_p)
{
        static simtime_t min_buff;
        min_buff = *node_min_p;
        mpi_lock();
        MPI_Iallreduce(&min_buff, node_min_p, 1, MPI_DOUBLE, MPI_MIN,
                MPI_COMM_WORLD, &reduce_min_req);
        mpi_unlock();
}
 
bool mpi_reduce_min_done(void)
{
        int flag = 0;
        mpi_lock();
        MPI_Test(&reduce_min_req, &flag, MPI_STATUS_IGNORE);
        mpi_unlock();
        return flag;
}
 
void mpi_node_barrier(void)
{
        mpi_lock();
        MPI_Barrier(MPI_COMM_WORLD);
        mpi_unlock();
}
 
void mpi_blocking_data_send(const void *data, int data_size, nid_t dest)
{
        mpi_lock();
        MPI_Send(data, data_size, MPI_BYTE, dest, MSG_CTRL_RAW, MPI_COMM_WORLD);
        mpi_unlock();
}
 
void *mpi_blocking_data_rcv(int *data_size_p, nid_t src)
{
        MPI_Status status;
        MPI_Message mpi_msg;
        mpi_lock();
        MPI_Mprobe(src, MSG_CTRL_RAW, MPI_COMM_WORLD, &mpi_msg, &status);
        int data_size;
        MPI_Get_count(&status, MPI_BYTE, &data_size);
        char *ret = mm_alloc(data_size);
        MPI_Mrecv(ret, data_size, MPI_BYTE, &mpi_msg, MPI_STATUS_IGNORE);
        if (data_size_p != NULL)
                *data_size_p = data_size;
        mpi_unlock();
        return ret;
}
 
#endif