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"
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_EXPORT_NO_IMPORT(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);
29 static void SIMIX_comm_start(smx_action_t action);
31 void SIMIX_network_init(void)
33 rdv_points = xbt_dict_new_homogeneous(SIMIX_rdv_free);
36 void SIMIX_network_exit(void)
38 xbt_dict_free(&rdv_points);
41 /******************************************************************************/
42 /* Rendez-Vous Points */
43 /******************************************************************************/
45 smx_rdv_t SIMIX_pre_rdv_create(smx_simcall_t simcall, const char *name){
46 return SIMIX_rdv_create(name);
48 smx_rdv_t SIMIX_rdv_create(const char *name)
50 /* two processes may have pushed the same rdv_create simcall at the same time */
51 smx_rdv_t rdv = name ? xbt_dict_get_or_null(rdv_points, name) : NULL;
54 rdv = xbt_new0(s_smx_rvpoint_t, 1);
55 rdv->name = name ? xbt_strdup(name) : NULL;
56 rdv->comm_fifo = xbt_fifo_new();
57 rdv->done_comm_fifo = xbt_fifo_new();
58 rdv->permanent_receiver=NULL;
60 XBT_DEBUG("Creating a mailbox at %p with name %s\n", rdv, name);
63 xbt_dict_set(rdv_points, rdv->name, rdv, NULL);
68 void SIMIX_pre_rdv_destroy(smx_simcall_t simcall, smx_rdv_t rdv){
69 return SIMIX_rdv_destroy(rdv);
71 void SIMIX_rdv_destroy(smx_rdv_t rdv)
74 xbt_dict_remove(rdv_points, rdv->name);
77 void SIMIX_rdv_free(void *data)
79 XBT_DEBUG("rdv free %p", data);
80 smx_rdv_t rdv = (smx_rdv_t) data;
82 xbt_fifo_free(rdv->comm_fifo);
83 xbt_fifo_free(rdv->done_comm_fifo);
88 xbt_dict_t SIMIX_get_rdv_points()
93 smx_rdv_t SIMIX_pre_rdv_get_by_name(smx_simcall_t simcall, const char *name){
94 return SIMIX_rdv_get_by_name(name);
96 smx_rdv_t SIMIX_rdv_get_by_name(const char *name)
98 return xbt_dict_get_or_null(rdv_points, name);
101 int SIMIX_pre_rdv_comm_count_by_host(smx_simcall_t simcall, smx_rdv_t rdv, smx_host_t host){
102 return SIMIX_rdv_comm_count_by_host(rdv, host);
104 int SIMIX_rdv_comm_count_by_host(smx_rdv_t rdv, smx_host_t host)
106 smx_action_t comm = NULL;
107 xbt_fifo_item_t item = NULL;
110 xbt_fifo_foreach(rdv->comm_fifo, item, comm, smx_action_t) {
111 if (comm->comm.src_proc->smx_host == host)
118 smx_action_t SIMIX_pre_rdv_get_head(smx_simcall_t simcall, smx_rdv_t rdv){
119 return SIMIX_rdv_get_head(rdv);
121 smx_action_t SIMIX_rdv_get_head(smx_rdv_t rdv)
123 return xbt_fifo_get_item_content(xbt_fifo_get_first_item(rdv->comm_fifo));
126 smx_process_t SIMIX_pre_rdv_get_receiver(smx_simcall_t simcall, smx_rdv_t rdv){
127 return SIMIX_rdv_get_receiver(rdv);
130 * \brief get the receiver (process associated to the mailbox)
131 * \param rdv The rendez-vous point
132 * \return process The receiving process (NULL if not set)
134 smx_process_t SIMIX_rdv_get_receiver(smx_rdv_t rdv)
136 return rdv->permanent_receiver;
139 void SIMIX_pre_rdv_set_receiver(smx_simcall_t simcall, smx_rdv_t rdv,
140 smx_process_t process){
141 SIMIX_rdv_set_receiver(rdv, process);
144 * \brief set the receiver of the rendez vous point to allow eager sends
145 * \param rdv The rendez-vous point
146 * \param process The receiving process
148 void SIMIX_rdv_set_receiver(smx_rdv_t rdv, smx_process_t process)
150 rdv->permanent_receiver=process;
154 * \brief Pushes a communication action into a rendez-vous point
155 * \param rdv The rendez-vous point
156 * \param comm The communication action
158 static XBT_INLINE void SIMIX_rdv_push(smx_rdv_t rdv, smx_action_t comm)
160 xbt_fifo_push(rdv->comm_fifo, comm);
161 comm->comm.rdv = rdv;
165 * \brief Removes a communication action from a rendez-vous point
166 * \param rdv The rendez-vous point
167 * \param comm The communication action
169 XBT_INLINE void SIMIX_rdv_remove(smx_rdv_t rdv, smx_action_t comm)
171 xbt_fifo_remove(rdv->comm_fifo, comm);
172 comm->comm.rdv = NULL;
176 * \brief Checks if there is a communication action queued in a fifo matching our needs
177 * \param type The type of communication we are looking for (comm_send, comm_recv)
178 * \return The communication action if found, NULL otherwise
180 smx_action_t SIMIX_fifo_get_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
181 int (*match_fun)(void *, void *,smx_action_t),
182 void *this_user_data, smx_action_t my_action)
185 xbt_fifo_item_t item;
186 void* other_user_data = NULL;
188 xbt_fifo_foreach(fifo, item, action, smx_action_t) {
189 if (action->comm.type == SIMIX_COMM_SEND) {
190 other_user_data = action->comm.src_data;
191 } else if (action->comm.type == SIMIX_COMM_RECEIVE) {
192 other_user_data = action->comm.dst_data;
194 if (action->comm.type == type &&
195 (!match_fun || match_fun(this_user_data, other_user_data, action)) &&
196 (!action->comm.match_fun || action->comm.match_fun(other_user_data, this_user_data, my_action))) {
197 XBT_DEBUG("Found a matching communication action %p", action);
198 xbt_fifo_remove_item(fifo, item);
199 xbt_fifo_free_item(item);
200 action->comm.refcount++;
202 action->comm.rdv_cpy = action->comm.rdv;
204 action->comm.rdv = NULL;
207 XBT_DEBUG("Sorry, communication action %p does not match our needs:"
208 " its type is %d but we are looking for a comm of type %d (or maybe the filtering didn't match)",
209 action, (int)action->comm.type, (int)type);
211 XBT_DEBUG("No matching communication action found");
217 * \brief Checks if there is a communication action queued in a fifo matching our needs, but leave it there
218 * \param type The type of communication we are looking for (comm_send, comm_recv)
219 * \return The communication action if found, NULL otherwise
221 smx_action_t SIMIX_fifo_probe_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
222 int (*match_fun)(void *, void *,smx_action_t),
223 void *this_user_data, smx_action_t my_action)
226 xbt_fifo_item_t item;
227 void* other_user_data = NULL;
229 xbt_fifo_foreach(fifo, item, action, smx_action_t) {
230 if (action->comm.type == SIMIX_COMM_SEND) {
231 other_user_data = action->comm.src_data;
232 } else if (action->comm.type == SIMIX_COMM_RECEIVE) {
233 other_user_data = action->comm.dst_data;
235 if (action->comm.type == type &&
236 (!match_fun || match_fun(this_user_data, other_user_data, action)) &&
237 (!action->comm.match_fun || action->comm.match_fun(other_user_data, this_user_data, my_action))) {
238 XBT_DEBUG("Found a matching communication action %p", action);
239 action->comm.refcount++;
243 XBT_DEBUG("Sorry, communication action %p does not match our needs:"
244 " its type is %d but we are looking for a comm of type %d (or maybe the filtering didn't match)",
245 action, (int)action->comm.type, (int)type);
247 XBT_DEBUG("No matching communication action found");
250 /******************************************************************************/
251 /* Communication Actions */
252 /******************************************************************************/
255 * \brief Creates a new communicate action
256 * \param type The direction of communication (comm_send, comm_recv)
257 * \return The new communicate action
259 smx_action_t SIMIX_comm_new(e_smx_comm_type_t type)
263 /* alloc structures */
264 act = xbt_mallocator_get(simix_global->action_mallocator);
266 act->type = SIMIX_ACTION_COMMUNICATE;
267 act->state = SIMIX_WAITING;
269 /* set communication */
270 act->comm.type = type;
271 act->comm.refcount = 1;
272 act->comm.src_data=NULL;
273 act->comm.dst_data=NULL;
276 #ifdef HAVE_LATENCY_BOUND_TRACKING
277 //initialize with unknown value
278 act->latency_limited = -1;
282 act->category = NULL;
285 XBT_DEBUG("Create communicate action %p", act);
292 * \brief Destroy a communicate action
293 * \param action The communicate action to be destroyed
295 void SIMIX_comm_destroy(smx_action_t action)
297 XBT_DEBUG("Destroy action %p (refcount: %d), state: %d",
298 action, action->comm.refcount, (int)action->state);
300 if (action->comm.refcount <= 0) {
301 xbt_backtrace_display_current();
302 xbt_die("The refcount of comm %p is already 0 before decreasing it. "
303 "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);
305 action->comm.refcount--;
306 if (action->comm.refcount > 0)
308 XBT_DEBUG("Really free communication %p; refcount is now %d", action,
309 action->comm.refcount);
311 #ifdef HAVE_LATENCY_BOUND_TRACKING
312 action->latency_limited = SIMIX_comm_is_latency_bounded( action ) ;
315 xbt_free(action->name);
316 SIMIX_comm_destroy_internal_actions(action);
318 if (action->comm.detached && action->state != SIMIX_DONE) {
319 /* the communication has failed and was detached:
320 * we have to free the buffer */
321 if (action->comm.clean_fun) {
322 action->comm.clean_fun(action->comm.src_buff);
324 action->comm.src_buff = NULL;
328 SIMIX_rdv_remove(action->comm.rdv, action);
330 xbt_mallocator_release(simix_global->action_mallocator, action);
333 void SIMIX_comm_destroy_internal_actions(smx_action_t action)
335 if (action->comm.surf_comm){
336 #ifdef HAVE_LATENCY_BOUND_TRACKING
337 action->latency_limited = SIMIX_comm_is_latency_bounded(action);
339 surf_action_unref(action->comm.surf_comm);
340 action->comm.surf_comm = NULL;
343 if (action->comm.src_timeout){
344 surf_action_unref(action->comm.src_timeout);
345 action->comm.src_timeout = NULL;
348 if (action->comm.dst_timeout){
349 surf_action_unref(action->comm.dst_timeout);
350 action->comm.dst_timeout = NULL;
354 void SIMIX_pre_comm_send(smx_simcall_t simcall, smx_rdv_t rdv,
355 double task_size, double rate,
356 void *src_buff, size_t src_buff_size,
357 int (*match_fun)(void *, void *,smx_action_t),
358 void (*copy_data_fun)(smx_action_t, void*, size_t),
359 void *data, double timeout){
360 smx_action_t comm = SIMIX_comm_isend(simcall->issuer, rdv, task_size, rate,
361 src_buff, src_buff_size, match_fun, NULL, copy_data_fun,
363 SIMCALL_SET_MC_VALUE(simcall, 0);
364 SIMIX_pre_comm_wait(simcall, comm, timeout);
366 smx_action_t SIMIX_pre_comm_isend(smx_simcall_t simcall, smx_rdv_t rdv,
367 double task_size, double rate,
368 void *src_buff, size_t src_buff_size,
369 int (*match_fun)(void *, void *,smx_action_t),
370 void (*clean_fun)(void *),
371 void (*copy_data_fun)(smx_action_t, void*, size_t),
372 void *data, int detached){
373 return SIMIX_comm_isend(simcall->issuer, rdv, task_size, rate, src_buff,
374 src_buff_size, match_fun, clean_fun, copy_data_fun, data, detached);
377 smx_action_t SIMIX_comm_isend(smx_process_t src_proc, smx_rdv_t rdv,
378 double task_size, double rate,
379 void *src_buff, size_t src_buff_size,
380 int (*match_fun)(void *, void *,smx_action_t),
381 void (*clean_fun)(void *), // used to free the action in case of problem after a detached send
382 void (*copy_data_fun)(smx_action_t, void*, size_t), // used to copy data if not default one
386 XBT_DEBUG("send from %p\n", rdv);
388 /* Prepare an action describing us, so that it gets passed to the user-provided filter of other side */
389 smx_action_t this_action = SIMIX_comm_new(SIMIX_COMM_SEND);
391 /* Look for communication action matching our needs. We also provide a description of
392 * ourself so that the other side also gets a chance of choosing if it wants to match with us.
394 * If it is not found then push our communication into the rendez-vous point */
395 smx_action_t other_action = SIMIX_fifo_get_comm(rdv->comm_fifo, SIMIX_COMM_RECEIVE, match_fun, data, this_action);
398 other_action = this_action;
400 if (rdv->permanent_receiver!=NULL){
401 //this mailbox is for small messages, which have to be sent right now
402 other_action->state = SIMIX_READY;
403 other_action->comm.dst_proc=rdv->permanent_receiver;
404 other_action->comm.refcount++;
405 xbt_fifo_push(rdv->done_comm_fifo,other_action);
406 other_action->comm.rdv=rdv;
407 XBT_DEBUG("pushing a message into the permanent receive fifo %p, comm %p \n", rdv, &(other_action->comm));
410 SIMIX_rdv_push(rdv, this_action);
413 XBT_DEBUG("Receive already pushed\n");
415 SIMIX_comm_destroy(this_action);
416 --smx_total_comms; // this creation was a pure waste
418 other_action->state = SIMIX_READY;
419 other_action->comm.type = SIMIX_COMM_READY;
422 xbt_fifo_push(src_proc->comms, other_action);
424 /* if the communication action is detached then decrease the refcount
425 * by one, so it will be eliminated by the receiver's destroy call */
427 other_action->comm.detached = 1;
428 other_action->comm.refcount--;
429 other_action->comm.clean_fun = clean_fun;
431 other_action->comm.clean_fun = NULL;
434 /* Setup the communication action */
435 other_action->comm.src_proc = src_proc;
436 other_action->comm.task_size = task_size;
437 other_action->comm.rate = rate;
438 other_action->comm.src_buff = src_buff;
439 other_action->comm.src_buff_size = src_buff_size;
440 other_action->comm.src_data = data;
442 other_action->comm.match_fun = match_fun;
443 other_action->comm.copy_data_fun = copy_data_fun;
446 if (MC_is_active()) {
447 other_action->state = SIMIX_RUNNING;
448 return (detached ? NULL : other_action);
451 SIMIX_comm_start(other_action);
452 return (detached ? NULL : other_action);
455 void SIMIX_pre_comm_recv(smx_simcall_t simcall, smx_rdv_t rdv,
456 void *dst_buff, size_t *dst_buff_size,
457 int (*match_fun)(void *, void *, smx_action_t),
458 void (*copy_data_fun)(smx_action_t, void*, size_t),
459 void *data, double timeout, double rate)
461 smx_action_t comm = SIMIX_comm_irecv(simcall->issuer, rdv, dst_buff,
462 dst_buff_size, match_fun, copy_data_fun, data, rate);
463 SIMCALL_SET_MC_VALUE(simcall, 0);
464 SIMIX_pre_comm_wait(simcall, comm, timeout);
467 smx_action_t SIMIX_pre_comm_irecv(smx_simcall_t simcall, smx_rdv_t rdv,
468 void *dst_buff, size_t *dst_buff_size,
469 int (*match_fun)(void *, void *, smx_action_t),
470 void (*copy_data_fun)(smx_action_t, void*, size_t),
471 void *data, double rate)
473 return SIMIX_comm_irecv(simcall->issuer, rdv, dst_buff, dst_buff_size,
474 match_fun, copy_data_fun, data, rate);
477 smx_action_t SIMIX_comm_irecv(smx_process_t dst_proc, smx_rdv_t rdv,
478 void *dst_buff, size_t *dst_buff_size,
479 int (*match_fun)(void *, void *, smx_action_t),
480 void (*copy_data_fun)(smx_action_t, void*, size_t), // used to copy data if not default one
481 void *data, double rate)
483 XBT_DEBUG("recv from %p %p\n", rdv, rdv->comm_fifo);
484 smx_action_t this_action = SIMIX_comm_new(SIMIX_COMM_RECEIVE);
486 smx_action_t other_action;
487 //communication already done, get it inside the fifo of completed comms
488 //permanent receive v1
489 //int already_received=0;
490 if(rdv->permanent_receiver && xbt_fifo_size(rdv->done_comm_fifo)!=0){
492 XBT_DEBUG("We have a comm that has probably already been received, trying to match it, to skip the communication\n");
493 //find a match in the already received fifo
494 other_action = SIMIX_fifo_get_comm(rdv->done_comm_fifo, SIMIX_COMM_SEND, match_fun, data, this_action);
495 //if not found, assume the receiver came first, register it to the mailbox in the classical way
497 XBT_DEBUG("We have messages in the permanent receive list, but not the one we are looking for, pushing request into fifo\n");
498 other_action = this_action;
499 SIMIX_rdv_push(rdv, this_action);
501 if(other_action->comm.surf_comm && SIMIX_comm_get_remains(other_action)==0.0)
503 XBT_DEBUG("comm %p has been already sent, and is finished, destroy it\n",&(other_action->comm));
504 other_action->state = SIMIX_DONE;
505 other_action->comm.type = SIMIX_COMM_DONE;
506 other_action->comm.rdv = NULL;
507 //SIMIX_comm_destroy(this_action);
508 //--smx_total_comms; // this creation was a pure waste
509 //already_received=1;
510 //other_action->comm.refcount--;
512 XBT_DEBUG("Not yet finished, we have to wait %d\n", xbt_fifo_size(rdv->comm_fifo));
514 other_action->comm.refcount--;
515 SIMIX_comm_destroy(this_action);
516 --smx_total_comms; // this creation was a pure waste
519 /* Prepare an action describing us, so that it gets passed to the user-provided filter of other side */
521 /* Look for communication action matching our needs. We also provide a description of
522 * ourself so that the other side also gets a chance of choosing if it wants to match with us.
524 * If it is not found then push our communication into the rendez-vous point */
525 other_action = SIMIX_fifo_get_comm(rdv->comm_fifo, SIMIX_COMM_SEND, match_fun, data, this_action);
528 XBT_DEBUG("Receive pushed first %d\n", xbt_fifo_size(rdv->comm_fifo));
529 other_action = this_action;
530 SIMIX_rdv_push(rdv, this_action);
532 SIMIX_comm_destroy(this_action);
533 --smx_total_comms; // this creation was a pure waste
534 other_action->state = SIMIX_READY;
535 other_action->comm.type = SIMIX_COMM_READY;
536 //other_action->comm.refcount--;
538 xbt_fifo_push(dst_proc->comms, other_action);
541 /* Setup communication action */
542 other_action->comm.dst_proc = dst_proc;
543 other_action->comm.dst_buff = dst_buff;
544 other_action->comm.dst_buff_size = dst_buff_size;
545 other_action->comm.dst_data = data;
548 (other_action->comm.rate == -1.0 || rate < other_action->comm.rate))
549 other_action->comm.rate = rate;
551 other_action->comm.match_fun = match_fun;
552 other_action->comm.copy_data_fun = copy_data_fun;
555 /*if(already_received)//do the actual copy, because the first one after the comm didn't have all the info
556 SIMIX_comm_copy_data(other_action);*/
559 if (MC_is_active()) {
560 other_action->state = SIMIX_RUNNING;
564 SIMIX_comm_start(other_action);
569 smx_action_t SIMIX_pre_comm_iprobe(smx_simcall_t simcall, smx_rdv_t rdv,
571 int (*match_fun)(void *, void *, smx_action_t),
573 return SIMIX_comm_iprobe(simcall->issuer, rdv, src, tag, match_fun, data);
576 smx_action_t SIMIX_comm_iprobe(smx_process_t dst_proc, smx_rdv_t rdv, int src,
577 int tag, int (*match_fun)(void *, void *, smx_action_t), void *data)
579 XBT_DEBUG("iprobe from %p %p\n", rdv, rdv->comm_fifo);
580 smx_action_t this_action = SIMIX_comm_new(SIMIX_COMM_RECEIVE);
582 smx_action_t other_action=NULL;
583 if(rdv->permanent_receiver && xbt_fifo_size(rdv->done_comm_fifo)!=0){
584 //find a match in the already received fifo
585 XBT_DEBUG("first try in the perm recv mailbox \n");
587 other_action = SIMIX_fifo_probe_comm(rdv->done_comm_fifo, SIMIX_COMM_SEND, match_fun, data, this_action);
591 XBT_DEBUG("second try in the other mailbox");
592 other_action = SIMIX_fifo_probe_comm(rdv->comm_fifo, SIMIX_COMM_SEND, match_fun, data, this_action);
595 if(other_action)other_action->comm.refcount--;
597 SIMIX_comm_destroy(this_action);
602 void SIMIX_pre_comm_wait(smx_simcall_t simcall, smx_action_t action, double timeout)
604 /* the simcall may be a wait, a send or a recv */
607 /* Associate this simcall to the wait action */
608 XBT_DEBUG("SIMIX_pre_comm_wait, %p", action);
610 xbt_fifo_push(action->simcalls, simcall);
611 simcall->issuer->waiting_action = action;
613 if (MC_is_active()) {
614 int idx = SIMCALL_GET_MC_VALUE(simcall);
616 action->state = SIMIX_DONE;
618 /* If we reached this point, the wait simcall must have a timeout */
619 /* Otherwise it shouldn't be enabled and executed by the MC */
623 if (action->comm.src_proc == simcall->issuer)
624 action->state = SIMIX_SRC_TIMEOUT;
626 action->state = SIMIX_DST_TIMEOUT;
629 SIMIX_comm_finish(action);
633 /* If the action has already finish perform the error handling, */
634 /* otherwise set up a waiting timeout on the right side */
635 if (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING) {
636 SIMIX_comm_finish(action);
637 } 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 */
638 sleep = surf_workstation_sleep(simcall->issuer->smx_host, timeout);
639 surf_action_set_data(sleep, action);
641 if (simcall->issuer == action->comm.src_proc)
642 action->comm.src_timeout = sleep;
644 action->comm.dst_timeout = sleep;
648 void SIMIX_pre_comm_test(smx_simcall_t simcall, smx_action_t action)
651 simcall_comm_test__set__result(simcall, action->comm.src_proc && action->comm.dst_proc);
652 if(simcall_comm_test__get__result(simcall)){
653 action->state = SIMIX_DONE;
654 xbt_fifo_push(action->simcalls, simcall);
655 SIMIX_comm_finish(action);
657 SIMIX_simcall_answer(simcall);
662 simcall_comm_test__set__result(simcall, (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING));
663 if (simcall_comm_test__get__result(simcall)) {
664 xbt_fifo_push(action->simcalls, simcall);
665 SIMIX_comm_finish(action);
667 SIMIX_simcall_answer(simcall);
671 void SIMIX_pre_comm_testany(smx_simcall_t simcall, xbt_dynar_t actions)
675 simcall_comm_testany__set__result(simcall, -1);
678 int idx = SIMCALL_GET_MC_VALUE(simcall);
680 SIMIX_simcall_answer(simcall);
682 action = xbt_dynar_get_as(actions, idx, smx_action_t);
683 simcall_comm_testany__set__result(simcall, idx);
684 xbt_fifo_push(action->simcalls, simcall);
685 action->state = SIMIX_DONE;
686 SIMIX_comm_finish(action);
691 xbt_dynar_foreach(simcall_comm_testany__get__comms(simcall), cursor,action) {
692 if (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING) {
693 simcall_comm_testany__set__result(simcall, cursor);
694 xbt_fifo_push(action->simcalls, simcall);
695 SIMIX_comm_finish(action);
699 SIMIX_simcall_answer(simcall);
702 void SIMIX_pre_comm_waitany(smx_simcall_t simcall, xbt_dynar_t actions)
705 unsigned int cursor = 0;
708 int idx = SIMCALL_GET_MC_VALUE(simcall);
709 action = xbt_dynar_get_as(actions, idx, smx_action_t);
710 xbt_fifo_push(action->simcalls, simcall);
711 simcall_comm_waitany__set__result(simcall, idx);
712 action->state = SIMIX_DONE;
713 SIMIX_comm_finish(action);
717 xbt_dynar_foreach(actions, cursor, action){
718 /* associate this simcall to the the action */
719 xbt_fifo_push(action->simcalls, simcall);
721 /* see if the action is already finished */
722 if (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING){
723 SIMIX_comm_finish(action);
729 void SIMIX_waitany_remove_simcall_from_actions(smx_simcall_t simcall)
732 unsigned int cursor = 0;
733 xbt_dynar_t actions = simcall_comm_waitany__get__comms(simcall);
735 xbt_dynar_foreach(actions, cursor, action) {
736 xbt_fifo_remove(action->simcalls, simcall);
741 * \brief Starts the simulation of a communication action.
742 * \param action the communication action
744 static XBT_INLINE void SIMIX_comm_start(smx_action_t action)
746 /* If both the sender and the receiver are already there, start the communication */
747 if (action->state == SIMIX_READY) {
749 smx_host_t sender = action->comm.src_proc->smx_host;
750 smx_host_t receiver = action->comm.dst_proc->smx_host;
752 XBT_DEBUG("Starting communication %p from '%s' to '%s'", action,
753 SIMIX_host_get_name(sender), SIMIX_host_get_name(receiver));
755 action->comm.surf_comm = surf_workstation_model_communicate(surf_workstation_model,
757 action->comm.task_size, action->comm.rate);
759 surf_action_set_data(action->comm.surf_comm, action);
761 action->state = SIMIX_RUNNING;
763 /* If a link is failed, detect it immediately */
764 if (surf_action_get_state(action->comm.surf_comm) == SURF_ACTION_FAILED) {
765 XBT_DEBUG("Communication from '%s' to '%s' failed to start because of a link failure",
766 SIMIX_host_get_name(sender), SIMIX_host_get_name(receiver));
767 action->state = SIMIX_LINK_FAILURE;
768 SIMIX_comm_destroy_internal_actions(action);
771 /* If any of the process is suspend, create the action but stop its execution,
772 it will be restarted when the sender process resume */
773 if (SIMIX_process_is_suspended(action->comm.src_proc) ||
774 SIMIX_process_is_suspended(action->comm.dst_proc)) {
775 /* FIXME: check what should happen with the action state */
777 if (SIMIX_process_is_suspended(action->comm.src_proc))
778 XBT_DEBUG("The communication is suspended on startup because src (%s:%s) were suspended since it initiated the communication",
779 SIMIX_host_get_name(action->comm.src_proc->smx_host), action->comm.src_proc->name);
781 XBT_DEBUG("The communication is suspended on startup because dst (%s:%s) were suspended since it initiated the communication",
782 SIMIX_host_get_name(action->comm.dst_proc->smx_host), action->comm.dst_proc->name);
784 surf_action_suspend(action->comm.surf_comm);
791 * \brief Answers the SIMIX simcalls associated to a communication action.
792 * \param action a finished communication action
794 void SIMIX_comm_finish(smx_action_t action)
796 unsigned int destroy_count = 0;
797 smx_simcall_t simcall;
800 while ((simcall = xbt_fifo_shift(action->simcalls))) {
802 /* If a waitany simcall is waiting for this action to finish, then remove
803 it from the other actions in the waitany list. Afterwards, get the
804 position of the actual action in the waitany dynar and
805 return it as the result of the simcall */
806 if (simcall->call == SIMCALL_COMM_WAITANY) {
807 SIMIX_waitany_remove_simcall_from_actions(simcall);
809 simcall_comm_waitany__set__result(simcall, xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &action));
812 /* If the action is still in a rendez-vous point then remove from it */
813 if (action->comm.rdv)
814 SIMIX_rdv_remove(action->comm.rdv, action);
816 XBT_DEBUG("SIMIX_comm_finish: action state = %d", (int)action->state);
818 /* Check out for errors */
819 switch (action->state) {
822 XBT_DEBUG("Communication %p complete!", action);
823 SIMIX_comm_copy_data(action);
826 case SIMIX_SRC_TIMEOUT:
827 SMX_EXCEPTION(simcall->issuer, timeout_error, 0,
828 "Communication timeouted because of sender");
831 case SIMIX_DST_TIMEOUT:
832 SMX_EXCEPTION(simcall->issuer, timeout_error, 0,
833 "Communication timeouted because of receiver");
836 case SIMIX_SRC_HOST_FAILURE:
837 if (simcall->issuer == action->comm.src_proc)
838 simcall->issuer->context->iwannadie = 1;
839 // SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
841 SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
844 case SIMIX_DST_HOST_FAILURE:
845 if (simcall->issuer == action->comm.dst_proc)
846 simcall->issuer->context->iwannadie = 1;
847 // SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
849 SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
852 case SIMIX_LINK_FAILURE:
853 XBT_DEBUG("Link failure in action %p between '%s' and '%s': posting an exception to the issuer: %s (%p) detached:%d",
855 action->comm.src_proc ? sg_host_name(action->comm.src_proc->smx_host) : NULL,
856 action->comm.dst_proc ? sg_host_name(action->comm.dst_proc->smx_host) : NULL,
857 simcall->issuer->name, simcall->issuer, action->comm.detached);
858 if (action->comm.src_proc == simcall->issuer) {
859 XBT_DEBUG("I'm source");
860 } else if (action->comm.dst_proc == simcall->issuer) {
861 XBT_DEBUG("I'm dest");
863 XBT_DEBUG("I'm neither source nor dest");
865 SMX_EXCEPTION(simcall->issuer, network_error, 0, "Link failure");
869 if (simcall->issuer == action->comm.dst_proc)
870 SMX_EXCEPTION(simcall->issuer, cancel_error, 0,
871 "Communication canceled by the sender");
873 SMX_EXCEPTION(simcall->issuer, cancel_error, 0,
874 "Communication canceled by the receiver");
878 xbt_die("Unexpected action state in SIMIX_comm_finish: %d", (int)action->state);
881 /* if there is an exception during a waitany or a testany, indicate the position of the failed communication */
882 if (simcall->issuer->doexception) {
883 if (simcall->call == SIMCALL_COMM_WAITANY) {
884 simcall->issuer->running_ctx->exception.value = xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &action);
886 else if (simcall->call == SIMCALL_COMM_TESTANY) {
887 simcall->issuer->running_ctx->exception.value = xbt_dynar_search(simcall_comm_testany__get__comms(simcall), &action);
891 if (surf_resource_get_state(surf_workstation_resource_priv(simcall->issuer->smx_host)) != SURF_RESOURCE_ON) {
892 simcall->issuer->context->iwannadie = 1;
895 simcall->issuer->waiting_action = NULL;
896 xbt_fifo_remove(simcall->issuer->comms, action);
897 if(action->comm.detached){
901 if(simcall->issuer == action->comm.src_proc){
902 if(action->comm.dst_proc){
903 xbt_swag_foreach(proc, simix_global->process_list)
905 if(proc==action->comm.dst_proc){
911 if(still_alive) xbt_fifo_remove(action->comm.dst_proc->comms, action);
913 if(simcall->issuer == action->comm.dst_proc){
914 if(action->comm.src_proc)
915 if(action->comm.dst_proc){
916 xbt_swag_foreach(proc, simix_global->process_list)
918 if(proc==action->comm.src_proc){
924 if(still_alive) xbt_fifo_remove(action->comm.src_proc->comms, action);
927 SIMIX_simcall_answer(simcall);
931 while (destroy_count-- > 0)
932 SIMIX_comm_destroy(action);
936 * \brief This function is called when a Surf communication action is finished.
937 * \param action the corresponding Simix communication
939 void SIMIX_post_comm(smx_action_t action)
941 /* Update action state */
942 if (action->comm.src_timeout &&
943 surf_action_get_state(action->comm.src_timeout) == SURF_ACTION_DONE)
944 action->state = SIMIX_SRC_TIMEOUT;
945 else if (action->comm.dst_timeout &&
946 surf_action_get_state(action->comm.dst_timeout) == SURF_ACTION_DONE)
947 action->state = SIMIX_DST_TIMEOUT;
948 else if (action->comm.src_timeout &&
949 surf_action_get_state(action->comm.src_timeout) == SURF_ACTION_FAILED)
950 action->state = SIMIX_SRC_HOST_FAILURE;
951 else if (action->comm.dst_timeout &&
952 surf_action_get_state(action->comm.dst_timeout) == SURF_ACTION_FAILED)
953 action->state = SIMIX_DST_HOST_FAILURE;
954 else if (action->comm.surf_comm &&
955 surf_action_get_state(action->comm.surf_comm) == SURF_ACTION_FAILED) {
956 XBT_DEBUG("Puta madre. Surf says that the link broke");
957 action->state = SIMIX_LINK_FAILURE;
959 action->state = SIMIX_DONE;
961 XBT_DEBUG("SIMIX_post_comm: comm %p, state %d, src_proc %p, dst_proc %p, detached: %d",
962 action, (int)action->state, action->comm.src_proc, action->comm.dst_proc, action->comm.detached);
964 /* destroy the surf actions associated with the Simix communication */
965 SIMIX_comm_destroy_internal_actions(action);
967 /* remove the communication action from the list of pending communications
968 * of both processes (if they still exist) */
969 if (action->comm.src_proc) {
970 xbt_fifo_remove(action->comm.src_proc->comms, action);
972 if (action->comm.dst_proc) {
973 xbt_fifo_remove(action->comm.dst_proc->comms, action);
976 /* if there are simcalls associated with the action, then answer them */
977 if (xbt_fifo_size(action->simcalls)) {
978 SIMIX_comm_finish(action);
982 void SIMIX_pre_comm_cancel(smx_simcall_t simcall, smx_action_t action){
983 SIMIX_comm_cancel(action);
985 void SIMIX_comm_cancel(smx_action_t action)
987 /* if the action is a waiting state means that it is still in a rdv */
988 /* so remove from it and delete it */
989 if (action->state == SIMIX_WAITING) {
990 SIMIX_rdv_remove(action->comm.rdv, action);
991 action->state = SIMIX_CANCELED;
993 else if (!MC_is_active() /* when running the MC there are no surf actions */
994 && (action->state == SIMIX_READY || action->state == SIMIX_RUNNING)) {
996 surf_action_cancel(action->comm.surf_comm);
1000 void SIMIX_comm_suspend(smx_action_t action)
1002 /*FIXME: shall we suspend also the timeout actions? */
1003 if (action->comm.surf_comm)
1004 surf_action_suspend(action->comm.surf_comm);
1005 /* in the other case, the action will be suspended on creation, in SIMIX_comm_start() */
1008 void SIMIX_comm_resume(smx_action_t action)
1010 /*FIXME: check what happen with the timeouts */
1011 if (action->comm.surf_comm)
1012 surf_action_resume(action->comm.surf_comm);
1013 /* in the other case, the action were not really suspended yet, see SIMIX_comm_suspend() and SIMIX_comm_start() */
1017 /************* Action Getters **************/
1019 double SIMIX_pre_comm_get_remains(smx_simcall_t simcall, smx_action_t action){
1020 return SIMIX_comm_get_remains(action);
1023 * \brief get the amount remaining from the communication
1024 * \param action The communication
1026 double SIMIX_comm_get_remains(smx_action_t action)
1034 switch (action->state) {
1037 remains = surf_action_get_remains(action->comm.surf_comm);
1042 remains = 0; /*FIXME: check what should be returned */
1046 remains = 0; /*FIXME: is this correct? */
1052 e_smx_state_t SIMIX_pre_comm_get_state(smx_simcall_t simcall, smx_action_t action){
1053 return SIMIX_comm_get_state(action);
1055 e_smx_state_t SIMIX_comm_get_state(smx_action_t action)
1057 return action->state;
1060 void* SIMIX_pre_comm_get_src_data(smx_simcall_t simcall, smx_action_t action){
1061 return SIMIX_comm_get_src_data(action);
1064 * \brief Return the user data associated to the sender of the communication
1065 * \param action The communication
1066 * \return the user data
1068 void* SIMIX_comm_get_src_data(smx_action_t action)
1070 return action->comm.src_data;
1073 void* SIMIX_pre_comm_get_dst_data(smx_simcall_t simcall, smx_action_t action){
1074 return SIMIX_comm_get_dst_data(action);
1077 * \brief Return the user data associated to the receiver of the communication
1078 * \param action The communication
1079 * \return the user data
1081 void* SIMIX_comm_get_dst_data(smx_action_t action)
1083 return action->comm.dst_data;
1086 smx_process_t SIMIX_pre_comm_get_src_proc(smx_simcall_t simcall, smx_action_t action){
1087 return SIMIX_comm_get_src_proc(action);
1089 smx_process_t SIMIX_comm_get_src_proc(smx_action_t action)
1091 return action->comm.src_proc;
1094 smx_process_t SIMIX_pre_comm_get_dst_proc(smx_simcall_t simcall, smx_action_t action){
1095 return SIMIX_comm_get_dst_proc(action);
1097 smx_process_t SIMIX_comm_get_dst_proc(smx_action_t action)
1099 return action->comm.dst_proc;
1102 #ifdef HAVE_LATENCY_BOUND_TRACKING
1103 int SIMIX_pre_comm_is_latency_bounded(smx_simcall_t simcall, smx_action_t action)
1105 return SIMIX_comm_is_latency_bounded(action);
1109 * \brief verify if communication is latency bounded
1110 * \param comm The communication
1112 int SIMIX_comm_is_latency_bounded(smx_action_t action)
1117 if (action->comm.surf_comm){
1118 XBT_DEBUG("Getting latency limited for surf_action (%p)", action->comm.surf_comm);
1119 action->latency_limited = surf_network_action_get_latency_limited(action->comm.surf_comm);
1120 XBT_DEBUG("Action limited is %d", action->latency_limited);
1122 return action->latency_limited;
1126 /******************************************************************************/
1127 /* SIMIX_comm_copy_data callbacks */
1128 /******************************************************************************/
1129 static void (*SIMIX_comm_copy_data_callback) (smx_action_t, void*, size_t) =
1130 &SIMIX_comm_copy_pointer_callback;
1133 SIMIX_comm_set_copy_data_callback(void (*callback) (smx_action_t, void*, size_t))
1135 SIMIX_comm_copy_data_callback = callback;
1138 void SIMIX_comm_copy_pointer_callback(smx_action_t comm, void* buff, size_t buff_size)
1140 xbt_assert((buff_size == sizeof(void *)),
1141 "Cannot copy %zu bytes: must be sizeof(void*)", buff_size);
1142 *(void **) (comm->comm.dst_buff) = buff;
1145 void SIMIX_comm_copy_buffer_callback(smx_action_t comm, void* buff, size_t buff_size)
1147 XBT_DEBUG("Copy the data over");
1148 memcpy(comm->comm.dst_buff, buff, buff_size);
1149 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
1151 comm->comm.src_buff = NULL;
1157 * \brief Copy the communication data from the sender's buffer to the receiver's one
1158 * \param comm The communication
1160 void SIMIX_comm_copy_data(smx_action_t comm)
1162 size_t buff_size = comm->comm.src_buff_size;
1163 /* If there is no data to be copy then return */
1164 if (!comm->comm.src_buff || !comm->comm.dst_buff || comm->comm.copied)
1167 XBT_DEBUG("Copying comm %p data from %s (%p) -> %s (%p) (%zu bytes)",
1169 comm->comm.src_proc ? sg_host_name(comm->comm.src_proc->smx_host) : "a finished process",
1170 comm->comm.src_buff,
1171 comm->comm.dst_proc ? sg_host_name(comm->comm.dst_proc->smx_host) : "a finished process",
1172 comm->comm.dst_buff, buff_size);
1174 /* Copy at most dst_buff_size bytes of the message to receiver's buffer */
1175 if (comm->comm.dst_buff_size)
1176 buff_size = MIN(buff_size, *(comm->comm.dst_buff_size));
1178 /* Update the receiver's buffer size to the copied amount */
1179 if (comm->comm.dst_buff_size)
1180 *comm->comm.dst_buff_size = buff_size;
1183 if(comm->comm.copy_data_fun)
1184 comm->comm.copy_data_fun (comm, comm->comm.src_buff, buff_size);
1186 SIMIX_comm_copy_data_callback (comm, comm->comm.src_buff, buff_size);
1190 /* Set the copied flag so we copy data only once */
1191 /* (this function might be called from both communication ends) */
1192 comm->comm.copied = 1;