src/smpi/src/smpi_base.c

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