1 /* Copyright (c) 2009, 2010. 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"
12 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_network, simix,
13 "Logging specific to SIMIX (network)");
15 static xbt_dict_t rdv_points = NULL;
16 XBT_IMPORT_NO_EXPORT(unsigned long int) smx_total_comms = 0;
18 static void SIMIX_waitany_remove_simcall_from_actions(smx_simcall_t simcall);
19 static void SIMIX_comm_copy_data(smx_action_t comm);
20 static smx_action_t SIMIX_comm_new(e_smx_comm_type_t type);
21 static XBT_INLINE void SIMIX_rdv_push(smx_rdv_t rdv, smx_action_t comm);
22 static smx_action_t SIMIX_fifo_probe_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
23 int (*match_fun)(void *, void *,smx_action_t),
24 void *user_data, smx_action_t my_action);
25 static smx_action_t SIMIX_fifo_get_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
26 int (*match_fun)(void *, void *,smx_action_t),
27 void *user_data, smx_action_t my_action);
28 static void SIMIX_rdv_free(void *data);
30 void SIMIX_network_init(void)
32 rdv_points = xbt_dict_new_homogeneous(SIMIX_rdv_free);
34 MC_ignore_data_bss(&smx_total_comms, sizeof(smx_total_comms));
37 void SIMIX_network_exit(void)
39 xbt_dict_free(&rdv_points);
42 /******************************************************************************/
43 /* Rendez-Vous Points */
44 /******************************************************************************/
46 smx_rdv_t SIMIX_pre_rdv_create(smx_simcall_t simcall, const char *name){
47 return SIMIX_rdv_create(name);
49 smx_rdv_t SIMIX_rdv_create(const char *name)
51 /* two processes may have pushed the same rdv_create simcall at the same time */
52 smx_rdv_t rdv = name ? xbt_dict_get_or_null(rdv_points, name) : NULL;
55 rdv = xbt_new0(s_smx_rvpoint_t, 1);
56 rdv->name = name ? xbt_strdup(name) : NULL;
57 rdv->comm_fifo = xbt_fifo_new();
58 rdv->done_comm_fifo = xbt_fifo_new();
59 rdv->permanent_receiver=NULL;
61 XBT_DEBUG("Creating a mailbox at %p with name %s\n", rdv, name);
64 xbt_dict_set(rdv_points, rdv->name, rdv, NULL);
69 void SIMIX_pre_rdv_destroy(smx_simcall_t simcall, smx_rdv_t rdv){
70 return SIMIX_rdv_destroy(rdv);
72 void SIMIX_rdv_destroy(smx_rdv_t rdv)
75 xbt_dict_remove(rdv_points, rdv->name);
78 void SIMIX_rdv_free(void *data)
80 XBT_DEBUG("rdv free %p", data);
81 smx_rdv_t rdv = (smx_rdv_t) data;
83 xbt_fifo_free(rdv->comm_fifo);
84 xbt_fifo_free(rdv->done_comm_fifo);
89 xbt_dict_t SIMIX_get_rdv_points()
94 smx_rdv_t SIMIX_pre_rdv_get_by_name(smx_simcall_t simcall, const char *name){
95 return SIMIX_rdv_get_by_name(name);
97 smx_rdv_t SIMIX_rdv_get_by_name(const char *name)
99 return xbt_dict_get_or_null(rdv_points, name);
102 int SIMIX_pre_rdv_comm_count_by_host(smx_simcall_t simcall, smx_rdv_t rdv, smx_host_t host){
103 return SIMIX_rdv_comm_count_by_host(rdv, host);
105 int SIMIX_rdv_comm_count_by_host(smx_rdv_t rdv, smx_host_t host)
107 smx_action_t comm = NULL;
108 xbt_fifo_item_t item = NULL;
111 xbt_fifo_foreach(rdv->comm_fifo, item, comm, smx_action_t) {
112 if (comm->comm.src_proc->smx_host == host)
119 smx_action_t SIMIX_pre_rdv_get_head(smx_simcall_t simcall, smx_rdv_t rdv){
120 return SIMIX_rdv_get_head(rdv);
122 smx_action_t SIMIX_rdv_get_head(smx_rdv_t rdv)
124 return xbt_fifo_get_item_content(xbt_fifo_get_first_item(rdv->comm_fifo));
127 smx_process_t SIMIX_pre_rdv_get_receiver(smx_simcall_t simcall, smx_rdv_t rdv){
128 return SIMIX_rdv_get_receiver(rdv);
131 * \brief get the receiver (process associated to the mailbox)
132 * \param rdv The rendez-vous point
133 * \return process The receiving process (NULL if not set)
135 smx_process_t SIMIX_rdv_get_receiver(smx_rdv_t rdv)
137 return rdv->permanent_receiver;
140 void SIMIX_pre_rdv_set_receiver(smx_simcall_t simcall, smx_rdv_t rdv,
141 smx_process_t process){
142 SIMIX_rdv_set_receiver(rdv, process);
145 * \brief set the receiver of the rendez vous point to allow eager sends
146 * \param rdv The rendez-vous point
147 * \param process The receiving process
149 void SIMIX_rdv_set_receiver(smx_rdv_t rdv, smx_process_t process)
151 rdv->permanent_receiver=process;
155 * \brief Pushes a communication action into a rendez-vous point
156 * \param rdv The rendez-vous point
157 * \param comm The communication action
159 static XBT_INLINE void SIMIX_rdv_push(smx_rdv_t rdv, smx_action_t comm)
161 xbt_fifo_push(rdv->comm_fifo, comm);
162 comm->comm.rdv = rdv;
166 * \brief Removes a communication action from a rendez-vous point
167 * \param rdv The rendez-vous point
168 * \param comm The communication action
170 XBT_INLINE void SIMIX_rdv_remove(smx_rdv_t rdv, smx_action_t comm)
172 xbt_fifo_remove(rdv->comm_fifo, comm);
173 comm->comm.rdv = NULL;
177 * \brief Checks if there is a communication action queued in a fifo matching our needs
178 * \param type The type of communication we are looking for (comm_send, comm_recv)
179 * \return The communication action if found, NULL otherwise
181 smx_action_t SIMIX_fifo_get_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
182 int (*match_fun)(void *, void *,smx_action_t),
183 void *this_user_data, smx_action_t my_action)
186 xbt_fifo_item_t item;
187 void* other_user_data = NULL;
189 xbt_fifo_foreach(fifo, item, action, smx_action_t) {
190 if (action->comm.type == SIMIX_COMM_SEND) {
191 other_user_data = action->comm.src_data;
192 } else if (action->comm.type == SIMIX_COMM_RECEIVE) {
193 other_user_data = action->comm.dst_data;
195 if (action->comm.type == type &&
196 (!match_fun || match_fun(this_user_data, other_user_data, action)) &&
197 (!action->comm.match_fun || action->comm.match_fun(other_user_data, this_user_data, my_action))) {
198 XBT_DEBUG("Found a matching communication action %p", action);
199 xbt_fifo_remove_item(fifo, item);
200 xbt_fifo_free_item(item);
201 action->comm.refcount++;
202 action->comm.rdv = NULL;
205 XBT_DEBUG("Sorry, communication action %p does not match our needs:"
206 " its type is %d but we are looking for a comm of type %d (or maybe the filtering didn't match)",
207 action, (int)action->comm.type, (int)type);
209 XBT_DEBUG("No matching communication action found");
215 * \brief Checks if there is a communication action queued in a fifo matching our needs, but leave it there
216 * \param type The type of communication we are looking for (comm_send, comm_recv)
217 * \return The communication action if found, NULL otherwise
219 smx_action_t SIMIX_fifo_probe_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
220 int (*match_fun)(void *, void *,smx_action_t),
221 void *this_user_data, smx_action_t my_action)
224 xbt_fifo_item_t item;
225 void* other_user_data = NULL;
227 xbt_fifo_foreach(fifo, item, action, smx_action_t) {
228 if (action->comm.type == SIMIX_COMM_SEND) {
229 other_user_data = action->comm.src_data;
230 } else if (action->comm.type == SIMIX_COMM_RECEIVE) {
231 other_user_data = action->comm.dst_data;
233 if (action->comm.type == type &&
234 (!match_fun || match_fun(this_user_data, other_user_data, action)) &&
235 (!action->comm.match_fun || action->comm.match_fun(other_user_data, this_user_data, my_action))) {
236 XBT_DEBUG("Found a matching communication action %p", action);
237 action->comm.refcount++;
241 XBT_DEBUG("Sorry, communication action %p does not match our needs:"
242 " its type is %d but we are looking for a comm of type %d (or maybe the filtering didn't match)",
243 action, (int)action->comm.type, (int)type);
245 XBT_DEBUG("No matching communication action found");
248 /******************************************************************************/
249 /* Communication Actions */
250 /******************************************************************************/
253 * \brief Creates a new communicate action
254 * \param type The direction of communication (comm_send, comm_recv)
255 * \return The new communicate action
257 smx_action_t SIMIX_comm_new(e_smx_comm_type_t type)
261 /* alloc structures */
262 act = xbt_mallocator_get(simix_global->action_mallocator);
264 act->type = SIMIX_ACTION_COMMUNICATE;
265 act->state = SIMIX_WAITING;
267 /* set communication */
268 act->comm.type = type;
269 act->comm.refcount = 1;
270 act->comm.src_data=NULL;
271 act->comm.dst_data=NULL;
274 #ifdef HAVE_LATENCY_BOUND_TRACKING
275 //initialize with unknown value
276 act->latency_limited = -1;
280 act->category = NULL;
283 XBT_DEBUG("Create communicate action %p", act);
289 void SIMIX_pre_comm_destroy(smx_simcall_t simcall, smx_action_t action){
290 SIMIX_comm_destroy(action);
293 * \brief Destroy a communicate action
294 * \param action The communicate action to be destroyed
296 void SIMIX_comm_destroy(smx_action_t action)
298 XBT_DEBUG("Destroy action %p (refcount: %d), state: %d",
299 action, action->comm.refcount, (int)action->state);
301 if (action->comm.refcount <= 0) {
302 xbt_backtrace_display_current();
303 xbt_die("The refcount of comm %p is already 0 before decreasing it. "
304 "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);
306 action->comm.refcount--;
307 if (action->comm.refcount > 0)
309 XBT_DEBUG("Really free communication %p; refcount is now %d", action,
310 action->comm.refcount);
312 #ifdef HAVE_LATENCY_BOUND_TRACKING
313 action->latency_limited = SIMIX_comm_is_latency_bounded( action ) ;
316 xbt_free(action->name);
317 SIMIX_comm_destroy_internal_actions(action);
319 if (action->comm.detached && action->state != SIMIX_DONE) {
320 /* the communication has failed and was detached:
321 * we have to free the buffer */
322 if (action->comm.clean_fun) {
323 action->comm.clean_fun(action->comm.src_buff);
325 action->comm.src_buff = NULL;
328 xbt_mallocator_release(simix_global->action_mallocator, action);
331 void SIMIX_comm_destroy_internal_actions(smx_action_t action)
333 if (action->comm.surf_comm){
334 #ifdef HAVE_LATENCY_BOUND_TRACKING
335 action->latency_limited = SIMIX_comm_is_latency_bounded(action);
337 action->comm.surf_comm->model_type->action_unref(action->comm.surf_comm);
338 action->comm.surf_comm = NULL;
341 if (action->comm.src_timeout){
342 action->comm.src_timeout->model_type->action_unref(action->comm.src_timeout);
343 action->comm.src_timeout = NULL;
346 if (action->comm.dst_timeout){
347 action->comm.dst_timeout->model_type->action_unref(action->comm.dst_timeout);
348 action->comm.dst_timeout = NULL;
352 void SIMIX_pre_comm_send(smx_simcall_t simcall, smx_rdv_t rdv,
353 double task_size, double rate,
354 void *src_buff, size_t src_buff_size,
355 int (*match_fun)(void *, void *,smx_action_t),
356 void *data, double timeout){
357 smx_action_t comm = SIMIX_comm_isend(simcall->issuer, rdv, task_size, rate,
358 src_buff, src_buff_size, match_fun, NULL,
360 simcall->mc_value = 0;
361 SIMIX_pre_comm_wait(simcall, comm, timeout);
363 smx_action_t SIMIX_pre_comm_isend(smx_simcall_t simcall, smx_rdv_t rdv,
364 double task_size, double rate,
365 void *src_buff, size_t src_buff_size,
366 int (*match_fun)(void *, void *,smx_action_t),
367 void (*clean_fun)(void *),
368 void *data, int detached){
369 return SIMIX_comm_isend(simcall->issuer, rdv, task_size, rate, src_buff,
370 src_buff_size, match_fun, clean_fun, data, detached);
373 smx_action_t SIMIX_comm_isend(smx_process_t src_proc, smx_rdv_t rdv,
374 double task_size, double rate,
375 void *src_buff, size_t src_buff_size,
376 int (*match_fun)(void *, void *,smx_action_t),
377 void (*clean_fun)(void *), // used to free the action in case of problem after a detached send
381 XBT_DEBUG("send from %p\n", rdv);
383 /* Prepare an action describing us, so that it gets passed to the user-provided filter of other side */
384 smx_action_t this_action = SIMIX_comm_new(SIMIX_COMM_SEND);
386 /* Look for communication action matching our needs. We also provide a description of
387 * ourself so that the other side also gets a chance of choosing if it wants to match with us.
389 * If it is not found then push our communication into the rendez-vous point */
390 smx_action_t other_action = SIMIX_fifo_get_comm(rdv->comm_fifo, SIMIX_COMM_RECEIVE, match_fun, data, this_action);
393 other_action = this_action;
395 if (rdv->permanent_receiver!=NULL){
396 //this mailbox is for small messages, which have to be sent right now
397 other_action->state = SIMIX_READY;
398 other_action->comm.dst_proc=rdv->permanent_receiver;
399 other_action->comm.refcount++;
400 other_action->comm.rdv = rdv;
401 xbt_fifo_push(rdv->done_comm_fifo,other_action);
402 other_action->comm.rdv=rdv;
403 XBT_DEBUG("pushing a message into the permanent receive fifo %p, comm %p \n", rdv, &(other_action->comm));
406 SIMIX_rdv_push(rdv, this_action);
409 XBT_DEBUG("Receive already pushed\n");
411 SIMIX_comm_destroy(this_action);
412 --smx_total_comms; // this creation was a pure waste
414 other_action->state = SIMIX_READY;
415 other_action->comm.type = SIMIX_COMM_READY;
418 xbt_fifo_push(src_proc->comms, other_action);
420 /* if the communication action is detached then decrease the refcount
421 * by one, so it will be eliminated by the receiver's destroy call */
423 other_action->comm.detached = 1;
424 other_action->comm.refcount--;
425 other_action->comm.clean_fun = clean_fun;
427 other_action->comm.clean_fun = NULL;
430 /* Setup the communication action */
431 other_action->comm.src_proc = src_proc;
432 other_action->comm.task_size = task_size;
433 other_action->comm.rate = rate;
434 other_action->comm.src_buff = src_buff;
435 other_action->comm.src_buff_size = src_buff_size;
436 other_action->comm.src_data = data;
438 other_action->comm.match_fun = match_fun;
440 if (MC_is_active()) {
441 other_action->state = SIMIX_RUNNING;
445 SIMIX_comm_start(other_action);
446 return (detached ? NULL : other_action);
449 void SIMIX_pre_comm_recv(smx_simcall_t simcall, smx_rdv_t rdv,
450 void *dst_buff, size_t *dst_buff_size,
451 int (*match_fun)(void *, void *, smx_action_t),
452 void *data, double timeout){
453 smx_action_t comm = SIMIX_comm_irecv(simcall->issuer, rdv, dst_buff,
454 dst_buff_size, match_fun, data);
455 simcall->mc_value = 0;
456 SIMIX_pre_comm_wait(simcall, comm, timeout);
458 smx_action_t SIMIX_pre_comm_irecv(smx_simcall_t simcall, smx_rdv_t rdv,
459 void *dst_buff, size_t *dst_buff_size,
460 int (*match_fun)(void *, void *, smx_action_t),
462 return SIMIX_comm_irecv(simcall->issuer, rdv, dst_buff, dst_buff_size,
465 smx_action_t SIMIX_comm_irecv(smx_process_t dst_proc, smx_rdv_t rdv,
466 void *dst_buff, size_t *dst_buff_size,
467 int (*match_fun)(void *, void *, smx_action_t), void *data)
469 XBT_DEBUG("recv from %p %p\n", rdv, rdv->comm_fifo);
470 smx_action_t this_action = SIMIX_comm_new(SIMIX_COMM_RECEIVE);
472 smx_action_t other_action;
473 //communication already done, get it inside the fifo of completed comms
474 //permanent receive v1
475 //int already_received=0;
476 if(rdv->permanent_receiver && xbt_fifo_size(rdv->done_comm_fifo)!=0){
478 XBT_DEBUG("We have a comm that has probably already been received, trying to match it, to skip the communication\n");
479 //find a match in the already received fifo
480 other_action = SIMIX_fifo_get_comm(rdv->done_comm_fifo, SIMIX_COMM_SEND, match_fun, data, this_action);
481 //if not found, assume the receiver came first, register it to the mailbox in the classical way
483 XBT_DEBUG("We have messages in the permanent receive list, but not the one we are looking for, pushing request into fifo\n");
484 other_action = this_action;
485 SIMIX_rdv_push(rdv, this_action);
487 if(other_action->comm.surf_comm && SIMIX_comm_get_remains(other_action)==0.0)
489 XBT_DEBUG("comm %p has been already sent, and is finished, destroy it\n",&(other_action->comm));
490 other_action->state = SIMIX_DONE;
491 other_action->comm.type = SIMIX_COMM_DONE;
492 other_action->comm.rdv = NULL;
493 //SIMIX_comm_destroy(this_action);
494 //--smx_total_comms; // this creation was a pure waste
495 //already_received=1;
496 //other_action->comm.refcount--;
498 XBT_DEBUG("Not yet finished, we have to wait %d\n", xbt_fifo_size(rdv->comm_fifo));
500 other_action->comm.refcount--;
501 SIMIX_comm_destroy(this_action);
502 --smx_total_comms; // this creation was a pure waste
505 /* Prepare an action describing us, so that it gets passed to the user-provided filter of other side */
507 /* Look for communication action matching our needs. We also provide a description of
508 * ourself so that the other side also gets a chance of choosing if it wants to match with us.
510 * If it is not found then push our communication into the rendez-vous point */
511 other_action = SIMIX_fifo_get_comm(rdv->comm_fifo, SIMIX_COMM_SEND, match_fun, data, this_action);
514 XBT_DEBUG("Receive pushed first %d\n", xbt_fifo_size(rdv->comm_fifo));
515 other_action = this_action;
516 SIMIX_rdv_push(rdv, this_action);
518 SIMIX_comm_destroy(this_action);
519 --smx_total_comms; // this creation was a pure waste
520 other_action->state = SIMIX_READY;
521 other_action->comm.type = SIMIX_COMM_READY;
522 //other_action->comm.refcount--;
524 xbt_fifo_push(dst_proc->comms, other_action);
527 /* Setup communication action */
528 other_action->comm.dst_proc = dst_proc;
529 other_action->comm.dst_buff = dst_buff;
530 other_action->comm.dst_buff_size = dst_buff_size;
531 other_action->comm.dst_data = data;
533 other_action->comm.match_fun = match_fun;
536 /*if(already_received)//do the actual copy, because the first one after the comm didn't have all the info
537 SIMIX_comm_copy_data(other_action);*/
540 if (MC_is_active()) {
541 other_action->state = SIMIX_RUNNING;
545 SIMIX_comm_start(other_action);
550 smx_action_t SIMIX_pre_comm_iprobe(smx_simcall_t simcall, smx_rdv_t rdv,
552 int (*match_fun)(void *, void *, smx_action_t),
554 return SIMIX_comm_iprobe(simcall->issuer, rdv, src, tag, match_fun, data);
557 smx_action_t SIMIX_comm_iprobe(smx_process_t dst_proc, smx_rdv_t rdv, int src,
558 int tag, int (*match_fun)(void *, void *, smx_action_t), void *data)
560 XBT_DEBUG("iprobe from %p %p\n", rdv, rdv->comm_fifo);
561 smx_action_t this_action = SIMIX_comm_new(SIMIX_COMM_RECEIVE);
563 smx_action_t other_action=NULL;
564 if(rdv->permanent_receiver && xbt_fifo_size(rdv->done_comm_fifo)!=0){
565 //find a match in the already received fifo
566 XBT_DEBUG("first try in the perm recv mailbox \n");
568 other_action = SIMIX_fifo_probe_comm(rdv->done_comm_fifo, SIMIX_COMM_SEND, match_fun, data, this_action);
572 XBT_DEBUG("second try in the other mailbox");
573 other_action = SIMIX_fifo_probe_comm(rdv->comm_fifo, SIMIX_COMM_SEND, match_fun, data, this_action);
576 if(other_action)other_action->comm.refcount--;
578 SIMIX_comm_destroy(this_action);
583 void SIMIX_pre_comm_wait(smx_simcall_t simcall, smx_action_t action, double timeout)
585 int idx = simcall->mc_value;
586 /* the simcall may be a wait, a send or a recv */
589 /* Associate this simcall to the wait action */
590 XBT_DEBUG("SIMIX_pre_comm_wait, %p", action);
592 xbt_fifo_push(action->simcalls, simcall);
593 simcall->issuer->waiting_action = action;
595 if (MC_is_active()) {
597 action->state = SIMIX_DONE;
599 /* If we reached this point, the wait simcall must have a timeout */
600 /* Otherwise it shouldn't be enabled and executed by the MC */
604 if (action->comm.src_proc == simcall->issuer)
605 action->state = SIMIX_SRC_TIMEOUT;
607 action->state = SIMIX_DST_TIMEOUT;
610 SIMIX_comm_finish(action);
614 /* If the action has already finish perform the error handling, */
615 /* otherwise set up a waiting timeout on the right side */
616 if (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING) {
617 SIMIX_comm_finish(action);
618 } 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 */
619 sleep = surf_workstation_model->extension.workstation.sleep(simcall->issuer->smx_host->host, timeout);
620 surf_workstation_model->action_data_set(sleep, action);
622 if (simcall->issuer == action->comm.src_proc)
623 action->comm.src_timeout = sleep;
625 action->comm.dst_timeout = sleep;
629 void SIMIX_pre_comm_test(smx_simcall_t simcall, smx_action_t action)
632 simcall_comm_test__set__result(simcall, action->comm.src_proc && action->comm.dst_proc);
633 if(simcall_comm_test__get__result(simcall)){
634 action->state = SIMIX_DONE;
635 xbt_fifo_push(action->simcalls, simcall);
636 SIMIX_comm_finish(action);
638 SIMIX_simcall_answer(simcall);
643 simcall_comm_test__set__result(simcall, (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING));
644 if (simcall_comm_test__get__result(simcall)) {
645 xbt_fifo_push(action->simcalls, simcall);
646 SIMIX_comm_finish(action);
648 SIMIX_simcall_answer(simcall);
652 void SIMIX_pre_comm_testany(smx_simcall_t simcall, xbt_dynar_t actions)
654 int idx = simcall->mc_value;
657 simcall_comm_testany__set__result(simcall, -1);
661 SIMIX_simcall_answer(simcall);
663 action = xbt_dynar_get_as(actions, idx, smx_action_t);
664 simcall_comm_testany__set__result(simcall, idx);
665 xbt_fifo_push(action->simcalls, simcall);
666 action->state = SIMIX_DONE;
667 SIMIX_comm_finish(action);
672 xbt_dynar_foreach(simcall_comm_testany__get__comms(simcall), cursor,action) {
673 if (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING) {
674 simcall_comm_testany__set__result(simcall, cursor);
675 xbt_fifo_push(action->simcalls, simcall);
676 SIMIX_comm_finish(action);
680 SIMIX_simcall_answer(simcall);
683 void SIMIX_pre_comm_waitany(smx_simcall_t simcall, xbt_dynar_t actions)
685 int idx = simcall->mc_value;
687 unsigned int cursor = 0;
690 action = xbt_dynar_get_as(actions, idx, smx_action_t);
691 xbt_fifo_push(action->simcalls, simcall);
692 simcall_comm_waitany__set__result(simcall, idx);
693 action->state = SIMIX_DONE;
694 SIMIX_comm_finish(action);
698 xbt_dynar_foreach(actions, cursor, action){
699 /* associate this simcall to the the action */
700 xbt_fifo_push(action->simcalls, simcall);
702 /* see if the action is already finished */
703 if (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING){
704 SIMIX_comm_finish(action);
710 void SIMIX_waitany_remove_simcall_from_actions(smx_simcall_t simcall)
713 unsigned int cursor = 0;
714 xbt_dynar_t actions = simcall_comm_waitany__get__comms(simcall);
716 xbt_dynar_foreach(actions, cursor, action) {
717 xbt_fifo_remove(action->simcalls, simcall);
722 * \brief Starts the simulation of a communication action.
723 * \param action the communication action
725 XBT_INLINE void SIMIX_comm_start(smx_action_t action)
727 /* If both the sender and the receiver are already there, start the communication */
728 if (action->state == SIMIX_READY) {
730 smx_host_t sender = action->comm.src_proc->smx_host;
731 smx_host_t receiver = action->comm.dst_proc->smx_host;
733 XBT_DEBUG("Starting communication %p from '%s' to '%s'", action,
734 SIMIX_host_get_name(sender), SIMIX_host_get_name(receiver));
736 action->comm.surf_comm = surf_workstation_model->extension.workstation.
737 communicate(sender->host, receiver->host, action->comm.task_size, action->comm.rate);
739 surf_workstation_model->action_data_set(action->comm.surf_comm, action);
741 action->state = SIMIX_RUNNING;
743 /* If a link is failed, detect it immediately */
744 if (surf_workstation_model->action_state_get(action->comm.surf_comm) == SURF_ACTION_FAILED) {
745 XBT_DEBUG("Communication from '%s' to '%s' failed to start because of a link failure",
746 SIMIX_host_get_name(sender), SIMIX_host_get_name(receiver));
747 action->state = SIMIX_LINK_FAILURE;
748 SIMIX_comm_destroy_internal_actions(action);
751 /* If any of the process is suspend, create the action but stop its execution,
752 it will be restarted when the sender process resume */
753 if (SIMIX_process_is_suspended(action->comm.src_proc) ||
754 SIMIX_process_is_suspended(action->comm.dst_proc)) {
755 /* FIXME: check what should happen with the action state */
757 if (SIMIX_process_is_suspended(action->comm.src_proc))
758 XBT_DEBUG("The communication is suspended on startup because src (%s:%s) were suspended since it initiated the communication",
759 SIMIX_host_get_name(action->comm.src_proc->smx_host), action->comm.src_proc->name);
761 XBT_DEBUG("The communication is suspended on startup because dst (%s:%s) were suspended since it initiated the communication",
762 SIMIX_host_get_name(action->comm.dst_proc->smx_host), action->comm.dst_proc->name);
764 surf_workstation_model->suspend(action->comm.surf_comm);
771 * \brief Answers the SIMIX simcalls associated to a communication action.
772 * \param action a finished communication action
774 void SIMIX_comm_finish(smx_action_t action)
776 unsigned int destroy_count = 0;
777 smx_simcall_t simcall;
779 while ((simcall = xbt_fifo_shift(action->simcalls))) {
781 /* If a waitany simcall is waiting for this action to finish, then remove
782 it from the other actions in the waitany list. Afterwards, get the
783 position of the actual action in the waitany dynar and
784 return it as the result of the simcall */
785 if (simcall->call == SIMCALL_COMM_WAITANY) {
786 SIMIX_waitany_remove_simcall_from_actions(simcall);
788 simcall_comm_waitany__set__result(simcall, xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &action));
791 /* If the action is still in a rendez-vous point then remove from it */
792 if (action->comm.rdv)
793 SIMIX_rdv_remove(action->comm.rdv, action);
795 XBT_DEBUG("SIMIX_comm_finish: action state = %d", (int)action->state);
797 /* Check out for errors */
798 switch (action->state) {
801 XBT_DEBUG("Communication %p complete!", action);
802 SIMIX_comm_copy_data(action);
805 case SIMIX_SRC_TIMEOUT:
806 SMX_EXCEPTION(simcall->issuer, timeout_error, 0,
807 "Communication timeouted because of sender");
810 case SIMIX_DST_TIMEOUT:
811 SMX_EXCEPTION(simcall->issuer, timeout_error, 0,
812 "Communication timeouted because of receiver");
815 case SIMIX_SRC_HOST_FAILURE:
816 if (simcall->issuer == action->comm.src_proc)
817 simcall->issuer->context->iwannadie = 1;
818 // SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
820 SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
823 case SIMIX_DST_HOST_FAILURE:
824 if (simcall->issuer == action->comm.dst_proc)
825 simcall->issuer->context->iwannadie = 1;
826 // SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
828 SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
831 case SIMIX_LINK_FAILURE:
832 XBT_DEBUG("Link failure in action %p between '%s' and '%s': posting an exception to the issuer: %s (%p) detached:%d",
834 action->comm.src_proc ? action->comm.src_proc->smx_host->name : NULL,
835 action->comm.dst_proc ? action->comm.dst_proc->smx_host->name : NULL,
836 simcall->issuer->name, simcall->issuer, action->comm.detached);
837 if (action->comm.src_proc == simcall->issuer) {
838 XBT_DEBUG("I'm source");
839 } else if (action->comm.dst_proc == simcall->issuer) {
840 XBT_DEBUG("I'm dest");
842 XBT_DEBUG("I'm neither source nor dest");
844 SMX_EXCEPTION(simcall->issuer, network_error, 0, "Link failure");
848 if (simcall->issuer == action->comm.dst_proc)
849 SMX_EXCEPTION(simcall->issuer, cancel_error, 0,
850 "Communication canceled by the sender");
852 SMX_EXCEPTION(simcall->issuer, cancel_error, 0,
853 "Communication canceled by the receiver");
857 xbt_die("Unexpected action state in SIMIX_comm_finish: %d", (int)action->state);
860 /* if there is an exception during a waitany or a testany, indicate the position of the failed communication */
861 if (simcall->issuer->doexception) {
862 if (simcall->call == SIMCALL_COMM_WAITANY) {
863 simcall->issuer->running_ctx->exception.value = xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &action);
865 else if (simcall->call == SIMCALL_COMM_TESTANY) {
866 simcall->issuer->running_ctx->exception.value = xbt_dynar_search(simcall_comm_testany__get__comms(simcall), &action);
870 if (surf_workstation_model->extension.
871 workstation.get_state(simcall->issuer->smx_host->host) != SURF_RESOURCE_ON) {
872 simcall->issuer->context->iwannadie = 1;
875 simcall->issuer->waiting_action = NULL;
876 xbt_fifo_remove(simcall->issuer->comms, action);
877 SIMIX_simcall_answer(simcall);
881 while (destroy_count-- > 0)
882 SIMIX_comm_destroy(action);
886 * \brief This function is called when a Surf communication action is finished.
887 * \param action the corresponding Simix communication
889 void SIMIX_post_comm(smx_action_t action)
891 /* Update action state */
892 if (action->comm.src_timeout &&
893 surf_workstation_model->action_state_get(action->comm.src_timeout) == SURF_ACTION_DONE)
894 action->state = SIMIX_SRC_TIMEOUT;
895 else if (action->comm.dst_timeout &&
896 surf_workstation_model->action_state_get(action->comm.dst_timeout) == SURF_ACTION_DONE)
897 action->state = SIMIX_DST_TIMEOUT;
898 else if (action->comm.src_timeout &&
899 surf_workstation_model->action_state_get(action->comm.src_timeout) == SURF_ACTION_FAILED)
900 action->state = SIMIX_SRC_HOST_FAILURE;
901 else if (action->comm.dst_timeout &&
902 surf_workstation_model->action_state_get(action->comm.dst_timeout) == SURF_ACTION_FAILED)
903 action->state = SIMIX_DST_HOST_FAILURE;
904 else if (action->comm.surf_comm &&
905 surf_workstation_model->action_state_get(action->comm.surf_comm) == SURF_ACTION_FAILED) {
906 XBT_DEBUG("Puta madre. Surf says that the link broke");
907 action->state = SIMIX_LINK_FAILURE;
909 action->state = SIMIX_DONE;
911 XBT_DEBUG("SIMIX_post_comm: comm %p, state %d, src_proc %p, dst_proc %p, detached: %d",
912 action, (int)action->state, action->comm.src_proc, action->comm.dst_proc, action->comm.detached);
914 /* destroy the surf actions associated with the Simix communication */
915 SIMIX_comm_destroy_internal_actions(action);
917 /* remove the communication action from the list of pending communications
918 * of both processes (if they still exist) */
919 if (action->comm.src_proc) {
920 xbt_fifo_remove(action->comm.src_proc->comms, action);
922 if (action->comm.dst_proc) {
923 xbt_fifo_remove(action->comm.dst_proc->comms, action);
926 /* if there are simcalls associated with the action, then answer them */
927 if (xbt_fifo_size(action->simcalls)) {
928 SIMIX_comm_finish(action);
932 void SIMIX_pre_comm_cancel(smx_simcall_t simcall, smx_action_t action){
933 SIMIX_comm_cancel(action);
935 void SIMIX_comm_cancel(smx_action_t action)
937 /* if the action is a waiting state means that it is still in a rdv */
938 /* so remove from it and delete it */
939 if (action->state == SIMIX_WAITING) {
940 SIMIX_rdv_remove(action->comm.rdv, action);
941 action->state = SIMIX_CANCELED;
943 else if (!MC_is_active() /* when running the MC there are no surf actions */
944 && (action->state == SIMIX_READY || action->state == SIMIX_RUNNING)) {
946 surf_workstation_model->action_cancel(action->comm.surf_comm);
950 void SIMIX_comm_suspend(smx_action_t action)
952 /*FIXME: shall we suspend also the timeout actions? */
953 if (action->comm.surf_comm)
954 surf_workstation_model->suspend(action->comm.surf_comm);
955 /* in the other case, the action will be suspended on creation, in SIMIX_comm_start() */
958 void SIMIX_comm_resume(smx_action_t action)
960 /*FIXME: check what happen with the timeouts */
961 if (action->comm.surf_comm)
962 surf_workstation_model->resume(action->comm.surf_comm);
963 /* in the other case, the action were not really suspended yet, see SIMIX_comm_suspend() and SIMIX_comm_start() */
967 /************* Action Getters **************/
969 double SIMIX_pre_comm_get_remains(smx_simcall_t simcall, smx_action_t action){
970 return SIMIX_comm_get_remains(action);
973 * \brief get the amount remaining from the communication
974 * \param action The communication
976 double SIMIX_comm_get_remains(smx_action_t action)
984 switch (action->state) {
987 remains = surf_workstation_model->get_remains(action->comm.surf_comm);
992 remains = 0; /*FIXME: check what should be returned */
996 remains = 0; /*FIXME: is this correct? */
1002 e_smx_state_t SIMIX_pre_comm_get_state(smx_simcall_t simcall, smx_action_t action){
1003 return SIMIX_comm_get_state(action);
1005 e_smx_state_t SIMIX_comm_get_state(smx_action_t action)
1007 return action->state;
1010 void* SIMIX_pre_comm_get_src_data(smx_simcall_t simcall, smx_action_t action){
1011 return SIMIX_comm_get_src_data(action);
1014 * \brief Return the user data associated to the sender of the communication
1015 * \param action The communication
1016 * \return the user data
1018 void* SIMIX_comm_get_src_data(smx_action_t action)
1020 return action->comm.src_data;
1023 void* SIMIX_pre_comm_get_dst_data(smx_simcall_t simcall, smx_action_t action){
1024 return SIMIX_comm_get_dst_data(action);
1027 * \brief Return the user data associated to the receiver of the communication
1028 * \param action The communication
1029 * \return the user data
1031 void* SIMIX_comm_get_dst_data(smx_action_t action)
1033 return action->comm.dst_data;
1036 smx_process_t SIMIX_pre_comm_get_src_proc(smx_simcall_t simcall, smx_action_t action){
1037 return SIMIX_comm_get_src_proc(action);
1039 smx_process_t SIMIX_comm_get_src_proc(smx_action_t action)
1041 return action->comm.src_proc;
1044 smx_process_t SIMIX_pre_comm_get_dst_proc(smx_simcall_t simcall, smx_action_t action){
1045 return SIMIX_comm_get_dst_proc(action);
1047 smx_process_t SIMIX_comm_get_dst_proc(smx_action_t action)
1049 return action->comm.dst_proc;
1052 #ifdef HAVE_LATENCY_BOUND_TRACKING
1054 * \brief verify if communication is latency bounded
1055 * \param comm The communication
1057 XBT_INLINE int SIMIX_comm_is_latency_bounded(smx_action_t action)
1062 if (action->comm.surf_comm){
1063 XBT_DEBUG("Getting latency limited for surf_action (%p)", action->comm.surf_comm);
1064 action->latency_limited = surf_workstation_model->get_latency_limited(action->comm.surf_comm);
1065 XBT_DEBUG("Action limited is %d", action->latency_limited);
1067 return action->latency_limited;
1071 /******************************************************************************/
1072 /* SIMIX_comm_copy_data callbacks */
1073 /******************************************************************************/
1074 static void (*SIMIX_comm_copy_data_callback) (smx_action_t, void*, size_t) =
1075 &SIMIX_comm_copy_pointer_callback;
1078 SIMIX_comm_set_copy_data_callback(void (*callback) (smx_action_t, void*, size_t))
1080 SIMIX_comm_copy_data_callback = callback;
1083 void SIMIX_comm_copy_pointer_callback(smx_action_t comm, void* buff, size_t buff_size)
1085 xbt_assert((buff_size == sizeof(void *)),
1086 "Cannot copy %zu bytes: must be sizeof(void*)", buff_size);
1087 *(void **) (comm->comm.dst_buff) = buff;
1090 void SIMIX_comm_copy_buffer_callback(smx_action_t comm, void* buff, size_t buff_size)
1092 XBT_DEBUG("Copy the data over");
1093 memcpy(comm->comm.dst_buff, buff, buff_size);
1094 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
1096 comm->comm.src_buff = NULL;
1102 * \brief Copy the communication data from the sender's buffer to the receiver's one
1103 * \param comm The communication
1105 void SIMIX_comm_copy_data(smx_action_t comm)
1107 size_t buff_size = comm->comm.src_buff_size;
1108 /* If there is no data to be copy then return */
1109 if (!comm->comm.src_buff || !comm->comm.dst_buff || comm->comm.copied)
1112 XBT_DEBUG("Copying comm %p data from %s (%p) -> %s (%p) (%zu bytes)",
1114 comm->comm.src_proc ? comm->comm.src_proc->smx_host->name : "a finished process",
1115 comm->comm.src_buff,
1116 comm->comm.dst_proc ? comm->comm.dst_proc->smx_host->name : "a finished process",
1117 comm->comm.dst_buff, buff_size);
1119 /* Copy at most dst_buff_size bytes of the message to receiver's buffer */
1120 if (comm->comm.dst_buff_size)
1121 buff_size = MIN(buff_size, *(comm->comm.dst_buff_size));
1123 /* Update the receiver's buffer size to the copied amount */
1124 if (comm->comm.dst_buff_size)
1125 *comm->comm.dst_buff_size = buff_size;
1128 SIMIX_comm_copy_data_callback (comm, comm->comm.src_buff, buff_size);
1130 /* Set the copied flag so we copy data only once */
1131 /* (this function might be called from both communication ends) */
1132 comm->comm.copied = 1;