ports/openmpi
ports/openmpi
1
1
Форкнуть 0
Код Релизы Активность
openmpi/ompi/mca/common/monitoring/monitoring_prof.c
Noah Evans ee3517427e Add threads framework 
Add a framework to support different types of threading models including
user space thread packages such as Qthreads and argobot:

https://github.com/pmodels/argobots

https://github.com/Qthreads/qthreads

The default threading model is pthreads.  Alternate thread models are
specificed at configure time using the --with-threads=X option.

The framework is static.  The theading model to use is selected at
Open MPI configure/build time.

mca/threads: implement Argobots threading layer

config: fix thread configury

- Add double quotations
- Change Argobot to Argobots
config: implement Argobots check

If the poll time is too long, MPI hangs.

This quick fix just sets it to 0, but it is not good for the
Pthreads version. Need to find a good way to abstract it.

Note that even 1 (= 1 millisecond) causes disastrous performance
degradation.

rework threads MCA framework configury

It now works more like the ompi/mca/rte configury,
modulo some edge items that are special for threading package
linking, etc.

qthreads module
some argobots cleanup

Signed-off-by: Noah Evans <noah.evans@gmail.com>
Signed-off-by: Shintaro Iwasaki <siwasaki@anl.gov>
Signed-off-by: Howard Pritchard <howardp@lanl.gov>
2020-03-27 10:15:45 -06:00

457 строки
19 KiB
C
Исходник Ответственный История

