7 SMPI_Global_t smpi_global = NULL;
9 SMPI_MPI_Global_t smpi_mpi_global = NULL;
11 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi,XBT_LOG_ROOT_CAT, "All SMPI categories (see \ref SMPI_API)");
13 int inline smpi_mpi_comm_size(smpi_mpi_communicator_t *comm)
18 // FIXME: smarter algorithm?
19 int smpi_mpi_comm_rank(smpi_mpi_communicator_t *comm, smx_host_t host)
23 for(i = comm->size - 1; i > 0 && host != comm->hosts[i]; i--);
28 int inline smpi_mpi_comm_rank_self(smpi_mpi_communicator_t *comm)
30 return smpi_mpi_comm_rank(comm, SIMIX_host_self());
33 int smpi_sender(int argc, char **argv)
39 xbt_fifo_t request_queue;
40 smx_mutex_t request_queue_mutex;
43 int running_hosts_count;
45 smpi_mpi_request_t *request;
51 smpi_received_message_t *message;
55 smx_process_t receiver_process;
57 self = SIMIX_process_self();
58 shost = SIMIX_host_self();
59 rank = smpi_mpi_comm_rank(smpi_mpi_global->mpi_comm_world, shost);
61 // make sure root is done before own initialization
62 SIMIX_mutex_lock(smpi_global->start_stop_mutex);
63 if (!smpi_global->root_ready) {
64 SIMIX_cond_wait(smpi_global->start_stop_cond, smpi_global->start_stop_mutex);
66 SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
68 request_queue = smpi_global->pending_send_request_queues[rank];
69 request_queue_mutex = smpi_global->pending_send_request_queues_mutexes[rank];
70 size = smpi_mpi_comm_size(smpi_mpi_global->mpi_comm_world);
72 smpi_global->sender_processes[rank] = self;
74 // wait for all nodes to signal initializatin complete
75 SIMIX_mutex_lock(smpi_global->start_stop_mutex);
76 smpi_global->ready_process_count++;
77 if (smpi_global->ready_process_count < 3 * size) {
78 SIMIX_cond_wait(smpi_global->start_stop_cond, smpi_global->start_stop_mutex);
80 SIMIX_cond_broadcast(smpi_global->start_stop_cond);
82 SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
86 SIMIX_mutex_lock(request_queue_mutex);
87 request = xbt_fifo_shift(request_queue);
88 SIMIX_mutex_unlock(request_queue_mutex);
90 if (NULL == request) {
91 SIMIX_process_suspend(self);
94 message = xbt_mallocator_get(smpi_global->message_mallocator);
96 SIMIX_mutex_lock(request->simdata->mutex);
98 message->comm = request->comm;
99 message->src = request->src;
100 message->dst = request->dst;
101 message->tag = request->tag;
102 message->buf = xbt_malloc(request->datatype->size * request->count);
103 memcpy(message->buf, request->buf, request->datatype->size * request->count);
105 dhost = request->comm->hosts[request->dst];
106 drank = smpi_mpi_comm_rank(smpi_mpi_global->mpi_comm_world, dhost);
108 SIMIX_mutex_lock(smpi_global->received_message_queues_mutexes[drank]);
109 xbt_fifo_push(smpi_global->received_message_queues[drank], message);
110 SIMIX_mutex_unlock(smpi_global->received_message_queues_mutexes[drank]);
112 request->completed = 1;
114 action = SIMIX_action_communicate(shost, dhost, NULL, request->datatype->size * request->count * 1.0, -1.0);
116 SIMIX_register_action_to_condition(action, request->simdata->cond);
118 SIMIX_cond_wait(request->simdata->cond, request->simdata->mutex);
120 SIMIX_mutex_unlock(request->simdata->mutex);
122 // wake up receiver if necessary
123 receiver_process = smpi_global->receiver_processes[drank];
124 if (SIMIX_process_is_suspended(receiver_process)) {
125 SIMIX_process_resume(receiver_process);
130 SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex);
131 running_hosts_count = smpi_global->running_hosts_count;
132 SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex);
134 } while (0 < running_hosts_count);
136 SIMIX_mutex_lock(smpi_global->start_stop_mutex);
137 smpi_global->ready_process_count--;
138 if (smpi_global->ready_process_count == 0) {
139 SIMIX_cond_broadcast(smpi_global->start_stop_cond);
140 } else if (smpi_global->ready_process_count < 0) {
141 // FIXME: can't happen! abort!
143 SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
148 int smpi_receiver(int argc, char **argv)
153 xbt_fifo_t request_queue;
154 smx_mutex_t request_queue_mutex;
155 xbt_fifo_t message_queue;
156 smx_mutex_t message_queue_mutex;
159 int running_hosts_count;
161 smpi_mpi_request_t *request;
162 smpi_received_message_t *message;
164 xbt_fifo_item_t request_item;
165 xbt_fifo_item_t message_item;
167 self = SIMIX_process_self();
168 rank = smpi_mpi_comm_rank_self(smpi_mpi_global->mpi_comm_world);
170 // make sure root is done before own initialization
171 SIMIX_mutex_lock(smpi_global->start_stop_mutex);
172 if (!smpi_global->root_ready) {
173 SIMIX_cond_wait(smpi_global->start_stop_cond, smpi_global->start_stop_mutex);
175 SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
177 request_queue = smpi_global->pending_recv_request_queues[rank];
178 request_queue_mutex = smpi_global->pending_recv_request_queues_mutexes[rank];
179 message_queue = smpi_global->received_message_queues[rank];
180 message_queue_mutex = smpi_global->received_message_queues_mutexes[rank];
181 size = smpi_mpi_comm_size(smpi_mpi_global->mpi_comm_world);
183 smpi_global->receiver_processes[rank] = self;
185 // wait for all nodes to signal initializatin complete
186 SIMIX_mutex_lock(smpi_global->start_stop_mutex);
187 smpi_global->ready_process_count++;
188 if (smpi_global->ready_process_count < 3 * size) {
189 SIMIX_cond_wait(smpi_global->start_stop_cond, smpi_global->start_stop_mutex);
191 SIMIX_cond_broadcast(smpi_global->start_stop_cond);
193 SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
199 // FIXME: better algorithm, maybe some kind of balanced tree? or a heap?
201 // FIXME: not the best way to request multiple locks...
202 SIMIX_mutex_lock(request_queue_mutex);
203 SIMIX_mutex_lock(message_queue_mutex);
204 for (request_item = xbt_fifo_get_first_item(request_queue);
205 NULL != request_item;
206 request_item = xbt_fifo_get_next_item(request_item)) {
207 request = xbt_fifo_get_item_content(request_item);
208 for (message_item = xbt_fifo_get_first_item(message_queue);
209 NULL != message_item;
210 message_item = xbt_fifo_get_next_item(message_item)) {
211 message = xbt_fifo_get_item_content(message_item);
212 if (request->comm == message->comm &&
213 (MPI_ANY_SOURCE == request->src || request->src == message->src) &&
214 request->tag == message->tag) {
215 xbt_fifo_remove_item(request_queue, request_item);
216 xbt_fifo_remove_item(message_queue, message_item);
222 SIMIX_mutex_unlock(message_queue_mutex);
223 SIMIX_mutex_unlock(request_queue_mutex);
225 if (NULL == request || NULL == message) {
226 SIMIX_process_suspend(self);
228 SIMIX_mutex_lock(request->simdata->mutex);
230 memcpy(request->buf, message->buf, request->datatype->size * request->count);
231 request->src = message->src;
232 request->completed = 1;
233 SIMIX_cond_broadcast(request->simdata->cond);
235 SIMIX_mutex_unlock(request->simdata->mutex);
237 xbt_free(message->buf);
238 xbt_mallocator_release(smpi_global->message_mallocator, message);
241 SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex);
242 running_hosts_count = smpi_global->running_hosts_count;
243 SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex);
245 } while (0 < running_hosts_count);
247 SIMIX_mutex_lock(smpi_global->start_stop_mutex);
248 smpi_global->ready_process_count--;
249 if (smpi_global->ready_process_count == 0) {
250 SIMIX_cond_broadcast(smpi_global->start_stop_cond);
251 } else if (smpi_global->ready_process_count < 0) {
252 // FIXME: can't happen, abort!
254 SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
259 void *smpi_request_new()
261 smpi_mpi_request_t *request = xbt_new(smpi_mpi_request_t, 1);
263 request->completed = 0;
264 request->simdata = xbt_new(s_smpi_mpi_request_simdata_t, 1);
265 request->simdata->mutex = SIMIX_mutex_init();
266 request->simdata->cond = SIMIX_cond_init();
271 void smpi_request_free(void *pointer)
274 smpi_mpi_request_t *request = pointer;
276 if (NULL != request) {
277 SIMIX_cond_destroy(request->simdata->cond);
278 SIMIX_mutex_destroy(request->simdata->mutex);
279 xbt_free(request->simdata);
286 void smpi_request_reset(void *pointer)
292 void *smpi_message_new()
294 return xbt_new(smpi_received_message_t, 1);
297 void smpi_message_free(void *pointer)
299 if (NULL != pointer) {
306 void smpi_message_reset(void *pointer)
311 void smpi_global_init()
315 int size = SIMIX_host_get_number();
317 smpi_global = xbt_new(s_SMPI_Global_t, 1);
320 smpi_global->reference_speed = SMPI_DEFAULT_SPEED;
322 smpi_global->root_ready = 0;
323 smpi_global->ready_process_count = 0;
326 smpi_global->start_stop_mutex = SIMIX_mutex_init();
327 smpi_global->start_stop_cond = SIMIX_cond_init();
330 smpi_global->sender_processes = xbt_new(smx_process_t, size);
331 smpi_global->receiver_processes = xbt_new(smx_process_t, size);
334 smpi_global->running_hosts_count_mutex = SIMIX_mutex_init();
335 smpi_global->running_hosts_count = 0;
338 smpi_global->request_mallocator = xbt_mallocator_new(SMPI_REQUEST_MALLOCATOR_SIZE,
339 smpi_request_new, smpi_request_free, smpi_request_reset);
340 smpi_global->message_mallocator = xbt_mallocator_new(SMPI_MESSAGE_MALLOCATOR_SIZE,
341 smpi_message_new, smpi_message_free, smpi_message_reset);
344 smpi_global->pending_send_request_queues = xbt_new(xbt_fifo_t, size);
345 smpi_global->pending_send_request_queues_mutexes = xbt_new(smx_mutex_t, size);
346 smpi_global->pending_recv_request_queues = xbt_new(xbt_fifo_t, size);
347 smpi_global->pending_recv_request_queues_mutexes = xbt_new(smx_mutex_t, size);
348 smpi_global->received_message_queues = xbt_new(xbt_fifo_t, size);
349 smpi_global->received_message_queues_mutexes = xbt_new(smx_mutex_t, size);
350 smpi_global->timers = xbt_new(xbt_os_timer_t, size);
351 smpi_global->timers_mutexes = xbt_new(smx_mutex_t, size);
353 for(i = 0; i < size; i++) {
354 smpi_global->pending_send_request_queues[i] = xbt_fifo_new();
355 smpi_global->pending_send_request_queues_mutexes[i] = SIMIX_mutex_init();
356 smpi_global->pending_recv_request_queues[i] = xbt_fifo_new();
357 smpi_global->pending_recv_request_queues_mutexes[i] = SIMIX_mutex_init();
358 smpi_global->received_message_queues[i] = xbt_fifo_new();
359 smpi_global->received_message_queues_mutexes[i] = SIMIX_mutex_init();
360 smpi_global->timers[i] = xbt_os_timer_new();
361 smpi_global->timers_mutexes[i] = SIMIX_mutex_init();
366 void smpi_global_destroy()
370 int size = SIMIX_host_get_number();
373 SIMIX_mutex_destroy(smpi_global->start_stop_mutex);
374 SIMIX_cond_destroy(smpi_global->start_stop_cond);
377 xbt_free(smpi_global->sender_processes);
378 xbt_free(smpi_global->receiver_processes);
381 SIMIX_mutex_destroy(smpi_global->running_hosts_count_mutex);
384 xbt_mallocator_free(smpi_global->request_mallocator);
385 xbt_mallocator_free(smpi_global->message_mallocator);
387 for(i = 0; i < size; i++) {
388 xbt_fifo_free(smpi_global->pending_send_request_queues[i]);
389 SIMIX_mutex_destroy(smpi_global->pending_send_request_queues_mutexes[i]);
390 xbt_fifo_free(smpi_global->pending_recv_request_queues[i]);
391 SIMIX_mutex_destroy(smpi_global->pending_recv_request_queues_mutexes[i]);
392 xbt_fifo_free(smpi_global->received_message_queues[i]);
393 SIMIX_mutex_destroy(smpi_global->received_message_queues_mutexes[i]);
394 xbt_os_timer_free(smpi_global->timers[i]);
395 SIMIX_mutex_destroy(smpi_global->timers_mutexes[i]);
398 xbt_free(smpi_global->pending_send_request_queues);
399 xbt_free(smpi_global->pending_send_request_queues_mutexes);
400 xbt_free(smpi_global->pending_recv_request_queues);
401 xbt_free(smpi_global->pending_recv_request_queues_mutexes);
402 xbt_free(smpi_global->received_message_queues);
403 xbt_free(smpi_global->received_message_queues_mutexes);
404 xbt_free(smpi_global->timers);
405 xbt_free(smpi_global->timers_mutexes);
407 xbt_free(smpi_global);
410 int smpi_run_simulation(int argc, char **argv)
412 xbt_fifo_item_t cond_item = NULL;
413 smx_cond_t cond = NULL;
414 smx_action_t action = NULL;
416 xbt_fifo_t actions_failed = xbt_fifo_new();
417 xbt_fifo_t actions_done = xbt_fifo_new();
419 srand(SMPI_RAND_SEED);
421 SIMIX_global_init(&argc, argv);
423 SIMIX_function_register("smpi_simulated_main", smpi_simulated_main);
424 SIMIX_function_register("smpi_sender", smpi_sender);
425 SIMIX_function_register("smpi_receiver", smpi_receiver);
427 // FIXME: ought to verify these files...
428 SIMIX_create_environment(argv[1]);
430 // must initialize globals between creating environment and launching app....
433 SIMIX_launch_application(argv[2]);
435 /* Prepare to display some more info when dying on Ctrl-C pressing */
436 // FIXME: doesn't work
437 //signal(SIGINT, inthandler);
439 /* Clean IO before the run */
443 while (SIMIX_solve(actions_done, actions_failed) != -1.0) {
444 while ((action = xbt_fifo_pop(actions_failed))) {
445 DEBUG1("** %s failed **", action->name);
446 xbt_fifo_foreach(action->cond_list, cond_item, cond, smx_cond_t) {
447 SIMIX_cond_broadcast(cond);
448 SIMIX_unregister_action_to_condition(action, cond);
450 SIMIX_action_destroy(action);
452 while ((action = xbt_fifo_pop(actions_done))) {
453 DEBUG1("** %s done **",action->name);
454 xbt_fifo_foreach(action->cond_list, cond_item, cond, smx_cond_t) {
455 SIMIX_cond_broadcast(cond);
456 SIMIX_unregister_action_to_condition(action, cond);
458 SIMIX_action_destroy(action);
462 xbt_fifo_free(actions_failed);
463 xbt_fifo_free(actions_done);
465 INFO1("simulation time %g", SIMIX_get_clock());
467 smpi_global_destroy();
474 void smpi_mpi_land_func(void *x, void *y, void *z)
476 *(int *)z = *(int *)x && *(int *)y;
479 void smpi_mpi_sum_func(void *x, void *y, void *z)
481 *(int *)z = *(int *)x + *(int *)y;
487 smx_process_t process;
492 SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex);
493 smpi_global->running_hosts_count++;
494 SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex);
496 // initialize some local variables
497 process = SIMIX_process_self();
498 host = SIMIX_host_self();
499 hosts = SIMIX_host_get_table();
500 size = SIMIX_host_get_number();
502 // node 0 sets the globals
503 if (host == hosts[0]) {
505 smpi_mpi_global = xbt_new(s_SMPI_MPI_Global_t, 1);
507 // global communicator
508 smpi_mpi_global->mpi_comm_world = xbt_new(smpi_mpi_communicator_t, 1);
509 smpi_mpi_global->mpi_comm_world->size = size;
510 smpi_mpi_global->mpi_comm_world->simdata = xbt_new(s_smpi_mpi_communicator_simdata_t, 1);
511 smpi_mpi_global->mpi_comm_world->simdata->barrier_count = 0;
512 smpi_mpi_global->mpi_comm_world->simdata->barrier_mutex = SIMIX_mutex_init();
513 smpi_mpi_global->mpi_comm_world->simdata->barrier_cond = SIMIX_cond_init();
514 smpi_mpi_global->mpi_comm_world->simdata->hosts = hosts;
515 smpi_mpi_global->mpi_comm_world->simdata->processes = xbt_new(smx_process_t, size);
516 smpi_mpi_global->mpi_comm_world->simdata->processes[0] = process;
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);
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;
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);
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);
545 SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
547 smpi_mpi_global->mpi_comm_world->processes[smpi_mpi_comm_rank_self(smpi_mpi_global->mpi_comm_world)] = process;
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);
556 SIMIX_cond_broadcast(smpi_global->start_stop_cond);
558 SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
563 void smpi_mpi_finalize()
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);
571 SIMIX_mutex_lock(smpi_global->start_stop_mutex);
572 smpi_global->ready_process_count--;
573 SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
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]);
582 if (SIMIX_process_is_suspended(smpi_global->receiver_processes[i])) {
583 SIMIX_process_resume(smpi_global->receiver_processes[i]);
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);
592 SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
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);
599 xbt_free(smpi_mpi_global->mpi_byte);
600 xbt_free(smpi_mpi_global->mpi_int);
601 xbt_free(smpi_mpi_global->mpi_double);
603 xbt_free(smpi_mpi_global->mpi_land);
604 xbt_free(smpi_mpi_global->mpi_sum);
606 xbt_free(smpi_mpi_global);
611 // FIXME: could cause trouble with multithreaded procs on same host...
612 void smpi_bench_begin()
614 int rank = smpi_mpi_comm_rank_self(smpi_mpi_global->mpi_comm_world);
615 SIMIX_mutex_lock(smpi_global->timers_mutexes[rank]);
616 xbt_os_timer_start(smpi_global->timers[rank]);
620 void smpi_bench_end()
622 int rank = smpi_mpi_comm_rank_self(smpi_mpi_global->mpi_comm_world);
625 smx_action_t compute_action;
629 xbt_os_timer_stop(smpi_global->timers[rank]);
631 duration = xbt_os_timer_elapsed(smpi_global->timers[rank]);
632 SIMIX_mutex_unlock(smpi_global->timers_mutexes[rank]);
634 host = smpi_mpi_global->mpi_comm_world->hosts[rank];
635 compute_action = SIMIX_action_execute(host, NULL, duration * SMPI_DEFAULT_SPEED);
636 mutex = SIMIX_mutex_init();
637 cond = SIMIX_cond_init();
639 SIMIX_mutex_lock(mutex);
640 SIMIX_register_action_to_condition(compute_action, cond);
641 SIMIX_cond_wait(cond, mutex);
642 //SIMIX_unregister_action_to_condition(compute_action, cond);
643 SIMIX_mutex_unlock(mutex);
645 SIMIX_mutex_destroy(mutex);
646 SIMIX_cond_destroy(cond);
648 // FIXME: check for success/failure?
653 void smpi_barrier(smpi_mpi_communicator_t *comm)
656 SIMIX_mutex_lock(comm->simdata->barrier_mutex);
657 if(++comm->simdata->barrier_count < comm->size) {
658 SIMIX_cond_wait(comm->simdata->barrier_cond, comm->simdata->barrier_mutex);
660 comm->simdata->barrier_count = 0;
661 SIMIX_cond_broadcast(comm->simdata->barrier_cond);
663 SIMIX_mutex_unlock(comm->simdata->barrier_mutex);
668 // FIXME: smarter algorithm...
669 int smpi_comm_rank(smpi_mpi_communicator_t *comm, smx_host_t host)
672 for(i = 0; i < comm->size && host != comm->hosts[i]; i++);
673 if (i >= comm->size) i = -1;
677 int smpi_create_request(void *buf, int count, smpi_mpi_datatype_t *datatype,
678 int src, int dst, int tag, smpi_mpi_communicator_t *comm, smpi_mpi_request_t **request)
680 int retval = MPI_SUCCESS;
685 retval = MPI_ERR_COUNT;
686 } else if (NULL == buf) {
687 retval = MPI_ERR_INTERN;
688 } else if (NULL == datatype) {
689 retval = MPI_ERR_TYPE;
690 } else if (NULL == comm) {
691 retval = MPI_ERR_COMM;
692 } else if (MPI_ANY_SOURCE != src && (0 > src || comm->size <= src)) {
693 retval = MPI_ERR_RANK;
694 } else if (0 > dst || comm->size <= dst) {
695 retval = MPI_ERR_RANK;
696 } else if (0 > tag) {
697 retval = MPI_ERR_TAG;
699 *request = xbt_mallocator_get(smpi_global->request_mallocator);
700 (*request)->comm = comm;
701 (*request)->src = src;
702 (*request)->dst = dst;
703 (*request)->tag = tag;
704 (*request)->buf = buf;
705 (*request)->count = count;
706 (*request)->datatype = datatype;
711 int smpi_isend(smpi_mpi_request_t *request)
713 int retval = MPI_SUCCESS;
714 int rank = smpi_mpi_comm_rank_self(smpi_mpi_global->mpi_comm_world);
716 if (NULL != request) {
717 SIMIX_mutex_lock(smpi_global->pending_send_request_queues_mutexes[rank]);
718 xbt_fifo_push(smpi_global->pending_send_request_queues[rank], request);
719 SIMIX_mutex_unlock(smpi_global->pending_send_request_queues_mutexes[rank]);
722 if (SIMIX_process_is_suspended(smpi_global->sender_processes[rank])) {
723 SIMIX_process_resume(smpi_global->sender_processes[rank]);
729 int smpi_irecv(smpi_mpi_request_t *request)
731 int retval = MPI_SUCCESS;
732 int rank = smpi_mpi_comm_rank_self(smpi_mpi_global->mpi_comm_world);
734 if (NULL != request) {
735 SIMIX_mutex_lock(smpi_global->pending_recv_request_queues_mutexes[rank]);
736 xbt_fifo_push(smpi_global->pending_recv_request_queues[rank], request);
737 SIMIX_mutex_unlock(smpi_global->pending_recv_request_queues_mutexes[rank]);
740 if (SIMIX_process_is_suspended(smpi_global->receiver_processes[rank])) {
741 SIMIX_process_resume(smpi_global->receiver_processes[rank]);
747 void smpi_wait(smpi_mpi_request_t *request, smpi_mpi_status_t *status)
749 if (NULL != request) {
750 SIMIX_mutex_lock(request->simdata->mutex);
751 if (!request->completed) {
752 SIMIX_cond_wait(request->simdata->cond, request->simdata->mutex);
754 if (NULL != status) {
755 status->MPI_SOURCE = request->src;
757 SIMIX_mutex_unlock(request->simdata->mutex);
761 // FIXME: move into own file
762 int smpi_gettimeofday(struct timeval *tv, struct timezone *tz)
770 now = SIMIX_get_clock();
772 tv->tv_usec = ((now - (double)tv->tv_sec) * 1000000.0);
778 unsigned int smpi_sleep(unsigned int seconds)
783 smx_action_t sleep_action;
786 host = SIMIX_host_self();
787 sleep_action = SIMIX_action_sleep(host, seconds);
788 mutex = SIMIX_mutex_init();
789 cond = SIMIX_cond_init();
791 SIMIX_mutex_lock(mutex);
792 SIMIX_register_action_to_condition(sleep_action, cond);
793 SIMIX_cond_wait(cond, mutex);
794 //SIMIX_unregister_action_to_condition(sleep_action, cond);
795 SIMIX_mutex_unlock(mutex);
797 SIMIX_mutex_destroy(mutex);
798 SIMIX_cond_destroy(cond);
800 // FIXME: check for success/failure?
806 void smpi_exit(int status)
809 SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex);
810 smpi_global->running_hosts_count--;
811 SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex);
812 SIMIX_process_kill(SIMIX_process_self());