src/smpi/src/smpi_base.c

   1 #include <stdio.h>
   2 #include <signal.h>
   3 #include <sys/time.h>
   4
   5 #include "../include/private.h"
   6
   7 SMPI_Global_t     smpi_global     = NULL;
   8
   9 SMPI_MPI_Global_t smpi_mpi_global = NULL;
  10
  11 XBT_LOG_NEW_DEFAULT_CATEGORY(smpi, "SMPI");
  12
  13 int inline smpi_mpi_comm_size(smpi_mpi_communicator_t *comm)
  14 {
  15         return comm->size;
  16 }
  17
  18 // FIXME: smarter algorithm?
  19 int smpi_mpi_comm_rank(smpi_mpi_communicator_t *comm, smx_host_t host)
  20 {
  21         int i;
  22
  23         for(i = comm->size - 1; i > 0 && host != comm->hosts[i]; i--);
  24
  25         return i;
  26 }
  27
  28 int inline smpi_mpi_comm_rank_self(smpi_mpi_communicator_t *comm)
  29 {
  30         return smpi_mpi_comm_rank(comm, SIMIX_host_self());
  31 }
  32
  33 int inline smpi_mpi_comm_world_rank_self()
  34 {
  35         return smpi_mpi_comm_rank(smpi_mpi_global->mpi_comm_world, SIMIX_host_self());
  36 }
  37
  38 int smpi_sender(int argc, char **argv)
  39 {
  40         smx_process_t self;
  41         smx_host_t shost;
  42         int rank;
  43
  44         xbt_fifo_t request_queue;
  45         smx_mutex_t request_queue_mutex;
  46         int size;
  47
  48         int running_hosts_count;
  49
  50         smpi_mpi_request_t *request;
  51
  52         smx_host_t dhost;
  53
  54         smx_action_t communicate_action;
  55
  56         smpi_received_message_t *message;
  57
  58         int drank;
  59
  60         smx_process_t receiver_process;
  61
  62         self  = SIMIX_process_self();
  63         shost = SIMIX_host_self();
  64         rank  = smpi_mpi_comm_rank(smpi_mpi_global->mpi_comm_world, shost);
  65
  66         // make sure root is done before own initialization
  67         SIMIX_mutex_lock(smpi_global->start_stop_mutex);
  68         if (!smpi_global->root_ready) {
  69                 SIMIX_cond_wait(smpi_global->start_stop_cond, smpi_global->start_stop_mutex);
  70         }
  71         SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
  72
  73         request_queue       = smpi_global->pending_send_request_queues[rank];
  74         request_queue_mutex = smpi_global->pending_send_request_queues_mutexes[rank];
  75         size                = smpi_mpi_comm_size(smpi_mpi_global->mpi_comm_world);
  76
  77         smpi_global->sender_processes[rank] = self;
  78
  79         // wait for all nodes to signal initializatin complete
  80         SIMIX_mutex_lock(smpi_global->start_stop_mutex);
  81         smpi_global->ready_process_count++;
  82         if (smpi_global->ready_process_count < 3 * size) {
  83                 SIMIX_cond_wait(smpi_global->start_stop_cond, smpi_global->start_stop_mutex);
  84         } else {
  85                 SIMIX_cond_broadcast(smpi_global->start_stop_cond);
  86         }
  87         SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
  88
  89         do {
  90
  91                 SIMIX_mutex_lock(request_queue_mutex);
  92                 request = xbt_fifo_shift(request_queue);
  93                 SIMIX_mutex_unlock(request_queue_mutex);
  94
  95                 if (NULL == request) {
  96                         SIMIX_process_suspend(self);
  97                 } else {
  98
  99                         SIMIX_mutex_lock(request->mutex);
 100
 101                         // copy request to appropriate received queue
 102                         message       = xbt_mallocator_get(smpi_global->message_mallocator);
 103                         message->comm = request->comm;
 104                         message->src  = request->src;
 105                         message->dst  = request->dst;
 106                         message->tag  = request->tag;
 107                         message->buf  = xbt_malloc(request->datatype->size * request->count);
 108                         memcpy(message->buf, request->buf, request->datatype->size * request->count);
 109
 110                         dhost = request->comm->hosts[request->dst];
 111                         drank = smpi_mpi_comm_rank(smpi_mpi_global->mpi_comm_world, dhost);
 112
 113                         SIMIX_mutex_lock(smpi_global->received_message_queues_mutexes[drank]);
 114                         xbt_fifo_push(smpi_global->received_message_queues[drank], message);
 115                         SIMIX_mutex_unlock(smpi_global->received_message_queues_mutexes[drank]);
 116
 117                         request->completed = 1;
 118
 119                         communicate_action = SIMIX_action_communicate(shost, dhost,
 120                                 "communication", request->datatype->size * request->count * 1.0, -1.0);
 121
 122                         SIMIX_register_condition_to_action(communicate_action, request->cond);
 123                         SIMIX_register_action_to_condition(communicate_action, request->cond);
 124
 125                         SIMIX_cond_wait(request->cond, request->mutex);
 126
 127                         SIMIX_mutex_unlock(request->mutex);
 128
 129                         // wake up receiver if necessary
 130                         receiver_process = smpi_global->receiver_processes[drank];
 131
 132                         if (SIMIX_process_is_suspended(receiver_process)) {
 133                                 SIMIX_process_resume(receiver_process);
 134                         }
 135
 136                 }
 137
 138                 SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex);
 139                 running_hosts_count = smpi_global->running_hosts_count;
 140                 SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex);
 141
 142         } while (0 < running_hosts_count);
 143
 144         SIMIX_mutex_lock(smpi_global->start_stop_mutex);
 145         smpi_global->ready_process_count--;
 146         if (smpi_global->ready_process_count == 0) {
 147                 SIMIX_cond_broadcast(smpi_global->start_stop_cond);
 148         } else if (smpi_global->ready_process_count < 0) {
 149                 // FIXME: can't happen! abort!
 150         }
 151         SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
 152
 153         return 0;
 154 }
 155
 156 int smpi_receiver(int argc, char **argv)
 157 {
 158         smx_process_t self;
 159         int rank;
 160
 161         xbt_fifo_t request_queue;
 162         smx_mutex_t request_queue_mutex;
 163         xbt_fifo_t message_queue;
 164         smx_mutex_t message_queue_mutex;
 165         int size;
 166
 167         int running_hosts_count;
 168
 169         smpi_mpi_request_t *request;
 170         smpi_received_message_t *message;
 171
 172         xbt_fifo_item_t request_item;
 173         xbt_fifo_item_t message_item;
 174
 175         self  = SIMIX_process_self();
 176         rank  = smpi_mpi_comm_world_rank_self();
 177
 178         // make sure root is done before own initialization
 179         SIMIX_mutex_lock(smpi_global->start_stop_mutex);
 180         if (!smpi_global->root_ready) {
 181                 SIMIX_cond_wait(smpi_global->start_stop_cond, smpi_global->start_stop_mutex);
 182         }
 183         SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
 184
 185         request_queue       = smpi_global->pending_recv_request_queues[rank];
 186         request_queue_mutex = smpi_global->pending_recv_request_queues_mutexes[rank];
 187         message_queue       = smpi_global->received_message_queues[rank];
 188         message_queue_mutex = smpi_global->received_message_queues_mutexes[rank];
 189         size                = smpi_mpi_comm_size(smpi_mpi_global->mpi_comm_world);
 190
 191         smpi_global->receiver_processes[rank] = self;
 192
 193         // wait for all nodes to signal initializatin complete
 194         SIMIX_mutex_lock(smpi_global->start_stop_mutex);
 195         smpi_global->ready_process_count++;
 196         if (smpi_global->ready_process_count < 3 * size) {
 197                 SIMIX_cond_wait(smpi_global->start_stop_cond, smpi_global->start_stop_mutex);
 198         } else {
 199                 SIMIX_cond_broadcast(smpi_global->start_stop_cond);
 200         }
 201         SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
 202
 203         do {
 204                 request = NULL;
 205                 message = NULL;
 206
 207                 // FIXME: better algorithm, maybe some kind of balanced tree? or a heap?
 208
 209                 // FIXME: not the best way to request multiple locks...
 210                 SIMIX_mutex_lock(request_queue_mutex);
 211                 SIMIX_mutex_lock(message_queue_mutex);
 212                 for (request_item = xbt_fifo_get_first_item(request_queue);
 213                         NULL != request_item;
 214                         request_item = xbt_fifo_get_next_item(request_item)) {
 215                         request = xbt_fifo_get_item_content(request_item);
 216                         for (message_item = xbt_fifo_get_first_item(message_queue);
 217                                 NULL != message_item;
 218                                 message_item = xbt_fifo_get_next_item(message_item)) {
 219                                 message = xbt_fifo_get_item_content(message_item);
 220                                 if (request->comm == message->comm &&
 221                                                 (MPI_ANY_SOURCE == request->src || request->src == message->src) &&
 222                                                 request->tag == message->tag) {
 223                                         xbt_fifo_remove_item(request_queue, request_item);
 224                                         xbt_fifo_remove_item(message_queue, message_item);
 225                                         goto stopsearch;
 226                                 }
 227                         }
 228                 }
 229 stopsearch:
 230                 SIMIX_mutex_unlock(message_queue_mutex);
 231                 SIMIX_mutex_unlock(request_queue_mutex);
 232
 233                 if (NULL == request || NULL == message) {
 234                         SIMIX_process_suspend(self);
 235                 } else {
 236                         SIMIX_mutex_lock(request->mutex);
 237
 238                         memcpy(request->buf, message->buf, request->datatype->size * request->count);
 239                         request->src = message->src;
 240                         request->completed = 1;
 241                         SIMIX_cond_broadcast(request->cond);
 242
 243                         SIMIX_mutex_unlock(request->mutex);
 244
 245                         xbt_free(message->buf);
 246                         xbt_mallocator_release(smpi_global->message_mallocator, message);
 247                 }
 248
 249                 SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex);
 250                 running_hosts_count = smpi_global->running_hosts_count;
 251                 SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex);
 252
 253         } while (0 < running_hosts_count);
 254
 255         SIMIX_mutex_lock(smpi_global->start_stop_mutex);
 256         smpi_global->ready_process_count--;
 257         if (smpi_global->ready_process_count == 0) {
 258                 SIMIX_cond_broadcast(smpi_global->start_stop_cond);
 259         } else if (smpi_global->ready_process_count < 0) {
 260                 // FIXME: can't happen, abort!
 261         }
 262         SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
 263
 264         return 0;
 265 }
 266
 267 void *smpi_request_new()
 268 {
 269         smpi_mpi_request_t *request = xbt_new(smpi_mpi_request_t, 1);
 270
 271         request->completed = 0;
 272         request->mutex     = SIMIX_mutex_init();
 273         request->cond      = SIMIX_cond_init();
 274
 275         return request;
 276 }
 277
 278 void smpi_request_free(void *pointer) {
 279
 280         smpi_mpi_request_t *request = pointer;
 281
 282         if (NULL != request) {
 283                 SIMIX_cond_destroy(request->cond);
 284                 SIMIX_mutex_destroy(request->mutex);
 285                 xbt_free(request);
 286         }
 287
 288         return;
 289 }
 290
 291 void smpi_request_reset(void *pointer) {
 292         return;
 293 }
 294
 295
 296 void *smpi_message_new()
 297 {
 298         return xbt_new(smpi_received_message_t, 1);
 299 }
 300
 301 void smpi_message_free(void *pointer)
 302 {
 303         if (NULL != pointer) {
 304                 xbt_free(pointer);
 305         }
 306
 307         return;
 308 }
 309
 310 void smpi_message_reset(void *pointer)
 311 {
 312         return;
 313 }
 314
 315 void smpi_global_init()
 316 {
 317         int i;
 318
 319         int size = SIMIX_host_get_number();
 320
 321         smpi_global = xbt_new(s_SMPI_Global_t, 1);
 322
 323         // config variable
 324         smpi_global->reference_speed                     = SMPI_DEFAULT_SPEED;
 325
 326         smpi_global->root_ready                          = 0;
 327         smpi_global->ready_process_count                 = 0;
 328
 329         // start/stop
 330         smpi_global->start_stop_mutex                    = SIMIX_mutex_init();
 331         smpi_global->start_stop_cond                     = SIMIX_cond_init();
 332
 333         // processes
 334         smpi_global->sender_processes                    = xbt_new(smx_process_t, size);
 335         smpi_global->receiver_processes                  = xbt_new(smx_process_t, size);
 336
 337         // running hosts
 338         smpi_global->running_hosts_count_mutex           = SIMIX_mutex_init();
 339         smpi_global->running_hosts_count                 = 0;
 340
 341         // mallocators
 342         smpi_global->request_mallocator                  = xbt_mallocator_new(SMPI_REQUEST_MALLOCATOR_SIZE,
 343                                                              smpi_request_new, smpi_request_free, smpi_request_reset);
 344         smpi_global->message_mallocator                  = xbt_mallocator_new(SMPI_MESSAGE_MALLOCATOR_SIZE,
 345                                                              smpi_message_new, smpi_message_free, smpi_message_reset);
 346
 347         //
 348         smpi_global->pending_send_request_queues         = xbt_new(xbt_fifo_t,  size);
 349         smpi_global->pending_send_request_queues_mutexes = xbt_new(smx_mutex_t, size);
 350         smpi_global->pending_recv_request_queues         = xbt_new(xbt_fifo_t,  size);
 351         smpi_global->pending_recv_request_queues_mutexes = xbt_new(smx_mutex_t, size);
 352         smpi_global->received_message_queues             = xbt_new(xbt_fifo_t,  size);
 353         smpi_global->received_message_queues_mutexes     = xbt_new(smx_mutex_t, size);
 354         smpi_global->timers                              = xbt_new(xbt_os_timer_t, size);
 355         smpi_global->timers_mutexes                      = xbt_new(smx_mutex_t, size);
 356
 357         for(i = 0; i < size; i++) {
 358                 smpi_global->pending_send_request_queues[i]         = xbt_fifo_new();
 359                 smpi_global->pending_send_request_queues_mutexes[i] = SIMIX_mutex_init();
 360                 smpi_global->pending_recv_request_queues[i]         = xbt_fifo_new();
 361                 smpi_global->pending_recv_request_queues_mutexes[i] = SIMIX_mutex_init();
 362                 smpi_global->received_message_queues[i]             = xbt_fifo_new();
 363                 smpi_global->received_message_queues_mutexes[i]     = SIMIX_mutex_init();
 364                 smpi_global->timers[i]                              = xbt_os_timer_new();
 365                 smpi_global->timers_mutexes[i]                      = SIMIX_mutex_init();
 366         }
 367
 368 }
 369
 370 void smpi_global_destroy()
 371 {
 372         int i;
 373
 374         int size = SIMIX_host_get_number();
 375
 376         // start/stop
 377         SIMIX_mutex_destroy(smpi_global->start_stop_mutex);
 378         SIMIX_cond_destroy(smpi_global->start_stop_cond);
 379
 380         // processes
 381         xbt_free(smpi_global->sender_processes);
 382         xbt_free(smpi_global->receiver_processes);
 383
 384         // running hosts
 385         SIMIX_mutex_destroy(smpi_global->running_hosts_count_mutex);
 386
 387         // mallocators
 388         xbt_mallocator_free(smpi_global->request_mallocator);
 389         xbt_mallocator_free(smpi_global->message_mallocator);
 390
 391         for(i = 0; i < size; i++) {
 392                 xbt_fifo_free(smpi_global->pending_send_request_queues[i]);
 393                 SIMIX_mutex_destroy(smpi_global->pending_send_request_queues_mutexes[i]);
 394                 xbt_fifo_free(smpi_global->pending_recv_request_queues[i]);
 395                 SIMIX_mutex_destroy(smpi_global->pending_recv_request_queues_mutexes[i]);
 396                 xbt_fifo_free(smpi_global->received_message_queues[i]);
 397                 SIMIX_mutex_destroy(smpi_global->received_message_queues_mutexes[i]);
 398                 xbt_os_timer_free(smpi_global->timers[i]);
 399                 SIMIX_mutex_destroy(smpi_global->timers_mutexes[i]);
 400         }
 401
 402         xbt_free(smpi_global->pending_send_request_queues);
 403         xbt_free(smpi_global->pending_send_request_queues_mutexes);
 404         xbt_free(smpi_global->pending_recv_request_queues);
 405         xbt_free(smpi_global->pending_recv_request_queues_mutexes);
 406         xbt_free(smpi_global->received_message_queues);
 407         xbt_free(smpi_global->received_message_queues_mutexes);
 408         xbt_free(smpi_global->timers);
 409         xbt_free(smpi_global->timers_mutexes);
 410
 411         xbt_free(smpi_global);
 412 }
 413
 414 int smpi_run_simulation(int argc, char **argv)
 415 {
 416         smx_cond_t   cond           = NULL;
 417         smx_action_t action         = NULL;
 418
 419         xbt_fifo_t   actions_failed = xbt_fifo_new();
 420         xbt_fifo_t   actions_done   = xbt_fifo_new();
 421
 422         srand(SMPI_RAND_SEED);
 423
 424         SIMIX_global_init(&argc, argv);
 425
 426         SIMIX_function_register("smpi_simulated_main", smpi_simulated_main);
 427         SIMIX_function_register("smpi_sender",         smpi_sender);
 428         SIMIX_function_register("smpi_receiver",       smpi_receiver);
 429
 430         // FIXME: ought to verify these files...
 431         SIMIX_create_environment(argv[1]);
 432
 433         // must initialize globals between creating environment and launching app....
 434         smpi_global_init();
 435
 436         SIMIX_launch_application(argv[2]);
 437
 438         /* Prepare to display some more info when dying on Ctrl-C pressing */
 439         // FIXME: doesn't work
 440         //signal(SIGINT, inthandler);
 441
 442         /* Clean IO before the run */
 443         fflush(stdout);
 444         fflush(stderr);
 445
 446         while (SIMIX_solve(actions_done, actions_failed) != -1.0) {
 447                 while (action = xbt_fifo_pop(actions_failed)) {
 448                         DEBUG1("** %s failed **", action->name);
 449                         while (cond = xbt_fifo_pop(action->cond_list)) {
 450                                 SIMIX_cond_broadcast(cond);
 451                         }
 452                         SIMIX_action_destroy(action);
 453                 }
 454                 while (action = xbt_fifo_pop(actions_done)) {
 455                         DEBUG1("** %s done **",action->name);
 456                         while (cond = xbt_fifo_pop(action->cond_list)) {
 457                                 SIMIX_cond_broadcast(cond);
 458                         }
 459                         SIMIX_action_destroy(action);
 460                 }
 461         }
 462
 463         xbt_fifo_free(actions_failed);
 464         xbt_fifo_free(actions_done);
 465
 466         INFO1("simulation time %g", SIMIX_get_clock());
 467
 468         smpi_global_destroy();
 469
 470         SIMIX_clean();
 471
 472         return 0;
 473 }
 474
 475 void smpi_mpi_land_func(void *x, void *y, void *z)
 476 {
 477         *(int *)z = *(int *)x && *(int *)y;
 478 }
 479
 480 void smpi_mpi_sum_func(void *x, void *y, void *z)
 481 {
 482         *(int *)z = *(int *)x + *(int *)y;
 483 }
 484
 485
 486 void smpi_mpi_init()
 487 {
 488         smx_process_t process;
 489         smx_host_t host;
 490         smx_host_t *hosts;
 491         int size;
 492
 493         SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex);
 494         smpi_global->running_hosts_count++;
 495         SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex);
 496
 497         // initialize some local variables
 498         process = SIMIX_process_self();
 499         host    = SIMIX_host_self();
 500         hosts   = SIMIX_host_get_table();
 501         size    = SIMIX_host_get_number();
 502
 503         // node 0 sets the globals
 504         if (host == hosts[0]) {
 505
 506                 smpi_mpi_global                                = xbt_new(s_SMPI_MPI_Global_t, 1);
 507
 508                 // global communicator
 509                 smpi_mpi_global->mpi_comm_world                = xbt_new(smpi_mpi_communicator_t, 1);
 510                 smpi_mpi_global->mpi_comm_world->size          = size;
 511                 smpi_mpi_global->mpi_comm_world->barrier_count = 0;
 512                 smpi_mpi_global->mpi_comm_world->barrier_mutex = SIMIX_mutex_init();
 513                 smpi_mpi_global->mpi_comm_world->barrier_cond  = SIMIX_cond_init();
 514                 smpi_mpi_global->mpi_comm_world->hosts         = hosts;
 515                 smpi_mpi_global->mpi_comm_world->processes     = xbt_new(smx_process_t, size);
 516                 smpi_mpi_global->mpi_comm_world->processes[0]  = process;
 517
 518                 // mpi datatypes
 519                 smpi_mpi_global->mpi_byte                      = xbt_new(smpi_mpi_datatype_t, 1);
 520                 smpi_mpi_global->mpi_byte->size                = (size_t)1;
 521                 smpi_mpi_global->mpi_int                       = xbt_new(smpi_mpi_datatype_t, 1);
 522                 smpi_mpi_global->mpi_int->size                 = sizeof(int);
 523                 smpi_mpi_global->mpi_double                    = xbt_new(smpi_mpi_datatype_t, 1);
 524                 smpi_mpi_global->mpi_double->size              = sizeof(double);
 525
 526                 // mpi operations
 527                 smpi_mpi_global->mpi_land                      = xbt_new(smpi_mpi_op_t, 1);
 528                 smpi_mpi_global->mpi_land->func                = smpi_mpi_land_func;
 529                 smpi_mpi_global->mpi_sum                       = xbt_new(smpi_mpi_op_t, 1);
 530                 smpi_mpi_global->mpi_sum->func                 = smpi_mpi_sum_func;
 531
 532                 // signal all nodes to perform initialization
 533                 SIMIX_mutex_lock(smpi_global->start_stop_mutex);
 534                 smpi_global->root_ready = 1;
 535                 SIMIX_cond_broadcast(smpi_global->start_stop_cond);
 536                 SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
 537
 538         } else {
 539
 540                 // make sure root is done before own initialization
 541                 SIMIX_mutex_lock(smpi_global->start_stop_mutex);
 542                 if (!smpi_global->root_ready) {
 543                         SIMIX_cond_wait(smpi_global->start_stop_cond, smpi_global->start_stop_mutex);
 544                 }
 545                 SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
 546
 547                 smpi_mpi_global->mpi_comm_world->processes[smpi_mpi_comm_world_rank_self()] = process;
 548         }
 549
 550         // wait for all nodes to signal initializatin complete
 551         SIMIX_mutex_lock(smpi_global->start_stop_mutex);
 552         smpi_global->ready_process_count++;
 553         if (smpi_global->ready_process_count < 3 * size) {
 554                 SIMIX_cond_wait(smpi_global->start_stop_cond, smpi_global->start_stop_mutex);
 555         } else {
 556                 SIMIX_cond_broadcast(smpi_global->start_stop_cond);
 557         }
 558         SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
 559
 560         return;
 561 }
 562
 563 void smpi_mpi_finalize()
 564 {
 565         int i;
 566
 567         SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex);
 568         i = --smpi_global->running_hosts_count;
 569         SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex);
 570
 571         SIMIX_mutex_lock(smpi_global->start_stop_mutex);
 572         smpi_global->ready_process_count--;
 573         SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
 574
 575         if (0 >= i) {
 576
 577                 // wake up senders/receivers
 578                 for (i = 0; i < smpi_mpi_global->mpi_comm_world->size; i++) {
 579                         if (SIMIX_process_is_suspended(smpi_global->sender_processes[i])) {
 580                                 SIMIX_process_resume(smpi_global->sender_processes[i]);
 581                         }
 582                         if (SIMIX_process_is_suspended(smpi_global->receiver_processes[i])) {
 583                                 SIMIX_process_resume(smpi_global->receiver_processes[i]);
 584                         }
 585                 }
 586
 587                 // wait for senders/receivers to exit...
 588                 SIMIX_mutex_lock(smpi_global->start_stop_mutex);
 589                 if (smpi_global->ready_process_count > 0) {
 590                         SIMIX_cond_wait(smpi_global->start_stop_cond, smpi_global->start_stop_mutex);
 591                 }
 592                 SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
 593
 594                 SIMIX_mutex_destroy(smpi_mpi_global->mpi_comm_world->barrier_mutex);
 595                 SIMIX_cond_destroy(smpi_mpi_global->mpi_comm_world->barrier_cond);
 596                 xbt_free(smpi_mpi_global->mpi_comm_world->processes);
 597                 xbt_free(smpi_mpi_global->mpi_comm_world);
 598
 599                 xbt_free(smpi_mpi_global->mpi_byte);
 600                 xbt_free(smpi_mpi_global->mpi_int);
 601                 xbt_free(smpi_mpi_global->mpi_double);
 602
 603                 xbt_free(smpi_mpi_global->mpi_land);
 604                 xbt_free(smpi_mpi_global->mpi_sum);
 605
 606                 xbt_free(smpi_mpi_global);
 607         }
 608
 609 }
 610
 611 // FIXME: could cause trouble with multithreaded procs on same host...
 612 void smpi_bench_begin()
 613 {
 614         int rank = smpi_mpi_comm_world_rank_self();
 615         SIMIX_mutex_lock(smpi_global->timers_mutexes[rank]);
 616         xbt_os_timer_start(smpi_global->timers[rank]);
 617         return;
 618 }
 619
 620 void smpi_bench_end()
 621 {
 622         int rank = smpi_mpi_comm_world_rank_self();
 623         double duration;
 624         smx_host_t host;
 625         smx_action_t compute_action;
 626         smx_mutex_t mutex;
 627         smx_cond_t cond;
 628
 629         xbt_os_timer_stop(smpi_global->timers[rank]);
 630
 631         duration       = xbt_os_timer_elapsed(smpi_global->timers[rank]);
 632         SIMIX_mutex_unlock(smpi_global->timers_mutexes[rank]);
 633
 634         host           = smpi_mpi_global->mpi_comm_world->hosts[rank];
 635         compute_action = SIMIX_action_execute(host, "computation", duration * SMPI_DEFAULT_SPEED);
 636         mutex          = SIMIX_mutex_init();
 637         cond           = SIMIX_cond_init();
 638
 639         SIMIX_register_condition_to_action(compute_action, cond);
 640         SIMIX_register_action_to_condition(compute_action, cond);
 641         SIMIX_mutex_lock(mutex);
 642         SIMIX_cond_wait(cond, mutex);
 643         SIMIX_mutex_unlock(mutex);
 644
 645         SIMIX_mutex_destroy(mutex);
 646         SIMIX_cond_destroy(cond);
 647
 648         // FIXME: check for success/failure?
 649
 650         return;
 651 }
 652
 653 void smpi_barrier(smpi_mpi_communicator_t *comm) {
 654
 655         SIMIX_mutex_lock(comm->barrier_mutex);
 656         if(++comm->barrier_count < comm->size) {
 657                 SIMIX_cond_wait(comm->barrier_cond, comm->barrier_mutex);
 658         } else {
 659                 comm->barrier_count = 0;
 660                 SIMIX_cond_broadcast(comm->barrier_cond);
 661         }
 662         SIMIX_mutex_unlock(comm->barrier_mutex);
 663
 664         return;
 665 }
 666
 667 // FIXME: smarter algorithm...
 668 int smpi_comm_rank(smpi_mpi_communicator_t *comm, smx_host_t host)
 669 {
 670         int i;
 671         for(i = 0; i < comm->size && host != comm->hosts[i]; i++);
 672         if (i >= comm->size) i = -1;
 673         return i;
 674 }
 675
 676 int smpi_create_request(void *buf, int count, smpi_mpi_datatype_t *datatype,
 677         int src, int dst, int tag, smpi_mpi_communicator_t *comm, smpi_mpi_request_t **request)
 678 {
 679         int retval = MPI_SUCCESS;
 680
 681         *request = NULL;
 682
 683         if (0 > count) {
 684                 retval = MPI_ERR_COUNT;
 685         } else if (NULL == buf) {
 686                 retval = MPI_ERR_INTERN;
 687         } else if (NULL == datatype) {
 688                 retval = MPI_ERR_TYPE;
 689         } else if (NULL == comm) {
 690                 retval = MPI_ERR_COMM;
 691         } else if (MPI_ANY_SOURCE != src && (0 > src || comm->size <= src)) {
 692                 retval = MPI_ERR_RANK;
 693         } else if (0 > dst || comm->size <= dst) {
 694                 retval = MPI_ERR_RANK;
 695         } else if (0 > tag) {
 696                 retval = MPI_ERR_TAG;
 697         } else {
 698                 *request = xbt_mallocator_get(smpi_global->request_mallocator);
 699                 (*request)->comm       = comm;
 700                 (*request)->src        = src;
 701                 (*request)->dst        = dst;
 702                 (*request)->tag        = tag;
 703                 (*request)->buf        = buf;
 704                 (*request)->count      = count;
 705                 (*request)->datatype   = datatype;
 706         }
 707         return retval;
 708 }
 709
 710 int smpi_isend(smpi_mpi_request_t *request)
 711 {
 712         int retval = MPI_SUCCESS;
 713         int rank   = smpi_mpi_comm_world_rank_self();
 714
 715         if (NULL != request) {
 716                 SIMIX_mutex_lock(smpi_global->pending_send_request_queues_mutexes[rank]);
 717                 xbt_fifo_push(smpi_global->pending_send_request_queues[rank], request);
 718                 SIMIX_mutex_unlock(smpi_global->pending_send_request_queues_mutexes[rank]);
 719         }
 720
 721         if (SIMIX_process_is_suspended(smpi_global->sender_processes[rank])) {
 722                 SIMIX_process_resume(smpi_global->sender_processes[rank]);
 723         }
 724
 725         return retval;
 726 }
 727
 728 int smpi_irecv(smpi_mpi_request_t *request)
 729 {
 730         int retval = MPI_SUCCESS;
 731         int rank = smpi_mpi_comm_world_rank_self();
 732
 733         if (NULL != request) {
 734                 SIMIX_mutex_lock(smpi_global->pending_recv_request_queues_mutexes[rank]);
 735                 xbt_fifo_push(smpi_global->pending_recv_request_queues[rank], request);
 736                 SIMIX_mutex_unlock(smpi_global->pending_recv_request_queues_mutexes[rank]);
 737         }
 738
 739         if (SIMIX_process_is_suspended(smpi_global->receiver_processes[rank])) {
 740                 SIMIX_process_resume(smpi_global->receiver_processes[rank]);
 741         }
 742
 743         return retval;
 744 }
 745
 746 void smpi_wait(smpi_mpi_request_t *request, smpi_mpi_status_t *status)
 747 {
 748         if (NULL != request) {
 749                 SIMIX_mutex_lock(request->mutex);
 750                 if (!request->completed) {
 751                         SIMIX_cond_wait(request->cond, request->mutex);
 752                 }
 753                 if (NULL != status) {
 754                         status->MPI_SOURCE = request->src;
 755                 }
 756                 SIMIX_mutex_unlock(request->mutex);
 757         }
 758 }
 759
 760 // FIXME: move into own file
 761 int smpi_gettimeofday(struct timeval *tv, struct timezone *tz)
 762 {
 763         double now;
 764         int retval = 0;
 765         smpi_bench_end();
 766         if (NULL == tv) {
 767                 retval = -1;
 768         } else {
 769                 now = SIMIX_get_clock();
 770                 tv->tv_sec  = now;
 771                 tv->tv_usec = ((now - (double)tv->tv_sec) * 1000000.0);
 772         }
 773         smpi_bench_begin();
 774         return retval;
 775 }
 776
 777 unsigned int smpi_sleep(unsigned int seconds)
 778 {
 779         smx_mutex_t mutex;
 780         smx_cond_t cond;
 781         smx_host_t host;
 782         smx_action_t sleep_action;
 783
 784         smpi_bench_end();
 785         host         = SIMIX_host_self();
 786         sleep_action = SIMIX_action_sleep(host, seconds);
 787         mutex        = SIMIX_mutex_init();
 788         cond         = SIMIX_cond_init();
 789
 790         SIMIX_register_condition_to_action(sleep_action, cond);
 791         SIMIX_register_action_to_condition(sleep_action, cond);
 792         SIMIX_mutex_lock(mutex);
 793         SIMIX_cond_wait(cond, mutex);
 794         SIMIX_mutex_unlock(mutex);
 795
 796         SIMIX_mutex_destroy(mutex);
 797         SIMIX_cond_destroy(cond);
 798
 799         // FIXME: check for success/failure?
 800
 801         smpi_bench_begin();
 802         return 0;
 803 }
 804
 805 void smpi_exit(int status)
 806 {
 807         smpi_bench_end();
 808         SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex);
 809         smpi_global->running_hosts_count--;
 810         SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex);
 811         SIMIX_process_kill(SIMIX_process_self());
 812         return;
 813 }