X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/8cbcd8756ae9134b66658b4ccb9cee91352da649..eb9d1d343764af46a2fbd9c44f6246e769bea5d3:/examples/msg/actions/actions.c diff --git a/examples/msg/actions/actions.c b/examples/msg/actions/actions.c index f5be73322f..0958e5638e 100644 --- a/examples/msg/actions/actions.c +++ b/examples/msg/actions/actions.c @@ -23,7 +23,10 @@ typedef struct { int bcast_counter; int reduce_counter; int allReduce_counter; - xbt_dynar_t isends; + xbt_dynar_t isends; /* of msg_comm_t */ + /* Used to implement irecv+wait */ + xbt_dynar_t irecvs; /* of msg_comm_t */ + xbt_dynar_t tasks; /* of m_task_t */ } s_process_globals_t, *process_globals_t; /* Helper function */ @@ -56,15 +59,6 @@ static void asynchronous_cleanup(void) { } /* My actions */ -static int spawned_send(int argc, char *argv[]) -{ - DEBUG3("%s: Sending %s on %s", MSG_process_get_name(MSG_process_self()), - argv[1], argv[0]); - MSG_task_send(MSG_task_create(argv[0], 0, parse_double(argv[1]), NULL), - argv[0]); - return 0; -} - static void action_send(xbt_dynar_t action) { char *name = NULL; @@ -135,8 +129,7 @@ static void action_recv(xbt_dynar_t action) char mailbox_name[250]; m_task_t task = NULL; double clock = MSG_get_clock(); - //FIXME: argument of action ignored so far; semantic not clear - //char *from=xbt_dynar_get_as(action,2,char*); + sprintf(mailbox_name, "%s_%s", xbt_dynar_get_as(action, 2, char *), MSG_process_get_name(MSG_process_self())); @@ -165,55 +158,34 @@ static void action_recv(xbt_dynar_t action) asynchronous_cleanup(); } -static int spawned_recv(int argc, char *argv[]) -{ - m_task_t task = NULL; - DEBUG1("Receiving on %s", argv[0]); - MSG_task_receive(&task, argv[0]); - DEBUG1("Received %s", MSG_task_get_name(task)); - DEBUG1("waiter on %s", MSG_process_get_name(MSG_process_self())); - MSG_task_send(MSG_task_create("waiter", 0, 0, NULL), - MSG_process_get_name(MSG_process_self())); - - MSG_task_destroy(task); - return 0; -} - - static void action_Irecv(xbt_dynar_t action) { - char *name; - m_process_t comm_helper; - char mailbox_name[250]; - char **myargv; + char mailbox[250]; double clock = MSG_get_clock(); + process_globals_t globals = (process_globals_t) MSG_process_get_data(MSG_process_self()); - DEBUG1("Irecv on %s: spawn process ", - MSG_process_get_name(MSG_process_self())); + DEBUG1("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(xbt_dynar_get_as(action, 2, char *)); - process_globals_t counters = (process_globals_t) MSG_process_get_data(MSG_process_self()); - counters->last_Irecv_sender_id = src_traced; - MSG_process_set_data(MSG_process_self(), (void *) counters); + 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_name, "%s_%s", xbt_dynar_get_as(action, 2, char *), + sprintf(mailbox, "%s_%s", xbt_dynar_get_as(action, 2, char *), MSG_process_get_name(MSG_process_self())); - name = bprintf("%s_wait", MSG_process_get_name(MSG_process_self())); - myargv = (char **) calloc(2, sizeof(char *)); - - myargv[0] = xbt_strdup(mailbox_name); - myargv[1] = NULL; - comm_helper = MSG_process_create_with_arguments(name, spawned_recv, - NULL, MSG_host_self(), - 1, myargv); + 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); VERB2("%s %f", xbt_str_join(action, " "), MSG_get_clock() - clock); - free(name); #ifdef HAVE_TRACING TRACE_smpi_ptp_out(rank, src_traced, rank, "Irecv"); #endif @@ -225,9 +197,13 @@ static void action_Irecv(xbt_dynar_t action) static void action_wait(xbt_dynar_t action) { char *name = NULL; - char task_name[80]; 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_assert1(xbt_dynar_length(globals->irecvs), + "action wait not preceeded by any irecv: %s", xbt_str_join(action," ")); if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose)) name = xbt_str_join(action, " "); @@ -239,10 +215,12 @@ static void action_wait(xbt_dynar_t action) #endif DEBUG1("Entering %s", name); - sprintf(task_name, "%s_wait", MSG_process_get_name(MSG_process_self())); - DEBUG1("wait: %s", task_name); - MSG_task_receive(&task, task_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); + VERB2("%s %f", name, MSG_get_clock() - clock); if (XBT_LOG_ISENABLED(actions, xbt_log_priority_verbose)) free(name); @@ -296,129 +274,105 @@ static void action_barrier(xbt_dynar_t action) static void action_reduce(xbt_dynar_t action) { - int i; - char *name; - char task_name[80]; - char spawn_name[80]; - char **myargv; - char *comm_size = xbt_dynar_get_as(action, 2, char *); - char *comp_size = xbt_dynar_get_as(action, 3, char *); - m_process_t comm_helper = NULL; - m_task_t task = NULL, 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_assert0(communicator_size, "Size of Communicator is not defined" - ", can't use collective operations"); - - process_name = MSG_process_get_name(MSG_process_self()); - - name = bprintf("reduce_%d", counters->reduce_counter++); - - if (!strcmp(process_name, "p0")) { - DEBUG2("%s: %s is the Root", name, process_name); - for (i = 1; i < communicator_size; i++) { - sprintf(spawn_name, "%s_p%d_%s", name, i, - MSG_process_get_name(MSG_process_self())); - sprintf(task_name, "%s_wait", spawn_name); - myargv = (char **) calloc(2, sizeof(char *)); - - myargv[0] = xbt_strdup(spawn_name); - myargv[1] = NULL; - - comm_helper = - MSG_process_create_with_arguments(task_name, spawned_recv, - NULL, MSG_host_self(), - 1, myargv); - } - - for (i = 1; i < communicator_size; i++) { - sprintf(task_name, "%s_p%d_p0_wait", name, i); - MSG_task_receive(&task, task_name); - MSG_task_destroy(task); - task = NULL; - } - - comp_task = - MSG_task_create("reduce_comp", parse_double(comp_size), 0, NULL); - DEBUG1("%s: computing 'reduce_comp'", name); - MSG_task_execute(comp_task); - MSG_task_destroy(comp_task); - DEBUG1("%s: computed", name); - } else { - DEBUG2("%s: %s sends", name, process_name); - sprintf(task_name, "%s_%s_p0", name, process_name); - DEBUG1("put on %s", task_name); - MSG_task_send(MSG_task_create(name, 0, parse_double(comm_size), NULL), - task_name); - } - - MSG_process_set_data(MSG_process_self(), (void *) counters); - VERB2("%s %f", xbt_str_join(action, " "), MSG_get_clock() - clock); - free(name); + int i; + char *reduce_identifier; + char mailbox[80]; + double comm_size = parse_double(xbt_dynar_get_as(action, 2, char *)); + double comp_size = parse_double(xbt_dynar_get_as(action, 3, char *)); + 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_assert0(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")) { + DEBUG2("%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); + DEBUG1("%s: computing 'reduce_comp'", reduce_identifier); + MSG_task_execute(comp_task); + MSG_task_destroy(comp_task); + DEBUG1("%s: computed", reduce_identifier); + + } else { + DEBUG2("%s: %s sends", reduce_identifier, process_name); + sprintf(mailbox, "%s_%s_p0", reduce_identifier, process_name); + DEBUG1("put on %s", mailbox); + MSG_task_send(MSG_task_create(reduce_identifier, 0, comm_size, NULL), + mailbox); + } + + VERB2("%s %f", xbt_str_join(action, " "), MSG_get_clock() - clock); + free(reduce_identifier); } static void action_bcast(xbt_dynar_t action) { - int i; - char *name; - const char *process_name; - char task_name[80]; - char spawn_name[80]; - char **myargv; - m_process_t comm_helper = NULL; - m_task_t task = NULL; - char *size = xbt_dynar_get_as(action, 2, char *); - process_globals_t counters = (process_globals_t) MSG_process_get_data(MSG_process_self()); - double clock = MSG_get_clock(); - - xbt_assert0(communicator_size, "Size of Communicator is not defined" - ", can't use collective operations"); - - - process_name = MSG_process_get_name(MSG_process_self()); - - name = bprintf("bcast_%d", counters->bcast_counter++); - if (!strcmp(process_name, "p0")) { - DEBUG2("%s: %s is the Root", name, process_name); - - for (i = 1; i < communicator_size; i++) { - myargv = (char **) calloc(3, sizeof(char *)); - myargv[0] = xbt_strdup(name); - myargv[1] = xbt_strdup(size); - myargv[2] = NULL; - - sprintf(spawn_name, "%s_%d", myargv[0], i); - comm_helper = - MSG_process_create_with_arguments(spawn_name, spawned_send, - NULL, MSG_host_self(), 2, - myargv); - } - - for (i = 1; i < communicator_size; i++) { - sprintf(task_name, "p%d_wait", i); - DEBUG1("get on %s", task_name); - MSG_task_receive(&task, task_name); - MSG_task_destroy(task); - task = NULL; - } - DEBUG2("%s: all messages sent by %s have been received", - name, process_name); - } else { - DEBUG2("%s: %s receives", name, process_name); - MSG_task_receive(&task, name); - MSG_task_destroy(task); - DEBUG2("%s: %s has received", name, process_name); - sprintf(task_name, "%s_wait", process_name); - DEBUG1("put on %s", task_name); - MSG_task_send(MSG_task_create("waiter", 0, 0, NULL), task_name); - } - - MSG_process_set_data(MSG_process_self(), (void *) counters); - VERB2("%s %f", xbt_str_join(action, " "), MSG_get_clock() - clock); - free(name); + int i; + char *bcast_identifier; + char mailbox[80]; + double comm_size = parse_double(xbt_dynar_get_as(action, 2, char *)); + m_task_t 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_assert0(communicator_size, "Size of Communicator is not defined, " + "can't use collective operations"); + + process_name = MSG_process_get_name(MSG_process_self()); + + bcast_identifier = bprintf("bcast_%d", counters->bcast_counter++); + + if (!strcmp(process_name, "p0")) { + DEBUG2("%s: %s is the Root", bcast_identifier, process_name); + + msg_comm_t *comms = xbt_new0(msg_comm_t,communicator_size-1); + + for (i = 1; i < communicator_size; i++) { + sprintf(mailbox, "%s_p0_p%d", bcast_identifier, i); + comms[i-1] = + MSG_task_isend(MSG_task_create(mailbox,0,comm_size,NULL), + mailbox); + } + MSG_comm_waitall(comms,communicator_size-1,-1); + for (i = 1; i < communicator_size; i++) + MSG_comm_destroy(comms[i-1]); + free(comms); + + DEBUG2("%s: all messages sent by %s have been received", + bcast_identifier, process_name); + + } else { + sprintf(mailbox, "%s_p0_%s", bcast_identifier, process_name); + MSG_task_receive(&task, mailbox); + MSG_task_destroy(task); + DEBUG2("%s: %s has received", bcast_identifier, process_name); + } + + VERB2("%s %f", xbt_str_join(action, " "), MSG_get_clock() - clock); + free(bcast_identifier); } @@ -469,7 +423,7 @@ static void action_allReduce(xbt_dynar_t action) { } MSG_comm_waitall(comms,communicator_size-1,-1); for (i = 1; i < communicator_size; i++) { - // MSG_comm_destroy(comms[i-1]); + MSG_comm_destroy(comms[i-1]); MSG_task_destroy(tasks[i-1]); } free(tasks); @@ -487,8 +441,8 @@ static void action_allReduce(xbt_dynar_t action) { mailbox); } MSG_comm_waitall(comms,communicator_size-1,-1); - /* for (i = 1; i < communicator_size; i++) */ - /* MSG_comm_destroy(comms[i-1]); */ + for (i = 1; i < communicator_size; i++) + MSG_comm_destroy(comms[i-1]); free(comms); DEBUG2("%s: all messages sent by %s have been received", @@ -550,6 +504,8 @@ static void action_init(xbt_dynar_t action) DEBUG0("Initialize the counters"); process_globals_t globals = (process_globals_t) calloc(1, sizeof(s_process_globals_t)); globals->isends = xbt_dynar_new(sizeof(msg_comm_t),NULL); + globals->irecvs = xbt_dynar_new(sizeof(msg_comm_t),NULL); + globals->tasks = xbt_dynar_new(sizeof(m_task_t),NULL); MSG_process_set_data(MSG_process_self(),globals); }