src/smpi/smpi_base.c

   1 #include "private.h"
   2
   3 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_base, smpi, "Logging specific to SMPI (base)");
   4
   5 smpi_mpi_global_t smpi_mpi_global = NULL;
   6
   7 void smpi_mpi_land_func(void *a, void *b, int *length, MPI_Datatype *datatype);
   8
   9 void smpi_mpi_land_func(void *a, void *b, int *length, MPI_Datatype *datatype)
  10 {
  11         int i;
  12         if (*datatype == smpi_mpi_global->mpi_int) {
  13                 int *x = a, *y = b;
  14                 for (i = 0; i < *length; i++) {
  15                         y[i] = x[i] && y[i];
  16                 }
  17         }
  18 }
  19
  20 void smpi_mpi_sum_func(void *a, void *b, int *length, MPI_Datatype *datatype);
  21
  22 void smpi_mpi_sum_func(void *a, void *b, int *length, MPI_Datatype *datatype)
  23 {
  24         int i;
  25         if (*datatype == smpi_mpi_global->mpi_int) {
  26                 int *x = a, *y = b;
  27                 for (i = 0; i < *length; i++) {
  28                         y[i] = x[i] + y[i];
  29                 }
  30         }
  31 }
  32
  33 int smpi_mpi_comm_rank(smpi_mpi_communicator_t comm)
  34 {
  35         return comm->index_to_rank_map[smpi_host_index()];
  36 }
  37
  38 void smpi_mpi_init()
  39 {
  40         smx_host_t host;
  41         smx_host_t *hosts;
  42         int host_count;
  43         int i;
  44         smpi_host_data_t hdata;
  45
  46    /* Connect our log channels: that must be done manually under windows */
  47    /* (should be done only once, not for each process) */
  48    XBT_LOG_CONNECT(smpi_base, smpi);
  49    XBT_LOG_CONNECT(smpi_bench, smpi);
  50    XBT_LOG_CONNECT(smpi_kernel, smpi);
  51    XBT_LOG_CONNECT(smpi_mpi, smpi);
  52    XBT_LOG_CONNECT(smpi_receiver, smpi);
  53    XBT_LOG_CONNECT(smpi_sender, smpi);
  54    XBT_LOG_CONNECT(smpi_util, smpi);
  55
  56         SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex);
  57         smpi_global->running_hosts_count++;
  58         SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex);
  59
  60         // initialize some local variables
  61         host       = SIMIX_host_self();
  62         hosts      = SIMIX_host_get_table();
  63         host_count = SIMIX_host_get_number();
  64
  65         hdata      = xbt_new(s_smpi_host_data_t, 1);
  66
  67         for (i = 0; i < host_count && host != hosts[i]; i ++);
  68
  69         hdata->index = i;
  70         hdata->mutex = SIMIX_mutex_init();
  71         hdata->cond  = SIMIX_cond_init();
  72
  73         SIMIX_host_set_data(host, hdata);
  74
  75         // node 0 sets the globals
  76         if (0 == i) {
  77
  78                 smpi_global->hosts                              = hosts;
  79                 smpi_global->host_count                         = host_count;
  80
  81                 smpi_mpi_global                                 = xbt_new(s_smpi_mpi_global_t, 1);
  82
  83                 // global communicator
  84                 smpi_mpi_global->mpi_comm_world                 = xbt_new(s_smpi_mpi_communicator_t, 1);
  85                 smpi_mpi_global->mpi_comm_world->size           = host_count;
  86                 smpi_mpi_global->mpi_comm_world->barrier_count  = 0;
  87                 smpi_mpi_global->mpi_comm_world->barrier_mutex  = SIMIX_mutex_init();
  88                 smpi_mpi_global->mpi_comm_world->barrier_cond   = SIMIX_cond_init();
  89                 smpi_mpi_global->mpi_comm_world->rank_to_index_map = xbt_new(int, host_count);
  90                 smpi_mpi_global->mpi_comm_world->index_to_rank_map = xbt_new(int, host_count);
  91                 for (i = 0; i < host_count; i++) {
  92                         smpi_mpi_global->mpi_comm_world->rank_to_index_map[i] = i;
  93                         smpi_mpi_global->mpi_comm_world->index_to_rank_map[i] = i;
  94                 }
  95
  96                 // mpi datatypes
  97                 smpi_mpi_global->mpi_byte                       = xbt_new(s_smpi_mpi_datatype_t, 1);
  98                 smpi_mpi_global->mpi_byte->size                 = (size_t)1;
  99                 smpi_mpi_global->mpi_int                        = xbt_new(s_smpi_mpi_datatype_t, 1);
 100                 smpi_mpi_global->mpi_int->size                  = sizeof(int);
 101                 smpi_mpi_global->mpi_double                     = xbt_new(s_smpi_mpi_datatype_t, 1);
 102                 smpi_mpi_global->mpi_double->size               = sizeof(double);
 103
 104                 // mpi operations
 105                 smpi_mpi_global->mpi_land                       = xbt_new(s_smpi_mpi_op_t, 1);
 106                 smpi_mpi_global->mpi_land->func                 = smpi_mpi_land_func;
 107                 smpi_mpi_global->mpi_sum                        = xbt_new(s_smpi_mpi_op_t, 1);
 108                 smpi_mpi_global->mpi_sum->func                  = smpi_mpi_sum_func;
 109
 110                 // signal all nodes to perform initialization
 111                 SIMIX_mutex_lock(smpi_global->start_stop_mutex);
 112                 smpi_global->root_ready = 1;
 113                 SIMIX_cond_broadcast(smpi_global->start_stop_cond);
 114                 SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
 115
 116         } else {
 117
 118                 // make sure root is done before own initialization
 119                 SIMIX_mutex_lock(smpi_global->start_stop_mutex);
 120                 while (!smpi_global->root_ready) {
 121                         SIMIX_cond_wait(smpi_global->start_stop_cond, smpi_global->start_stop_mutex);
 122                 }
 123                 SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
 124
 125         }
 126
 127         // wait for all nodes to signal initializatin complete
 128         SIMIX_mutex_lock(smpi_global->start_stop_mutex);
 129         smpi_global->ready_process_count++;
 130         if (smpi_global->ready_process_count >= 3 * host_count) {
 131                 SIMIX_cond_broadcast(smpi_global->start_stop_cond);
 132         }
 133         while (smpi_global->ready_process_count < 3 * host_count) {
 134                 SIMIX_cond_wait(smpi_global->start_stop_cond, smpi_global->start_stop_mutex);
 135         }
 136         SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
 137
 138         return;
 139 }
 140
 141 void smpi_mpi_finalize()
 142 {
 143         int i;
 144
 145         SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex);
 146         i = --smpi_global->running_hosts_count;
 147         SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex);
 148
 149         SIMIX_mutex_lock(smpi_global->start_stop_mutex);
 150         smpi_global->ready_process_count--;
 151         SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
 152
 153         SIMIX_mutex_destroy(smpi_host_mutex());
 154         SIMIX_cond_destroy(smpi_host_cond());
 155
 156         if (0 >= i) {
 157
 158                 // wake up senders/receivers
 159                 for (i = 0; i < smpi_global->host_count; i++) {
 160                         if (SIMIX_process_is_suspended(smpi_global->sender_processes[i])) {
 161                                 SIMIX_process_resume(smpi_global->sender_processes[i]);
 162                         }
 163                         if (SIMIX_process_is_suspended(smpi_global->receiver_processes[i])) {
 164                                 SIMIX_process_resume(smpi_global->receiver_processes[i]);
 165                         }
 166                 }
 167
 168                 SIMIX_mutex_destroy(smpi_mpi_global->mpi_comm_world->barrier_mutex);
 169                 SIMIX_cond_destroy(smpi_mpi_global->mpi_comm_world->barrier_cond);
 170                 xbt_free(smpi_mpi_global->mpi_comm_world);
 171
 172                 xbt_free(smpi_mpi_global->mpi_byte);
 173                 xbt_free(smpi_mpi_global->mpi_int);
 174                 xbt_free(smpi_mpi_global->mpi_double);
 175
 176                 xbt_free(smpi_mpi_global->mpi_land);
 177                 xbt_free(smpi_mpi_global->mpi_sum);
 178
 179                 xbt_free(smpi_mpi_global);
 180
 181         }
 182
 183 }
 184
 185 int smpi_mpi_barrier(smpi_mpi_communicator_t comm)
 186 {
 187
 188         SIMIX_mutex_lock(comm->barrier_mutex);
 189         ++comm->barrier_count;
 190         if (comm->barrier_count > comm->size) { // only happens on second barrier...
 191                 comm->barrier_count = 0;
 192         } else if (comm->barrier_count == comm->size) {
 193                 SIMIX_cond_broadcast(comm->barrier_cond);
 194         }
 195         while (comm->barrier_count < comm->size) {
 196                 SIMIX_cond_wait(comm->barrier_cond, comm->barrier_mutex);
 197         }
 198         SIMIX_mutex_unlock(comm->barrier_mutex);
 199
 200         return MPI_SUCCESS;
 201 }
 202
 203 int smpi_mpi_isend(smpi_mpi_request_t request)
 204 {
 205         int retval = MPI_SUCCESS;
 206         int index  = smpi_host_index();
 207
 208         if (NULL == request) {
 209                 retval = MPI_ERR_INTERN;
 210         } else {
 211                 SIMIX_mutex_lock(smpi_global->pending_send_request_queues_mutexes[index]);
 212                 xbt_fifo_push(smpi_global->pending_send_request_queues[index], request);
 213                 SIMIX_mutex_unlock(smpi_global->pending_send_request_queues_mutexes[index]);
 214
 215                 if (SIMIX_process_is_suspended(smpi_global->sender_processes[index])) {
 216                         SIMIX_process_resume(smpi_global->sender_processes[index]);
 217                 }
 218         }
 219
 220         return retval;
 221 }
 222
 223 int smpi_mpi_irecv(smpi_mpi_request_t request)
 224 {
 225         int retval = MPI_SUCCESS;
 226         int index = smpi_host_index();
 227
 228         if (NULL == request) {
 229                 retval = MPI_ERR_INTERN;
 230         } else {
 231                 SIMIX_mutex_lock(smpi_global->pending_recv_request_queues_mutexes[index]);
 232                 xbt_fifo_push(smpi_global->pending_recv_request_queues[index], request);
 233                 SIMIX_mutex_unlock(smpi_global->pending_recv_request_queues_mutexes[index]);
 234
 235                 if (SIMIX_process_is_suspended(smpi_global->receiver_processes[index])) {
 236                         SIMIX_process_resume(smpi_global->receiver_processes[index]);
 237                 }
 238         }
 239
 240         return retval;
 241 }
 242
 243 int smpi_mpi_wait(smpi_mpi_request_t request, smpi_mpi_status_t *status)
 244 {
 245         int retval = MPI_SUCCESS;
 246
 247         if (NULL == request) {
 248                 retval = MPI_ERR_INTERN;
 249         } else {
 250                 SIMIX_mutex_lock(request->mutex);
 251                 while (!request->completed) {
 252                         SIMIX_cond_wait(request->cond, request->mutex);
 253                 }
 254                 if (NULL != status) {
 255                         status->MPI_SOURCE = request->src;
 256                         status->MPI_TAG    = request->tag;
 257                         status->MPI_ERROR  = MPI_SUCCESS;
 258                 }
 259                 SIMIX_mutex_unlock(request->mutex);
 260         }
 261
 262         return retval;
 263 }