+static void action_send(const char *const *action)
+{
+ char *name = NULL;
+ char to[250];
+ const char *size_str = action[3];
+ double size=parse_double(size_str);
+ double clock = MSG_get_clock(); /* this "call" is free thanks to inlining */
+
+ sprintf(to, "%s_%s", MSG_process_get_name(MSG_process_self()),action[2]);
+
+ if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
+ name = xbt_str_join_array(action, " ");
+
+#ifdef HAVE_TRACING
+ int rank = get_rank(MSG_process_get_name(MSG_process_self()));
+ int dst_traced = get_rank(action[2]);
+ TRACE_smpi_ptp_in(rank, rank, dst_traced, "send");
+ TRACE_smpi_send(rank, rank, dst_traced);
+#endif
+
+ XBT_DEBUG("Entering Send: %s (size: %lg)", name, size);
+ if (size<65536) {
+ action_Isend(action);
+ } else {
+ MSG_task_send(MSG_task_create(name, 0, size, NULL), to);
+ }
+
+ XBT_VERB("%s %f", name, MSG_get_clock() - clock);
+
+ if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
+ free(name);
+
+#ifdef HAVE_TRACING
+ TRACE_smpi_ptp_out(rank, rank, dst_traced, "send");
+#endif
+
+ asynchronous_cleanup();
+}
+
+static void action_Isend(const char *const *action)
+{
+ char to[250];
+ const char *size = action[3];
+ double clock = MSG_get_clock();
+ process_globals_t globals = (process_globals_t) MSG_process_get_data(MSG_process_self());
+
+
+ sprintf(to, "%s_%s", MSG_process_get_name(MSG_process_self()),action[2]);
+ msg_comm_t comm =
+ MSG_task_isend( MSG_task_create(to,0,parse_double(size),NULL), to);
+ xbt_dynar_push(globals->isends,&comm);
+
+ XBT_DEBUG("Isend on %s", MSG_process_get_name(MSG_process_self()));
+ XBT_VERB("%s %f", xbt_str_join_array(action, " "), MSG_get_clock() - clock);
+
+ asynchronous_cleanup();
+}
+
+
+static void action_recv(const char *const *action)
+{
+ char *name = NULL;
+ char mailbox_name[250];
+ m_task_t task = NULL;
+ double clock = MSG_get_clock();
+
+ sprintf(mailbox_name, "%s_%s", action[2],
+ MSG_process_get_name(MSG_process_self()));
+
+ if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
+ name = xbt_str_join_array(action, " ");
+
+#ifdef HAVE_TRACING
+ int rank = get_rank(MSG_process_get_name(MSG_process_self()));
+ int src_traced = get_rank(action[2]);
+ TRACE_smpi_ptp_in(rank, src_traced, rank, "recv");
+#endif
+
+ XBT_DEBUG("Receiving: %s", name);
+ MSG_error_t res = MSG_task_receive(&task, mailbox_name);
+ // MSG_task_receive(&task, MSG_process_get_name(MSG_process_self()));
+ XBT_VERB("%s %f", name, MSG_get_clock() - clock);
+
+ if (res == MSG_OK) {
+ MSG_task_destroy(task);
+ }
+
+ if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
+ free(name);
+#ifdef HAVE_TRACING
+ TRACE_smpi_ptp_out(rank, src_traced, rank, "recv");
+ TRACE_smpi_recv(rank, src_traced, rank);
+#endif
+
+ asynchronous_cleanup();
+}
+
+static void action_Irecv(const char *const *action)
+{
+ char mailbox[250];
+ double clock = MSG_get_clock();
+ process_globals_t globals = (process_globals_t) MSG_process_get_data(MSG_process_self());
+
+ XBT_DEBUG("Irecv on %s", MSG_process_get_name(MSG_process_self()));
+#ifdef HAVE_TRACING
+ int rank = get_rank(MSG_process_get_name(MSG_process_self()));
+ int src_traced = get_rank(action[2]);
+ globals->last_Irecv_sender_id = src_traced;
+ MSG_process_set_data(MSG_process_self(), (void *) globals);
+
+ TRACE_smpi_ptp_in(rank, src_traced, rank, "Irecv");
+#endif
+
+ sprintf(mailbox, "%s_%s", action[2],
+ MSG_process_get_name(MSG_process_self()));
+ m_task_t t=NULL;
+ xbt_dynar_push(globals->tasks,&t);
+ msg_comm_t c =
+ MSG_task_irecv(
+ xbt_dynar_get_ptr(globals->tasks, xbt_dynar_length(globals->tasks)-1),
+ mailbox);
+ xbt_dynar_push(globals->irecvs,&c);
+
+ XBT_VERB("%s %f", xbt_str_join_array(action, " "), MSG_get_clock() - clock);
+
+#ifdef HAVE_TRACING
+ TRACE_smpi_ptp_out(rank, src_traced, rank, "Irecv");
+#endif
+
+ asynchronous_cleanup();
+}
+
+
+static void action_wait(const char *const *action)
+{
+ char *name = NULL;
+ m_task_t task = NULL;
+ msg_comm_t comm;
+ double clock = MSG_get_clock();
+ process_globals_t globals = (process_globals_t) MSG_process_get_data(MSG_process_self());
+
+ xbt_assert(xbt_dynar_length(globals->irecvs),
+ "action wait not preceded by any irecv: %s", xbt_str_join_array(action," "));
+
+ if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
+ name = xbt_str_join_array(action, " ");
+#ifdef HAVE_TRACING
+ process_globals_t counters = (process_globals_t) MSG_process_get_data(MSG_process_self());
+ int src_traced = counters->last_Irecv_sender_id;
+ int rank = get_rank(MSG_process_get_name(MSG_process_self()));
+ TRACE_smpi_ptp_in(rank, src_traced, rank, "wait");
+#endif
+
+ XBT_DEBUG("Entering %s", name);
+ comm = xbt_dynar_pop_as(globals->irecvs,msg_comm_t);
+ MSG_comm_wait(comm,-1);
+ task = xbt_dynar_pop_as(globals->tasks,m_task_t);
+ MSG_comm_destroy(comm);
+ MSG_task_destroy(task);
+
+ XBT_VERB("%s %f", name, MSG_get_clock() - clock);
+ if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
+ free(name);
+#ifdef HAVE_TRACING
+ TRACE_smpi_ptp_out(rank, src_traced, rank, "wait");
+ TRACE_smpi_recv(rank, src_traced, rank);
+#endif
+
+}
+
+/* FIXME: that's a poor man's implementation: we should take the message exchanges into account */
+static void action_barrier(const char *const *action)
+{
+ char *name = NULL;
+ static smx_mutex_t mutex = NULL;
+ static smx_cond_t cond = NULL;
+ static int processes_arrived_sofar=0;
+
+ if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
+ name = xbt_str_join_array(action, " ");
+
+ if (mutex == NULL) { // first arriving on the barrier
+ mutex = SIMIX_req_mutex_init();
+ cond = SIMIX_req_cond_init();
+ processes_arrived_sofar=0;
+ }
+ XBT_DEBUG("Entering barrier: %s (%d already there)", name,processes_arrived_sofar);
+
+ SIMIX_req_mutex_lock(mutex);
+ if (++processes_arrived_sofar == communicator_size) {
+ SIMIX_req_cond_broadcast(cond);
+ SIMIX_req_mutex_unlock(mutex);
+ } else {
+ SIMIX_req_cond_wait(cond,mutex);
+ SIMIX_req_mutex_unlock(mutex);
+ }
+
+ XBT_DEBUG("Exiting barrier: %s", name);
+
+ processes_arrived_sofar--;
+ if (!processes_arrived_sofar) {
+ SIMIX_req_cond_destroy(cond);
+ SIMIX_req_mutex_destroy(mutex);
+ mutex=NULL;
+ }
+
+ if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose))
+ free(name);
+
+}
+
+static void action_reduce(const char *const *action)
+{
+ int i;
+ char *reduce_identifier;
+ char mailbox[80];
+ double comm_size = parse_double(action[2]);
+ double comp_size = parse_double(action[3]);
+ m_task_t comp_task = NULL;
+ const char *process_name;
+ double clock = MSG_get_clock();
+
+ process_globals_t counters = (process_globals_t) MSG_process_get_data(MSG_process_self());
+
+ xbt_assert(communicator_size, "Size of Communicator is not defined, "
+ "can't use collective operations");
+
+ process_name = MSG_process_get_name(MSG_process_self());
+
+ reduce_identifier = bprintf("reduce_%d", counters->reduce_counter++);
+
+ if (!strcmp(process_name, "p0")) {
+ XBT_DEBUG("%s: %s is the Root", reduce_identifier, process_name);
+
+ msg_comm_t *comms = xbt_new0(msg_comm_t,communicator_size-1);
+ m_task_t *tasks = xbt_new0(m_task_t,communicator_size-1);
+ for (i = 1; i < communicator_size; i++) {
+ sprintf(mailbox, "%s_p%d_p0", reduce_identifier, i);
+ comms[i-1] = MSG_task_irecv(&(tasks[i-1]),mailbox);
+ }
+ MSG_comm_waitall(comms,communicator_size-1,-1);
+ for (i = 1; i < communicator_size; i++) {
+ MSG_comm_destroy(comms[i-1]);
+ MSG_task_destroy(tasks[i-1]);
+ }
+ free(tasks);
+
+ comp_task = MSG_task_create("reduce_comp", comp_size, 0, NULL);
+ XBT_DEBUG("%s: computing 'reduce_comp'", reduce_identifier);
+ MSG_task_execute(comp_task);
+ MSG_task_destroy(comp_task);
+ XBT_DEBUG("%s: computed", reduce_identifier);
+
+ } else {
+ XBT_DEBUG("%s: %s sends", reduce_identifier, process_name);
+ sprintf(mailbox, "%s_%s_p0", reduce_identifier, process_name);
+ XBT_DEBUG("put on %s", mailbox);
+ MSG_task_send(MSG_task_create(reduce_identifier, 0, comm_size, NULL),
+ mailbox);
+ }
+
+ XBT_VERB("%s %f", xbt_str_join_array(action, " "), MSG_get_clock() - clock);
+ free(reduce_identifier);
+}
+
+static void action_bcast(const char *const *action)
+{
+ int i;
+ char *bcast_identifier;
+ char mailbox[80];
+ double comm_size = parse_double(action[2]);