/*
* Copyright (c) 2013-2017 The University of Tennessee and The University
* of Tennessee Research Foundation. All rights
* reserved.
* Copyright (c) 2013-2017 Inria. All rights reserved.
* Copyright (c) 2013-2015 Bull SAS. All rights reserved.
* Copyright (c) 2016-2018 Cisco Systems, Inc. All rights reserved.
* Copyright (c) 2017 Research Organization for Information Science
* and Technology (RIST). All rights reserved.
* $COPYRIGHT$
*
* Additional copyrights may follow
*
* $HEADER$
*/
/*
pml monitoring PMPI profiler
Designed by:
George Bosilca <bosilca@icl.utk.edu>
Emmanuel Jeannot <emmanuel.jeannot@inria.fr>
Guillaume Papauré <guillaume.papaure@bull.net>
Clément Foyer <clement.foyer@inria.fr>
Contact the authors for questions.
To be run as:
mpirun -np 4 \
--mca pml_monitoring_enable 1 \
-x LD_PRELOAD=ompi_install_dir/lib/ompi_monitoring_prof.so \
./my_app
...
...
...
writing 4x4 matrix to monitoring_msg.mat
writing 4x4 matrix to monitoring_size.mat
writing 4x4 matrix to monitoring_avg.mat
*/
#include "ompi_config.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdbool.h>
#if OMPI_BUILD_FORTRAN_BINDINGS
// Set these #defines in the same way that
// ompi/mpi/fortran/mpif-h/Makefile.am does when compiling the real
// Fortran mpif.h bindings. They set behaviors in the Fortran header
// files so that we can compile properly.
#define OMPI_BUILD_MPI_PROFILING 0
#define OMPI_COMPILING_FORTRAN_WRAPPERS 1
#endif
#include "opal/mca/threads/thread_usage.h"
#include "ompi/include/mpi.h"
#include "ompi/mpi/fortran/base/constants.h"
#include "ompi/mpi/fortran/base/fint_2_int.h"
#if OMPI_BUILD_FORTRAN_BINDINGS
#include "ompi/mpi/fortran/mpif-h/bindings.h"
#endif
static MPI_T_pvar_session session;
static int comm_world_size;
static int comm_world_rank;
struct monitoring_result
{
char * pvar_name;
int pvar_idx;
MPI_T_pvar_handle pvar_handle;
size_t * vector;
};
typedef struct monitoring_result monitoring_result;
/* PML Sent */
static monitoring_result pml_counts;
static monitoring_result pml_sizes;
/* OSC Sent */
static monitoring_result osc_scounts;
static monitoring_result osc_ssizes;
/* OSC Recv */
static monitoring_result osc_rcounts;
static monitoring_result osc_rsizes;
/* COLL Sent/Recv */
static monitoring_result coll_counts;
static monitoring_result coll_sizes;
static int write_mat(char *, size_t *, unsigned int);
static void init_monitoring_result(const char *, monitoring_result *);
static void start_monitoring_result(monitoring_result *);
static void stop_monitoring_result(monitoring_result *);
static void get_monitoring_result(monitoring_result *);
static void destroy_monitoring_result(monitoring_result *);
int MPI_Init(int* argc, char*** argv)
{
int result, MPIT_result;
int provided;
result = PMPI_Init(argc, argv);
PMPI_Comm_size(MPI_COMM_WORLD, &comm_world_size);
PMPI_Comm_rank(MPI_COMM_WORLD, &comm_world_rank);
MPIT_result = MPI_T_init_thread(MPI_THREAD_SINGLE, &provided);
if (MPIT_result != MPI_SUCCESS) {
fprintf(stderr, "ERROR : failed to intialize MPI_T interface, preventing to get monitoring results: check your OpenMPI installation\n");
PMPI_Abort(MPI_COMM_WORLD, MPIT_result);
}
MPIT_result = MPI_T_pvar_session_create(&session);
if (MPIT_result != MPI_SUCCESS) {
fprintf(stderr, "ERROR : failed to create MPI_T session, preventing to get monitoring results: check your OpenMPI installation\n");
PMPI_Abort(MPI_COMM_WORLD, MPIT_result);
}
init_monitoring_result("pml_monitoring_messages_count", &pml_counts);
init_monitoring_result("pml_monitoring_messages_size", &pml_sizes);
init_monitoring_result("osc_monitoring_messages_sent_count", &osc_scounts);
init_monitoring_result("osc_monitoring_messages_sent_size", &osc_ssizes);
init_monitoring_result("osc_monitoring_messages_recv_count", &osc_rcounts);
init_monitoring_result("osc_monitoring_messages_recv_size", &osc_rsizes);
init_monitoring_result("coll_monitoring_messages_count", &coll_counts);
init_monitoring_result("coll_monitoring_messages_size", &coll_sizes);
start_monitoring_result(&pml_counts);
start_monitoring_result(&pml_sizes);
start_monitoring_result(&osc_scounts);
start_monitoring_result(&osc_ssizes);
start_monitoring_result(&osc_rcounts);
start_monitoring_result(&osc_rsizes);
start_monitoring_result(&coll_counts);
start_monitoring_result(&coll_sizes);
return result;
}
int MPI_Finalize(void)
{
int result, MPIT_result;
size_t * exchange_count_matrix_1 = NULL;
size_t * exchange_size_matrix_1 = NULL;
size_t * exchange_count_matrix_2 = NULL;
size_t * exchange_size_matrix_2 = NULL;
size_t * exchange_all_size_matrix = NULL;
size_t * exchange_all_count_matrix = NULL;
size_t * exchange_all_avg_matrix = NULL;
stop_monitoring_result(&pml_counts);
stop_monitoring_result(&pml_sizes);
stop_monitoring_result(&osc_scounts);
stop_monitoring_result(&osc_ssizes);
stop_monitoring_result(&osc_rcounts);
stop_monitoring_result(&osc_rsizes);
stop_monitoring_result(&coll_counts);
stop_monitoring_result(&coll_sizes);
get_monitoring_result(&pml_counts);
get_monitoring_result(&pml_sizes);
get_monitoring_result(&osc_scounts);
get_monitoring_result(&osc_ssizes);
get_monitoring_result(&osc_rcounts);
get_monitoring_result(&osc_rsizes);
get_monitoring_result(&coll_counts);
get_monitoring_result(&coll_sizes);
if (0 == comm_world_rank) {
exchange_count_matrix_1 = (size_t *) calloc(comm_world_size * comm_world_size, sizeof(size_t));
exchange_size_matrix_1 = (size_t *) calloc(comm_world_size * comm_world_size, sizeof(size_t));
exchange_count_matrix_2 = (size_t *) calloc(comm_world_size * comm_world_size, sizeof(size_t));
exchange_size_matrix_2 = (size_t *) calloc(comm_world_size * comm_world_size, sizeof(size_t));
exchange_all_size_matrix = (size_t *) calloc(comm_world_size * comm_world_size, sizeof(size_t));
exchange_all_count_matrix = (size_t *) calloc(comm_world_size * comm_world_size, sizeof(size_t));
exchange_all_avg_matrix = (size_t *) calloc(comm_world_size * comm_world_size, sizeof(size_t));
}
/* Gather PML and COLL results */
PMPI_Gather(pml_counts.vector, comm_world_size, MPI_UNSIGNED_LONG, exchange_count_matrix_1, comm_world_size, MPI_UNSIGNED_LONG, 0, MPI_COMM_WORLD);
PMPI_Gather(pml_sizes.vector, comm_world_size, MPI_UNSIGNED_LONG, exchange_size_matrix_1, comm_world_size, MPI_UNSIGNED_LONG, 0, MPI_COMM_WORLD);
PMPI_Gather(coll_counts.vector, comm_world_size, MPI_UNSIGNED_LONG, exchange_count_matrix_2, comm_world_size, MPI_UNSIGNED_LONG, 0, MPI_COMM_WORLD);
PMPI_Gather(coll_sizes.vector, comm_world_size, MPI_UNSIGNED_LONG, exchange_size_matrix_2, comm_world_size, MPI_UNSIGNED_LONG, 0, MPI_COMM_WORLD);
if (0 == comm_world_rank) {
int i, j;
for (i = 0; i < comm_world_size; ++i) {
for (j = i + 1; j < comm_world_size; ++j) {
/* Reduce PML results */
exchange_count_matrix_1[i * comm_world_size + j] = exchange_count_matrix_1[j * comm_world_size + i] = (exchange_count_matrix_1[i * comm_world_size + j] + exchange_count_matrix_1[j * comm_world_size + i]) / 2;
exchange_size_matrix_1[i * comm_world_size + j] = exchange_size_matrix_1[j * comm_world_size + i] = (exchange_size_matrix_1[i * comm_world_size + j] + exchange_size_matrix_1[j * comm_world_size + i]) / 2;
if (exchange_count_matrix_1[i * comm_world_size + j] != 0)
exchange_all_size_matrix[i * comm_world_size + j] = exchange_all_size_matrix[j * comm_world_size + i] = exchange_size_matrix_1[i * comm_world_size + j] / exchange_count_matrix_1[i * comm_world_size + j];
/* Reduce COLL results */
exchange_count_matrix_2[i * comm_world_size + j] = exchange_count_matrix_2[j * comm_world_size + i] = (exchange_count_matrix_2[i * comm_world_size + j] + exchange_count_matrix_2[j * comm_world_size + i]) / 2;
exchange_size_matrix_2[i * comm_world_size + j] = exchange_size_matrix_2[j * comm_world_size + i] = (exchange_size_matrix_2[i * comm_world_size + j] + exchange_size_matrix_2[j * comm_world_size + i]) / 2;
if (exchange_count_matrix_2[i * comm_world_size + j] != 0)
exchange_all_count_matrix[i * comm_world_size + j] = exchange_all_count_matrix[j * comm_world_size + i] = exchange_size_matrix_2[i * comm_world_size + j] / exchange_count_matrix_2[i * comm_world_size + j];
}
}
/* Write PML matrices */
write_mat("monitoring_pml_msg.mat", exchange_count_matrix_1, comm_world_size);
write_mat("monitoring_pml_size.mat", exchange_size_matrix_1, comm_world_size);
write_mat("monitoring_pml_avg.mat", exchange_all_size_matrix, comm_world_size);
/* Write COLL matrices */
write_mat("monitoring_coll_msg.mat", exchange_count_matrix_2, comm_world_size);
write_mat("monitoring_coll_size.mat", exchange_size_matrix_2, comm_world_size);
write_mat("monitoring_coll_avg.mat", exchange_all_count_matrix, comm_world_size);
/* Aggregate PML and COLL in ALL matrices */
for (i = 0; i < comm_world_size; ++i) {
for (j = i + 1; j < comm_world_size; ++j) {
exchange_all_size_matrix[i * comm_world_size + j] = exchange_all_size_matrix[j * comm_world_size + i] = exchange_size_matrix_1[i * comm_world_size + j] + exchange_size_matrix_2[i * comm_world_size + j];
exchange_all_count_matrix[i * comm_world_size + j] = exchange_all_count_matrix[j * comm_world_size + i] = exchange_count_matrix_1[i * comm_world_size + j] + exchange_count_matrix_2[i * comm_world_size + j];
}
}
}
/* Gather OSC results */
PMPI_Gather(osc_scounts.vector, comm_world_size, MPI_UNSIGNED_LONG, exchange_count_matrix_1, comm_world_size, MPI_UNSIGNED_LONG, 0, MPI_COMM_WORLD);
PMPI_Gather(osc_ssizes.vector, comm_world_size, MPI_UNSIGNED_LONG, exchange_size_matrix_1, comm_world_size, MPI_UNSIGNED_LONG, 0, MPI_COMM_WORLD);
PMPI_Gather(osc_rcounts.vector, comm_world_size, MPI_UNSIGNED_LONG, exchange_count_matrix_2, comm_world_size, MPI_UNSIGNED_LONG, 0, MPI_COMM_WORLD);
PMPI_Gather(osc_rsizes.vector, comm_world_size, MPI_UNSIGNED_LONG, exchange_size_matrix_2, comm_world_size, MPI_UNSIGNED_LONG, 0, MPI_COMM_WORLD);
if (0 == comm_world_rank) {
int i, j;
for (i = 0; i < comm_world_size; ++i) {
for (j = i + 1; j < comm_world_size; ++j) {
/* Reduce OSC results */
exchange_count_matrix_1[i * comm_world_size + j] = exchange_count_matrix_1[j * comm_world_size + i] = (exchange_count_matrix_1[i * comm_world_size + j] + exchange_count_matrix_1[j * comm_world_size + i] + exchange_count_matrix_2[i * comm_world_size + j] + exchange_count_matrix_2[j * comm_world_size + i]) / 2;
exchange_size_matrix_1[i * comm_world_size + j] = exchange_size_matrix_1[j * comm_world_size + i] = (exchange_size_matrix_1[i * comm_world_size + j] + exchange_size_matrix_1[j * comm_world_size + i] + exchange_size_matrix_2[i * comm_world_size + j] + exchange_size_matrix_2[j * comm_world_size + i]) / 2;
if (exchange_count_matrix_1[i * comm_world_size + j] != 0)
exchange_all_avg_matrix[i * comm_world_size + j] = exchange_all_avg_matrix[j * comm_world_size + i] = exchange_size_matrix_1[i * comm_world_size + j] / exchange_count_matrix_1[i * comm_world_size + j];
}
}
/* Write OSC matrices */
write_mat("monitoring_osc_msg.mat", exchange_count_matrix_1, comm_world_size);
write_mat("monitoring_osc_size.mat", exchange_size_matrix_1, comm_world_size);
write_mat("monitoring_osc_avg.mat", exchange_all_avg_matrix, comm_world_size);
/* Aggregate OSC in ALL matrices and compute AVG */
for (i = 0; i < comm_world_size; ++i) {
for (j = i + 1; j < comm_world_size; ++j) {
exchange_all_size_matrix[i * comm_world_size + j] = exchange_all_size_matrix[j * comm_world_size + i] += exchange_size_matrix_1[i * comm_world_size + j];
exchange_all_count_matrix[i * comm_world_size + j] = exchange_all_count_matrix[j * comm_world_size + i] += exchange_count_matrix_1[i * comm_world_size + j];
if (exchange_all_count_matrix[i * comm_world_size + j] != 0)
exchange_all_avg_matrix[i * comm_world_size + j] = exchange_all_avg_matrix[j * comm_world_size + i] = exchange_all_size_matrix[i * comm_world_size + j] / exchange_all_count_matrix[i * comm_world_size + j];
}
}
/* Write ALL matrices */
write_mat("monitoring_all_msg.mat", exchange_all_count_matrix, comm_world_size);
write_mat("monitoring_all_size.mat", exchange_all_size_matrix, comm_world_size);
write_mat("monitoring_all_avg.mat", exchange_all_avg_matrix, comm_world_size);
/* Free matrices */
free(exchange_count_matrix_1);
free(exchange_size_matrix_1);
free(exchange_count_matrix_2);
free(exchange_size_matrix_2);
free(exchange_all_count_matrix);
free(exchange_all_size_matrix);
free(exchange_all_avg_matrix);
}
destroy_monitoring_result(&pml_counts);
destroy_monitoring_result(&pml_sizes);
destroy_monitoring_result(&osc_scounts);
destroy_monitoring_result(&osc_ssizes);
destroy_monitoring_result(&osc_rcounts);
destroy_monitoring_result(&osc_rsizes);
destroy_monitoring_result(&coll_counts);
destroy_monitoring_result(&coll_sizes);
MPIT_result = MPI_T_pvar_session_free(&session);
if (MPIT_result != MPI_SUCCESS) {
fprintf(stderr, "WARNING : failed to free MPI_T session, monitoring results may be impacted : check your OpenMPI installation\n");
}
MPIT_result = MPI_T_finalize();
if (MPIT_result != MPI_SUCCESS) {
fprintf(stderr, "WARNING : failed to finalize MPI_T interface, monitoring results may be impacted : check your OpenMPI installation\n");
}
result = PMPI_Finalize();
return result;
}
void init_monitoring_result(const char * pvar_name, monitoring_result * res)
{
int count;
int MPIT_result;
MPI_Comm comm_world = MPI_COMM_WORLD;
res->pvar_name = strdup(pvar_name);
MPIT_result = MPI_T_pvar_get_index(res->pvar_name, MPI_T_PVAR_CLASS_SIZE, &(res->pvar_idx));
if (MPIT_result != MPI_SUCCESS) {
fprintf(stderr, "ERROR : cannot find monitoring MPI_T \"%s\" pvar, check that you have monitoring pml\n", pvar_name);
PMPI_Abort(MPI_COMM_WORLD, MPIT_result);
}
MPIT_result = MPI_T_pvar_handle_alloc(session, res->pvar_idx, comm_world, &(res->pvar_handle), &count);
if (MPIT_result != MPI_SUCCESS) {
fprintf(stderr, "ERROR : failed to allocate handle on \"%s\" pvar, check that you have monitoring pml\n", pvar_name);
PMPI_Abort(MPI_COMM_WORLD, MPIT_result);
}
if (count != comm_world_size) {
fprintf(stderr, "ERROR : COMM_WORLD has %d ranks \"%s\" pvar contains %d values, check that you have monitoring pml\n", comm_world_size, pvar_name, count);
PMPI_Abort(MPI_COMM_WORLD, count);
}
res->vector = (size_t *) malloc(comm_world_size * sizeof(size_t));
}
void start_monitoring_result(monitoring_result * res)
{
int MPIT_result;
MPIT_result = MPI_T_pvar_start(session, res->pvar_handle);
if (MPIT_result != MPI_SUCCESS) {
fprintf(stderr, "ERROR : failed to start handle on \"%s\" pvar, check that you have enabled the monitoring pml\n", res->pvar_name);
PMPI_Abort(MPI_COMM_WORLD, MPIT_result);
}
}
void stop_monitoring_result(monitoring_result * res)
{
int MPIT_result;
MPIT_result = MPI_T_pvar_stop(session, res->pvar_handle);
if (MPIT_result != MPI_SUCCESS) {
fprintf(stderr, "ERROR : failed to stop handle on \"%s\" pvar, check that you have enabled the monitoring pml\n", res->pvar_name);
MPI_Abort(MPI_COMM_WORLD, MPIT_result);
}
}
void get_monitoring_result(monitoring_result * res)
{
int MPIT_result;
MPIT_result = MPI_T_pvar_read(session, res->pvar_handle, res->vector);
if (MPIT_result != MPI_SUCCESS) {
fprintf(stderr, "ERROR : failed to read \"%s\" pvar, check that you have enabled the monitoring pml\n", res->pvar_name);
PMPI_Abort(MPI_COMM_WORLD, MPIT_result);
}
}
void destroy_monitoring_result(monitoring_result * res)
{
int MPIT_result;
MPIT_result = MPI_T_pvar_handle_free(session, &(res->pvar_handle));
if (MPIT_result != MPI_SUCCESS) {
printf("ERROR : failed to free handle on \"%s\" pvar, check that you have enabled the monitoring pml\n", res->pvar_name);
MPI_Abort(MPI_COMM_WORLD, MPIT_result);
}
free(res->pvar_name);
free(res->vector);
}
int write_mat(char * filename, size_t * mat, unsigned int dim)
{
FILE *matrix_file;
int i, j;
matrix_file = fopen(filename, "w");
if (!matrix_file) {
fprintf(stderr, "ERROR : failed to open \"%s\" file in write mode, check your permissions\n", filename);
return -1;
}
printf("writing %ux%u matrix to %s\n", dim, dim, filename);
for (i = 0; i < comm_world_size; ++i) {
for (j = 0; j < comm_world_size; ++j) {
fprintf(matrix_file, "%zu ", mat[i * comm_world_size + j]);
}
fprintf(matrix_file, "\n");
}
fflush(matrix_file);
fclose(matrix_file);
return 0;
}
/**
* MPI binding for fortran
*/
void monitoring_prof_mpi_init_f2c( MPI_Fint * );
void monitoring_prof_mpi_finalize_f2c( MPI_Fint * );
void monitoring_prof_mpi_init_f2c( MPI_Fint *ierr ) {
int c_ierr;
int argc = 0;
char ** argv = NULL;
c_ierr = MPI_Init(&argc, &argv);
if (NULL != ierr) *ierr = OMPI_INT_2_FINT(c_ierr);
}
void monitoring_prof_mpi_finalize_f2c( MPI_Fint *ierr ) {
int c_ierr;
c_ierr = MPI_Finalize();
if (NULL != ierr) *ierr = OMPI_INT_2_FINT(c_ierr);
}
#if OPAL_HAVE_WEAK_SYMBOLS
#pragma weak MPI_INIT = monitoring_prof_mpi_init_f2c
#pragma weak mpi_init = monitoring_prof_mpi_init_f2c
#pragma weak mpi_init_ = monitoring_prof_mpi_init_f2c
#pragma weak mpi_init__ = monitoring_prof_mpi_init_f2c
#pragma weak MPI_Init_f = monitoring_prof_mpi_init_f2c
#pragma weak MPI_Init_f08 = monitoring_prof_mpi_init_f2c
#pragma weak MPI_FINALIZE = monitoring_prof_mpi_finalize_f2c
#pragma weak mpi_finalize = monitoring_prof_mpi_finalize_f2c
#pragma weak mpi_finalize_ = monitoring_prof_mpi_finalize_f2c
#pragma weak mpi_finalize__ = monitoring_prof_mpi_finalize_f2c
#pragma weak MPI_Finalize_f = monitoring_prof_mpi_finalize_f2c
#pragma weak MPI_Finalize_f08 = monitoring_prof_mpi_finalize_f2c
#elif OMPI_BUILD_FORTRAN_BINDINGS
OMPI_GENERATE_F77_BINDINGS (MPI_INIT,
mpi_init,
mpi_init_,
mpi_init__,
monitoring_prof_mpi_init_f2c,
(MPI_Fint *ierr),
(ierr) )
OMPI_GENERATE_F77_BINDINGS (MPI_FINALIZE,
mpi_finalize,
mpi_finalize_,
mpi_finalize__,
monitoring_prof_mpi_finalize_f2c,
(MPI_Fint *ierr),
(ierr) )
#endif
Показать git blame Копировать постоянную ссылку