1 /* Copyright (c) 2009-2014. The SimGrid Team.
2 * All rights reserved. */
4 /* This program is free software; you can redistribute it and/or modify it
5 * under the terms of the license (GNU LGPL) which comes with this package. */
7 #include "smx_private.h"
11 #include "smpi/private.h"
14 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_network, simix,
15 "Logging specific to SIMIX (network)");
17 static xbt_dict_t rdv_points = NULL;
18 XBT_EXPORT_NO_IMPORT(unsigned long int) smx_total_comms = 0;
20 static void SIMIX_waitany_remove_simcall_from_actions(smx_simcall_t simcall);
21 static void SIMIX_comm_copy_data(smx_action_t comm);
22 static smx_action_t SIMIX_comm_new(e_smx_comm_type_t type);
23 static XBT_INLINE void SIMIX_rdv_push(smx_rdv_t rdv, smx_action_t comm);
24 static smx_action_t SIMIX_fifo_probe_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
25 int (*match_fun)(void *, void *,smx_action_t),
26 void *user_data, smx_action_t my_action);
27 static smx_action_t SIMIX_fifo_get_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
28 int (*match_fun)(void *, void *,smx_action_t),
29 void *user_data, smx_action_t my_action);
30 static void SIMIX_rdv_free(void *data);
31 static void SIMIX_comm_start(smx_action_t action);
33 void SIMIX_network_init(void)
35 rdv_points = xbt_dict_new_homogeneous(SIMIX_rdv_free);
38 void SIMIX_network_exit(void)
40 xbt_dict_free(&rdv_points);
43 /******************************************************************************/
44 /* Rendez-Vous Points */
45 /******************************************************************************/
47 smx_rdv_t SIMIX_pre_rdv_create(smx_simcall_t simcall, const char *name){
48 return SIMIX_rdv_create(name);
50 smx_rdv_t SIMIX_rdv_create(const char *name)
52 /* two processes may have pushed the same rdv_create simcall at the same time */
53 smx_rdv_t rdv = name ? xbt_dict_get_or_null(rdv_points, name) : NULL;
56 rdv = xbt_new0(s_smx_rvpoint_t, 1);
57 rdv->name = name ? xbt_strdup(name) : NULL;
58 rdv->comm_fifo = xbt_fifo_new();
59 rdv->done_comm_fifo = xbt_fifo_new();
60 rdv->permanent_receiver=NULL;
62 XBT_DEBUG("Creating a mailbox at %p with name %s\n", rdv, name);
65 xbt_dict_set(rdv_points, rdv->name, rdv, NULL);
70 void SIMIX_pre_rdv_destroy(smx_simcall_t simcall, smx_rdv_t rdv){
71 return SIMIX_rdv_destroy(rdv);
73 void SIMIX_rdv_destroy(smx_rdv_t rdv)
76 xbt_dict_remove(rdv_points, rdv->name);
79 void SIMIX_rdv_free(void *data)
81 XBT_DEBUG("rdv free %p", data);
82 smx_rdv_t rdv = (smx_rdv_t) data;
84 xbt_fifo_free(rdv->comm_fifo);
85 xbt_fifo_free(rdv->done_comm_fifo);
90 xbt_dict_t SIMIX_get_rdv_points()
95 smx_rdv_t SIMIX_pre_rdv_get_by_name(smx_simcall_t simcall, const char *name){
96 return SIMIX_rdv_get_by_name(name);
98 smx_rdv_t SIMIX_rdv_get_by_name(const char *name)
100 return xbt_dict_get_or_null(rdv_points, name);
103 int SIMIX_pre_rdv_comm_count_by_host(smx_simcall_t simcall, smx_rdv_t rdv, smx_host_t host){
104 return SIMIX_rdv_comm_count_by_host(rdv, host);
106 int SIMIX_rdv_comm_count_by_host(smx_rdv_t rdv, smx_host_t host)
108 smx_action_t comm = NULL;
109 xbt_fifo_item_t item = NULL;
112 xbt_fifo_foreach(rdv->comm_fifo, item, comm, smx_action_t) {
113 if (comm->comm.src_proc->smx_host == host)
120 smx_action_t SIMIX_pre_rdv_get_head(smx_simcall_t simcall, smx_rdv_t rdv){
121 return SIMIX_rdv_get_head(rdv);
123 smx_action_t SIMIX_rdv_get_head(smx_rdv_t rdv)
125 return xbt_fifo_get_item_content(xbt_fifo_get_first_item(rdv->comm_fifo));
128 smx_process_t SIMIX_pre_rdv_get_receiver(smx_simcall_t simcall, smx_rdv_t rdv){
129 return SIMIX_rdv_get_receiver(rdv);
132 * \brief get the receiver (process associated to the mailbox)
133 * \param rdv The rendez-vous point
134 * \return process The receiving process (NULL if not set)
136 smx_process_t SIMIX_rdv_get_receiver(smx_rdv_t rdv)
138 return rdv->permanent_receiver;
141 void SIMIX_pre_rdv_set_receiver(smx_simcall_t simcall, smx_rdv_t rdv,
142 smx_process_t process){
143 SIMIX_rdv_set_receiver(rdv, process);
146 * \brief set the receiver of the rendez vous point to allow eager sends
147 * \param rdv The rendez-vous point
148 * \param process The receiving process
150 void SIMIX_rdv_set_receiver(smx_rdv_t rdv, smx_process_t process)
152 rdv->permanent_receiver=process;
156 * \brief Pushes a communication action into a rendez-vous point
157 * \param rdv The rendez-vous point
158 * \param comm The communication action
160 static XBT_INLINE void SIMIX_rdv_push(smx_rdv_t rdv, smx_action_t comm)
162 xbt_fifo_push(rdv->comm_fifo, comm);
163 comm->comm.rdv = rdv;
167 * \brief Removes a communication action from a rendez-vous point
168 * \param rdv The rendez-vous point
169 * \param comm The communication action
171 XBT_INLINE void SIMIX_rdv_remove(smx_rdv_t rdv, smx_action_t comm)
173 xbt_fifo_remove(rdv->comm_fifo, comm);
174 comm->comm.rdv = NULL;
178 * \brief Checks if there is a communication action queued in a fifo matching our needs
179 * \param type The type of communication we are looking for (comm_send, comm_recv)
180 * \return The communication action if found, NULL otherwise
182 smx_action_t SIMIX_fifo_get_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
183 int (*match_fun)(void *, void *,smx_action_t),
184 void *this_user_data, smx_action_t my_action)
187 xbt_fifo_item_t item;
188 void* other_user_data = NULL;
190 xbt_fifo_foreach(fifo, item, action, smx_action_t) {
191 if (action->comm.type == SIMIX_COMM_SEND) {
192 other_user_data = action->comm.src_data;
193 } else if (action->comm.type == SIMIX_COMM_RECEIVE) {
194 other_user_data = action->comm.dst_data;
196 if (action->comm.type == type &&
197 (!match_fun || match_fun(this_user_data, other_user_data, action)) &&
198 (!action->comm.match_fun || action->comm.match_fun(other_user_data, this_user_data, my_action))) {
199 XBT_DEBUG("Found a matching communication action %p", action);
200 xbt_fifo_remove_item(fifo, item);
201 xbt_fifo_free_item(item);
202 action->comm.refcount++;
204 action->comm.rdv_cpy = action->comm.rdv;
206 action->comm.rdv = NULL;
209 XBT_DEBUG("Sorry, communication action %p does not match our needs:"
210 " its type is %d but we are looking for a comm of type %d (or maybe the filtering didn't match)",
211 action, (int)action->comm.type, (int)type);
213 XBT_DEBUG("No matching communication action found");
219 * \brief Checks if there is a communication action queued in a fifo matching our needs, but leave it there
220 * \param type The type of communication we are looking for (comm_send, comm_recv)
221 * \return The communication action if found, NULL otherwise
223 smx_action_t SIMIX_fifo_probe_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
224 int (*match_fun)(void *, void *,smx_action_t),
225 void *this_user_data, smx_action_t my_action)
228 xbt_fifo_item_t item;
229 void* other_user_data = NULL;
231 xbt_fifo_foreach(fifo, item, action, smx_action_t) {
232 if (action->comm.type == SIMIX_COMM_SEND) {
233 other_user_data = action->comm.src_data;
234 } else if (action->comm.type == SIMIX_COMM_RECEIVE) {
235 other_user_data = action->comm.dst_data;
237 if (action->comm.type == type &&
238 (!match_fun || match_fun(this_user_data, other_user_data, action)) &&
239 (!action->comm.match_fun || action->comm.match_fun(other_user_data, this_user_data, my_action))) {
240 XBT_DEBUG("Found a matching communication action %p", action);
241 action->comm.refcount++;
245 XBT_DEBUG("Sorry, communication action %p does not match our needs:"
246 " its type is %d but we are looking for a comm of type %d (or maybe the filtering didn't match)",
247 action, (int)action->comm.type, (int)type);
249 XBT_DEBUG("No matching communication action found");
252 /******************************************************************************/
253 /* Communication Actions */
254 /******************************************************************************/
257 * \brief Creates a new communicate action
258 * \param type The direction of communication (comm_send, comm_recv)
259 * \return The new communicate action
261 smx_action_t SIMIX_comm_new(e_smx_comm_type_t type)
265 /* alloc structures */
266 act = xbt_mallocator_get(simix_global->action_mallocator);
268 act->type = SIMIX_ACTION_COMMUNICATE;
269 act->state = SIMIX_WAITING;
271 /* set communication */
272 act->comm.type = type;
273 act->comm.refcount = 1;
274 act->comm.src_data=NULL;
275 act->comm.dst_data=NULL;
278 #ifdef HAVE_LATENCY_BOUND_TRACKING
279 //initialize with unknown value
280 act->latency_limited = -1;
284 act->category = NULL;
287 XBT_DEBUG("Create communicate action %p", act);
294 * \brief Destroy a communicate action
295 * \param action The communicate action to be destroyed
297 void SIMIX_comm_destroy(smx_action_t action)
299 XBT_DEBUG("Destroy action %p (refcount: %d), state: %d",
300 action, action->comm.refcount, (int)action->state);
302 if (action->comm.refcount <= 0) {
303 xbt_backtrace_display_current();
304 xbt_die("The refcount of comm %p is already 0 before decreasing it. "
305 "That's a bug! If you didn't test and/or wait the same communication twice in your code, then the bug is SimGrid's...", action);
307 action->comm.refcount--;
308 if (action->comm.refcount > 0)
310 XBT_DEBUG("Really free communication %p; refcount is now %d", action,
311 action->comm.refcount);
313 #ifdef HAVE_LATENCY_BOUND_TRACKING
314 action->latency_limited = SIMIX_comm_is_latency_bounded( action ) ;
317 xbt_free(action->name);
318 SIMIX_comm_destroy_internal_actions(action);
320 if (action->comm.detached && action->state != SIMIX_DONE) {
321 /* the communication has failed and was detached:
322 * we have to free the buffer */
323 if (action->comm.clean_fun) {
324 action->comm.clean_fun(action->comm.src_buff);
326 action->comm.src_buff = NULL;
330 SIMIX_rdv_remove(action->comm.rdv, action);
332 xbt_mallocator_release(simix_global->action_mallocator, action);
335 void SIMIX_comm_destroy_internal_actions(smx_action_t action)
337 if (action->comm.surf_comm){
338 #ifdef HAVE_LATENCY_BOUND_TRACKING
339 action->latency_limited = SIMIX_comm_is_latency_bounded(action);
341 surf_action_unref(action->comm.surf_comm);
342 action->comm.surf_comm = NULL;
345 if (action->comm.src_timeout){
346 surf_action_unref(action->comm.src_timeout);
347 action->comm.src_timeout = NULL;
350 if (action->comm.dst_timeout){
351 surf_action_unref(action->comm.dst_timeout);
352 action->comm.dst_timeout = NULL;
356 void SIMIX_pre_comm_send(smx_simcall_t simcall, smx_process_t src, smx_rdv_t rdv,
357 double task_size, double rate,
358 void *src_buff, size_t src_buff_size,
359 int (*match_fun)(void *, void *,smx_action_t),
360 void (*copy_data_fun)(smx_action_t, void*, size_t),
361 void *data, double timeout){
362 smx_action_t comm = SIMIX_comm_isend(src, rdv, task_size, rate,
363 src_buff, src_buff_size, match_fun, NULL, copy_data_fun,
365 SIMCALL_SET_MC_VALUE(simcall, 0);
366 SIMIX_pre_comm_wait(simcall, comm, timeout);
368 smx_action_t SIMIX_pre_comm_isend(smx_simcall_t simcall, smx_process_t src, smx_rdv_t rdv,
369 double task_size, double rate,
370 void *src_buff, size_t src_buff_size,
371 int (*match_fun)(void *, void *,smx_action_t),
372 void (*clean_fun)(void *),
373 void (*copy_data_fun)(smx_action_t, void*, size_t),
374 void *data, int detached){
375 return SIMIX_comm_isend(src, rdv, task_size, rate, src_buff,
376 src_buff_size, match_fun, clean_fun, copy_data_fun, data, detached);
379 smx_action_t SIMIX_comm_isend(smx_process_t src_proc, smx_rdv_t rdv,
380 double task_size, double rate,
381 void *src_buff, size_t src_buff_size,
382 int (*match_fun)(void *, void *,smx_action_t),
383 void (*clean_fun)(void *), // used to free the action in case of problem after a detached send
384 void (*copy_data_fun)(smx_action_t, void*, size_t), // used to copy data if not default one
388 XBT_DEBUG("send from %p\n", rdv);
390 /* Prepare an action describing us, so that it gets passed to the user-provided filter of other side */
391 smx_action_t this_action = SIMIX_comm_new(SIMIX_COMM_SEND);
393 /* Look for communication action matching our needs. We also provide a description of
394 * ourself so that the other side also gets a chance of choosing if it wants to match with us.
396 * If it is not found then push our communication into the rendez-vous point */
397 smx_action_t other_action = SIMIX_fifo_get_comm(rdv->comm_fifo, SIMIX_COMM_RECEIVE, match_fun, data, this_action);
400 other_action = this_action;
402 if (rdv->permanent_receiver!=NULL){
403 //this mailbox is for small messages, which have to be sent right now
404 other_action->state = SIMIX_READY;
405 other_action->comm.dst_proc=rdv->permanent_receiver;
406 other_action->comm.refcount++;
407 xbt_fifo_push(rdv->done_comm_fifo,other_action);
408 other_action->comm.rdv=rdv;
409 XBT_DEBUG("pushing a message into the permanent receive fifo %p, comm %p \n", rdv, &(other_action->comm));
412 SIMIX_rdv_push(rdv, this_action);
415 XBT_DEBUG("Receive already pushed\n");
417 SIMIX_comm_destroy(this_action);
418 --smx_total_comms; // this creation was a pure waste
420 other_action->state = SIMIX_READY;
421 other_action->comm.type = SIMIX_COMM_READY;
424 xbt_fifo_push(src_proc->comms, other_action);
426 /* if the communication action is detached then decrease the refcount
427 * by one, so it will be eliminated by the receiver's destroy call */
429 other_action->comm.detached = 1;
430 other_action->comm.refcount--;
431 other_action->comm.clean_fun = clean_fun;
433 other_action->comm.clean_fun = NULL;
436 /* Setup the communication action */
437 other_action->comm.src_proc = src_proc;
438 other_action->comm.task_size = task_size;
439 other_action->comm.rate = rate;
440 other_action->comm.src_buff = src_buff;
441 other_action->comm.src_buff_size = src_buff_size;
442 other_action->comm.src_data = data;
444 other_action->comm.match_fun = match_fun;
445 other_action->comm.copy_data_fun = copy_data_fun;
448 if (MC_is_active()) {
449 other_action->state = SIMIX_RUNNING;
450 return (detached ? NULL : other_action);
453 SIMIX_comm_start(other_action);
454 return (detached ? NULL : other_action);
457 void SIMIX_pre_comm_recv(smx_simcall_t simcall, smx_rdv_t rdv,
458 void *dst_buff, size_t *dst_buff_size,
459 int (*match_fun)(void *, void *, smx_action_t),
460 void (*copy_data_fun)(smx_action_t, void*, size_t),
461 void *data, double timeout, double rate)
463 smx_action_t comm = SIMIX_comm_irecv(simcall->issuer, rdv, dst_buff,
464 dst_buff_size, match_fun, copy_data_fun, data, rate);
465 SIMCALL_SET_MC_VALUE(simcall, 0);
466 SIMIX_pre_comm_wait(simcall, comm, timeout);
469 smx_action_t SIMIX_pre_comm_irecv(smx_simcall_t simcall, smx_rdv_t rdv,
470 void *dst_buff, size_t *dst_buff_size,
471 int (*match_fun)(void *, void *, smx_action_t),
472 void (*copy_data_fun)(smx_action_t, void*, size_t),
473 void *data, double rate)
475 return SIMIX_comm_irecv(simcall->issuer, rdv, dst_buff, dst_buff_size,
476 match_fun, copy_data_fun, data, rate);
479 smx_action_t SIMIX_comm_irecv(smx_process_t dst_proc, smx_rdv_t rdv,
480 void *dst_buff, size_t *dst_buff_size,
481 int (*match_fun)(void *, void *, smx_action_t),
482 void (*copy_data_fun)(smx_action_t, void*, size_t), // used to copy data if not default one
483 void *data, double rate)
485 XBT_DEBUG("recv from %p %p\n", rdv, rdv->comm_fifo);
486 smx_action_t this_action = SIMIX_comm_new(SIMIX_COMM_RECEIVE);
488 smx_action_t other_action;
489 //communication already done, get it inside the fifo of completed comms
490 //permanent receive v1
491 //int already_received=0;
492 if(rdv->permanent_receiver && xbt_fifo_size(rdv->done_comm_fifo)!=0){
494 XBT_DEBUG("We have a comm that has probably already been received, trying to match it, to skip the communication\n");
495 //find a match in the already received fifo
496 other_action = SIMIX_fifo_get_comm(rdv->done_comm_fifo, SIMIX_COMM_SEND, match_fun, data, this_action);
497 //if not found, assume the receiver came first, register it to the mailbox in the classical way
499 XBT_DEBUG("We have messages in the permanent receive list, but not the one we are looking for, pushing request into fifo\n");
500 other_action = this_action;
501 SIMIX_rdv_push(rdv, this_action);
503 if(other_action->comm.surf_comm && SIMIX_comm_get_remains(other_action)==0.0)
505 XBT_DEBUG("comm %p has been already sent, and is finished, destroy it\n",&(other_action->comm));
506 other_action->state = SIMIX_DONE;
507 other_action->comm.type = SIMIX_COMM_DONE;
508 other_action->comm.rdv = NULL;
509 //SIMIX_comm_destroy(this_action);
510 //--smx_total_comms; // this creation was a pure waste
511 //already_received=1;
512 //other_action->comm.refcount--;
514 XBT_DEBUG("Not yet finished, we have to wait %d\n", xbt_fifo_size(rdv->comm_fifo));
516 other_action->comm.refcount--;
517 SIMIX_comm_destroy(this_action);
518 --smx_total_comms; // this creation was a pure waste
521 /* Prepare an action describing us, so that it gets passed to the user-provided filter of other side */
523 /* Look for communication action matching our needs. We also provide a description of
524 * ourself so that the other side also gets a chance of choosing if it wants to match with us.
526 * If it is not found then push our communication into the rendez-vous point */
527 other_action = SIMIX_fifo_get_comm(rdv->comm_fifo, SIMIX_COMM_SEND, match_fun, data, this_action);
530 XBT_DEBUG("Receive pushed first %d\n", xbt_fifo_size(rdv->comm_fifo));
531 other_action = this_action;
532 SIMIX_rdv_push(rdv, this_action);
534 SIMIX_comm_destroy(this_action);
535 --smx_total_comms; // this creation was a pure waste
536 other_action->state = SIMIX_READY;
537 other_action->comm.type = SIMIX_COMM_READY;
538 //other_action->comm.refcount--;
540 xbt_fifo_push(dst_proc->comms, other_action);
543 /* Setup communication action */
544 other_action->comm.dst_proc = dst_proc;
545 other_action->comm.dst_buff = dst_buff;
546 other_action->comm.dst_buff_size = dst_buff_size;
547 other_action->comm.dst_data = data;
550 (other_action->comm.rate == -1.0 || rate < other_action->comm.rate))
551 other_action->comm.rate = rate;
553 other_action->comm.match_fun = match_fun;
554 other_action->comm.copy_data_fun = copy_data_fun;
557 /*if(already_received)//do the actual copy, because the first one after the comm didn't have all the info
558 SIMIX_comm_copy_data(other_action);*/
561 if (MC_is_active()) {
562 other_action->state = SIMIX_RUNNING;
566 SIMIX_comm_start(other_action);
571 smx_action_t SIMIX_pre_comm_iprobe(smx_simcall_t simcall, smx_rdv_t rdv,
573 int (*match_fun)(void *, void *, smx_action_t),
575 return SIMIX_comm_iprobe(simcall->issuer, rdv, src, tag, match_fun, data);
578 smx_action_t SIMIX_comm_iprobe(smx_process_t dst_proc, smx_rdv_t rdv, int src,
579 int tag, int (*match_fun)(void *, void *, smx_action_t), void *data)
581 XBT_DEBUG("iprobe from %p %p\n", rdv, rdv->comm_fifo);
582 smx_action_t this_action = SIMIX_comm_new(SIMIX_COMM_RECEIVE);
584 smx_action_t other_action=NULL;
585 if(rdv->permanent_receiver && xbt_fifo_size(rdv->done_comm_fifo)!=0){
586 //find a match in the already received fifo
587 XBT_DEBUG("first try in the perm recv mailbox \n");
589 other_action = SIMIX_fifo_probe_comm(rdv->done_comm_fifo, SIMIX_COMM_SEND, match_fun, data, this_action);
593 XBT_DEBUG("second try in the other mailbox");
594 other_action = SIMIX_fifo_probe_comm(rdv->comm_fifo, SIMIX_COMM_SEND, match_fun, data, this_action);
597 if(other_action)other_action->comm.refcount--;
599 SIMIX_comm_destroy(this_action);
604 void SIMIX_pre_comm_wait(smx_simcall_t simcall, smx_action_t action, double timeout)
606 /* the simcall may be a wait, a send or a recv */
609 /* Associate this simcall to the wait action */
610 XBT_DEBUG("SIMIX_pre_comm_wait, %p", action);
612 xbt_fifo_push(action->simcalls, simcall);
613 simcall->issuer->waiting_action = action;
615 if (MC_is_active()) {
616 int idx = SIMCALL_GET_MC_VALUE(simcall);
618 action->state = SIMIX_DONE;
620 /* If we reached this point, the wait simcall must have a timeout */
621 /* Otherwise it shouldn't be enabled and executed by the MC */
625 if (action->comm.src_proc == simcall->issuer)
626 action->state = SIMIX_SRC_TIMEOUT;
628 action->state = SIMIX_DST_TIMEOUT;
631 SIMIX_comm_finish(action);
635 /* If the action has already finish perform the error handling, */
636 /* otherwise set up a waiting timeout on the right side */
637 if (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING) {
638 SIMIX_comm_finish(action);
639 } else { /* if (timeout >= 0) { we need a surf sleep action even when there is no timeout, otherwise surf won't tell us when the host fails */
640 sleep = surf_workstation_sleep(simcall->issuer->smx_host, timeout);
641 surf_action_set_data(sleep, action);
643 if (simcall->issuer == action->comm.src_proc)
644 action->comm.src_timeout = sleep;
646 action->comm.dst_timeout = sleep;
650 void SIMIX_pre_comm_test(smx_simcall_t simcall, smx_action_t action)
653 simcall_comm_test__set__result(simcall, action->comm.src_proc && action->comm.dst_proc);
654 if(simcall_comm_test__get__result(simcall)){
655 action->state = SIMIX_DONE;
656 xbt_fifo_push(action->simcalls, simcall);
657 SIMIX_comm_finish(action);
659 SIMIX_simcall_answer(simcall);
664 simcall_comm_test__set__result(simcall, (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING));
665 if (simcall_comm_test__get__result(simcall)) {
666 xbt_fifo_push(action->simcalls, simcall);
667 SIMIX_comm_finish(action);
669 SIMIX_simcall_answer(simcall);
673 void SIMIX_pre_comm_testany(smx_simcall_t simcall, xbt_dynar_t actions)
677 simcall_comm_testany__set__result(simcall, -1);
680 int idx = SIMCALL_GET_MC_VALUE(simcall);
682 SIMIX_simcall_answer(simcall);
684 action = xbt_dynar_get_as(actions, idx, smx_action_t);
685 simcall_comm_testany__set__result(simcall, idx);
686 xbt_fifo_push(action->simcalls, simcall);
687 action->state = SIMIX_DONE;
688 SIMIX_comm_finish(action);
693 xbt_dynar_foreach(simcall_comm_testany__get__comms(simcall), cursor,action) {
694 if (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING) {
695 simcall_comm_testany__set__result(simcall, cursor);
696 xbt_fifo_push(action->simcalls, simcall);
697 SIMIX_comm_finish(action);
701 SIMIX_simcall_answer(simcall);
704 void SIMIX_pre_comm_waitany(smx_simcall_t simcall, xbt_dynar_t actions)
707 unsigned int cursor = 0;
710 int idx = SIMCALL_GET_MC_VALUE(simcall);
711 action = xbt_dynar_get_as(actions, idx, smx_action_t);
712 xbt_fifo_push(action->simcalls, simcall);
713 simcall_comm_waitany__set__result(simcall, idx);
714 action->state = SIMIX_DONE;
715 SIMIX_comm_finish(action);
719 xbt_dynar_foreach(actions, cursor, action){
720 /* associate this simcall to the the action */
721 xbt_fifo_push(action->simcalls, simcall);
723 /* see if the action is already finished */
724 if (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING){
725 SIMIX_comm_finish(action);
731 void SIMIX_waitany_remove_simcall_from_actions(smx_simcall_t simcall)
734 unsigned int cursor = 0;
735 xbt_dynar_t actions = simcall_comm_waitany__get__comms(simcall);
737 xbt_dynar_foreach(actions, cursor, action) {
738 xbt_fifo_remove(action->simcalls, simcall);
743 * \brief Starts the simulation of a communication action.
744 * \param action the communication action
746 static XBT_INLINE void SIMIX_comm_start(smx_action_t action)
748 /* If both the sender and the receiver are already there, start the communication */
749 if (action->state == SIMIX_READY) {
751 smx_host_t sender = action->comm.src_proc->smx_host;
752 smx_host_t receiver = action->comm.dst_proc->smx_host;
754 XBT_DEBUG("Starting communication %p from '%s' to '%s'", action,
755 SIMIX_host_get_name(sender), SIMIX_host_get_name(receiver));
757 action->comm.surf_comm = surf_workstation_model_communicate(surf_workstation_model,
759 action->comm.task_size, action->comm.rate);
761 surf_action_set_data(action->comm.surf_comm, action);
763 action->state = SIMIX_RUNNING;
765 /* If a link is failed, detect it immediately */
766 if (surf_action_get_state(action->comm.surf_comm) == SURF_ACTION_FAILED) {
767 XBT_DEBUG("Communication from '%s' to '%s' failed to start because of a link failure",
768 SIMIX_host_get_name(sender), SIMIX_host_get_name(receiver));
769 action->state = SIMIX_LINK_FAILURE;
770 SIMIX_comm_destroy_internal_actions(action);
773 /* If any of the process is suspend, create the action but stop its execution,
774 it will be restarted when the sender process resume */
775 if (SIMIX_process_is_suspended(action->comm.src_proc) ||
776 SIMIX_process_is_suspended(action->comm.dst_proc)) {
777 /* FIXME: check what should happen with the action state */
779 if (SIMIX_process_is_suspended(action->comm.src_proc))
780 XBT_DEBUG("The communication is suspended on startup because src (%s:%s) were suspended since it initiated the communication",
781 SIMIX_host_get_name(action->comm.src_proc->smx_host), action->comm.src_proc->name);
783 XBT_DEBUG("The communication is suspended on startup because dst (%s:%s) were suspended since it initiated the communication",
784 SIMIX_host_get_name(action->comm.dst_proc->smx_host), action->comm.dst_proc->name);
786 surf_action_suspend(action->comm.surf_comm);
793 * \brief Answers the SIMIX simcalls associated to a communication action.
794 * \param action a finished communication action
796 void SIMIX_comm_finish(smx_action_t action)
798 unsigned int destroy_count = 0;
799 smx_simcall_t simcall;
802 while ((simcall = xbt_fifo_shift(action->simcalls))) {
804 /* If a waitany simcall is waiting for this action to finish, then remove
805 it from the other actions in the waitany list. Afterwards, get the
806 position of the actual action in the waitany dynar and
807 return it as the result of the simcall */
808 if (simcall->call == SIMCALL_COMM_WAITANY) {
809 SIMIX_waitany_remove_simcall_from_actions(simcall);
811 simcall_comm_waitany__set__result(simcall, xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &action));
814 /* If the action is still in a rendez-vous point then remove from it */
815 if (action->comm.rdv)
816 SIMIX_rdv_remove(action->comm.rdv, action);
818 XBT_DEBUG("SIMIX_comm_finish: action state = %d", (int)action->state);
820 /* Check out for errors */
821 switch (action->state) {
824 XBT_DEBUG("Communication %p complete!", action);
825 SIMIX_comm_copy_data(action);
828 case SIMIX_SRC_TIMEOUT:
829 SMX_EXCEPTION(simcall->issuer, timeout_error, 0,
830 "Communication timeouted because of sender");
833 case SIMIX_DST_TIMEOUT:
834 SMX_EXCEPTION(simcall->issuer, timeout_error, 0,
835 "Communication timeouted because of receiver");
838 case SIMIX_SRC_HOST_FAILURE:
839 if (simcall->issuer == action->comm.src_proc)
840 simcall->issuer->context->iwannadie = 1;
841 // SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
843 SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
846 case SIMIX_DST_HOST_FAILURE:
847 if (simcall->issuer == action->comm.dst_proc)
848 simcall->issuer->context->iwannadie = 1;
849 // SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
851 SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
854 case SIMIX_LINK_FAILURE:
855 XBT_DEBUG("Link failure in action %p between '%s' and '%s': posting an exception to the issuer: %s (%p) detached:%d",
857 action->comm.src_proc ? sg_host_name(action->comm.src_proc->smx_host) : NULL,
858 action->comm.dst_proc ? sg_host_name(action->comm.dst_proc->smx_host) : NULL,
859 simcall->issuer->name, simcall->issuer, action->comm.detached);
860 if (action->comm.src_proc == simcall->issuer) {
861 XBT_DEBUG("I'm source");
862 } else if (action->comm.dst_proc == simcall->issuer) {
863 XBT_DEBUG("I'm dest");
865 XBT_DEBUG("I'm neither source nor dest");
867 SMX_EXCEPTION(simcall->issuer, network_error, 0, "Link failure");
871 if (simcall->issuer == action->comm.dst_proc)
872 SMX_EXCEPTION(simcall->issuer, cancel_error, 0,
873 "Communication canceled by the sender");
875 SMX_EXCEPTION(simcall->issuer, cancel_error, 0,
876 "Communication canceled by the receiver");
880 xbt_die("Unexpected action state in SIMIX_comm_finish: %d", (int)action->state);
883 /* if there is an exception during a waitany or a testany, indicate the position of the failed communication */
884 if (simcall->issuer->doexception) {
885 if (simcall->call == SIMCALL_COMM_WAITANY) {
886 simcall->issuer->running_ctx->exception.value = xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &action);
888 else if (simcall->call == SIMCALL_COMM_TESTANY) {
889 simcall->issuer->running_ctx->exception.value = xbt_dynar_search(simcall_comm_testany__get__comms(simcall), &action);
893 if (surf_resource_get_state(surf_workstation_resource_priv(simcall->issuer->smx_host)) != SURF_RESOURCE_ON) {
894 simcall->issuer->context->iwannadie = 1;
897 simcall->issuer->waiting_action = NULL;
898 xbt_fifo_remove(simcall->issuer->comms, action);
899 if(action->comm.detached){
900 if(simcall->issuer == action->comm.src_proc){
901 if(action->comm.dst_proc)
902 xbt_fifo_remove(action->comm.dst_proc->comms, action);
904 if(simcall->issuer == action->comm.dst_proc){
905 if(action->comm.src_proc)
906 xbt_fifo_remove(action->comm.src_proc->comms, action);
909 SIMIX_simcall_answer(simcall);
913 while (destroy_count-- > 0)
914 SIMIX_comm_destroy(action);
918 * \brief This function is called when a Surf communication action is finished.
919 * \param action the corresponding Simix communication
921 void SIMIX_post_comm(smx_action_t action)
923 /* Update action state */
924 if (action->comm.src_timeout &&
925 surf_action_get_state(action->comm.src_timeout) == SURF_ACTION_DONE)
926 action->state = SIMIX_SRC_TIMEOUT;
927 else if (action->comm.dst_timeout &&
928 surf_action_get_state(action->comm.dst_timeout) == SURF_ACTION_DONE)
929 action->state = SIMIX_DST_TIMEOUT;
930 else if (action->comm.src_timeout &&
931 surf_action_get_state(action->comm.src_timeout) == SURF_ACTION_FAILED)
932 action->state = SIMIX_SRC_HOST_FAILURE;
933 else if (action->comm.dst_timeout &&
934 surf_action_get_state(action->comm.dst_timeout) == SURF_ACTION_FAILED)
935 action->state = SIMIX_DST_HOST_FAILURE;
936 else if (action->comm.surf_comm &&
937 surf_action_get_state(action->comm.surf_comm) == SURF_ACTION_FAILED) {
938 XBT_DEBUG("Puta madre. Surf says that the link broke");
939 action->state = SIMIX_LINK_FAILURE;
941 action->state = SIMIX_DONE;
943 XBT_DEBUG("SIMIX_post_comm: comm %p, state %d, src_proc %p, dst_proc %p, detached: %d",
944 action, (int)action->state, action->comm.src_proc, action->comm.dst_proc, action->comm.detached);
946 /* destroy the surf actions associated with the Simix communication */
947 SIMIX_comm_destroy_internal_actions(action);
949 /* remove the communication action from the list of pending communications
950 * of both processes (if they still exist) */
951 if (action->comm.src_proc) {
952 xbt_fifo_remove(action->comm.src_proc->comms, action);
954 if (action->comm.dst_proc) {
955 xbt_fifo_remove(action->comm.dst_proc->comms, action);
958 /* if there are simcalls associated with the action, then answer them */
959 if (xbt_fifo_size(action->simcalls)) {
960 SIMIX_comm_finish(action);
964 void SIMIX_pre_comm_cancel(smx_simcall_t simcall, smx_action_t action){
965 SIMIX_comm_cancel(action);
967 void SIMIX_comm_cancel(smx_action_t action)
969 /* if the action is a waiting state means that it is still in a rdv */
970 /* so remove from it and delete it */
971 if (action->state == SIMIX_WAITING) {
972 SIMIX_rdv_remove(action->comm.rdv, action);
973 action->state = SIMIX_CANCELED;
975 else if (!MC_is_active() /* when running the MC there are no surf actions */
976 && (action->state == SIMIX_READY || action->state == SIMIX_RUNNING)) {
978 surf_action_cancel(action->comm.surf_comm);
982 void SIMIX_comm_suspend(smx_action_t action)
984 /*FIXME: shall we suspend also the timeout actions? */
985 if (action->comm.surf_comm)
986 surf_action_suspend(action->comm.surf_comm);
987 /* in the other case, the action will be suspended on creation, in SIMIX_comm_start() */
990 void SIMIX_comm_resume(smx_action_t action)
992 /*FIXME: check what happen with the timeouts */
993 if (action->comm.surf_comm)
994 surf_action_resume(action->comm.surf_comm);
995 /* in the other case, the action were not really suspended yet, see SIMIX_comm_suspend() and SIMIX_comm_start() */
999 /************* Action Getters **************/
1001 double SIMIX_pre_comm_get_remains(smx_simcall_t simcall, smx_action_t action){
1002 return SIMIX_comm_get_remains(action);
1005 * \brief get the amount remaining from the communication
1006 * \param action The communication
1008 double SIMIX_comm_get_remains(smx_action_t action)
1016 switch (action->state) {
1019 remains = surf_action_get_remains(action->comm.surf_comm);
1024 remains = 0; /*FIXME: check what should be returned */
1028 remains = 0; /*FIXME: is this correct? */
1034 e_smx_state_t SIMIX_pre_comm_get_state(smx_simcall_t simcall, smx_action_t action){
1035 return SIMIX_comm_get_state(action);
1037 e_smx_state_t SIMIX_comm_get_state(smx_action_t action)
1039 return action->state;
1042 void* SIMIX_pre_comm_get_src_data(smx_simcall_t simcall, smx_action_t action){
1043 return SIMIX_comm_get_src_data(action);
1046 * \brief Return the user data associated to the sender of the communication
1047 * \param action The communication
1048 * \return the user data
1050 void* SIMIX_comm_get_src_data(smx_action_t action)
1052 return action->comm.src_data;
1055 void* SIMIX_pre_comm_get_dst_data(smx_simcall_t simcall, smx_action_t action){
1056 return SIMIX_comm_get_dst_data(action);
1059 * \brief Return the user data associated to the receiver of the communication
1060 * \param action The communication
1061 * \return the user data
1063 void* SIMIX_comm_get_dst_data(smx_action_t action)
1065 return action->comm.dst_data;
1068 smx_process_t SIMIX_pre_comm_get_src_proc(smx_simcall_t simcall, smx_action_t action){
1069 return SIMIX_comm_get_src_proc(action);
1071 smx_process_t SIMIX_comm_get_src_proc(smx_action_t action)
1073 return action->comm.src_proc;
1076 smx_process_t SIMIX_pre_comm_get_dst_proc(smx_simcall_t simcall, smx_action_t action){
1077 return SIMIX_comm_get_dst_proc(action);
1079 smx_process_t SIMIX_comm_get_dst_proc(smx_action_t action)
1081 return action->comm.dst_proc;
1084 #ifdef HAVE_LATENCY_BOUND_TRACKING
1085 int SIMIX_pre_comm_is_latency_bounded(smx_simcall_t simcall, smx_action_t action)
1087 return SIMIX_comm_is_latency_bounded(action);
1091 * \brief verify if communication is latency bounded
1092 * \param comm The communication
1094 int SIMIX_comm_is_latency_bounded(smx_action_t action)
1099 if (action->comm.surf_comm){
1100 XBT_DEBUG("Getting latency limited for surf_action (%p)", action->comm.surf_comm);
1101 action->latency_limited = surf_network_action_get_latency_limited(action->comm.surf_comm);
1102 XBT_DEBUG("Action limited is %d", action->latency_limited);
1104 return action->latency_limited;
1108 /******************************************************************************/
1109 /* SIMIX_comm_copy_data callbacks */
1110 /******************************************************************************/
1111 static void (*SIMIX_comm_copy_data_callback) (smx_action_t, void*, size_t) =
1112 &SIMIX_comm_copy_pointer_callback;
1115 SIMIX_comm_set_copy_data_callback(void (*callback) (smx_action_t, void*, size_t))
1117 SIMIX_comm_copy_data_callback = callback;
1120 void SIMIX_comm_copy_pointer_callback(smx_action_t comm, void* buff, size_t buff_size)
1122 xbt_assert((buff_size == sizeof(void *)),
1123 "Cannot copy %zu bytes: must be sizeof(void*)", buff_size);
1124 *(void **) (comm->comm.dst_buff) = buff;
1127 void SIMIX_comm_copy_buffer_callback(smx_action_t comm, void* buff, size_t buff_size)
1129 XBT_DEBUG("Copy the data over");
1130 memcpy(comm->comm.dst_buff, buff, buff_size);
1131 if (comm->comm.detached) { // if this is a detached send, the source buffer was duplicated by SMPI sender to make the original buffer available to the application ASAP
1133 comm->comm.src_buff = NULL;
1139 * \brief Copy the communication data from the sender's buffer to the receiver's one
1140 * \param comm The communication
1142 void SIMIX_comm_copy_data(smx_action_t comm)
1144 size_t buff_size = comm->comm.src_buff_size;
1145 /* If there is no data to be copy then return */
1146 if (!comm->comm.src_buff || !comm->comm.dst_buff || comm->comm.copied)
1149 XBT_DEBUG("Copying comm %p data from %s (%p) -> %s (%p) (%zu bytes)",
1151 comm->comm.src_proc ? sg_host_name(comm->comm.src_proc->smx_host) : "a finished process",
1152 comm->comm.src_buff,
1153 comm->comm.dst_proc ? sg_host_name(comm->comm.dst_proc->smx_host) : "a finished process",
1154 comm->comm.dst_buff, buff_size);
1156 /* Copy at most dst_buff_size bytes of the message to receiver's buffer */
1157 if (comm->comm.dst_buff_size)
1158 buff_size = MIN(buff_size, *(comm->comm.dst_buff_size));
1160 /* Update the receiver's buffer size to the copied amount */
1161 if (comm->comm.dst_buff_size)
1162 *comm->comm.dst_buff_size = buff_size;
1165 if(comm->comm.copy_data_fun)
1166 comm->comm.copy_data_fun (comm, comm->comm.src_buff, buff_size);
1168 SIMIX_comm_copy_data_callback (comm, comm->comm.src_buff, buff_size);
1172 /* Set the copied flag so we copy data only once */
1173 /* (this function might be called from both communication ends) */
1174 comm->comm.copied = 1;