1 /* Copyright (c) 2009-2015. 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 "src/surf/surf_interface.hpp"
8 #include "src/simix/smx_private.h"
11 #include "src/mc/mc_replay.h"
14 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_network, simix, "SIMIX network-related synchronization");
16 static xbt_dict_t rdv_points = NULL;
17 XBT_EXPORT_NO_IMPORT(unsigned long int) smx_total_comms = 0;
19 static void SIMIX_waitany_remove_simcall_from_actions(smx_simcall_t simcall);
20 static void SIMIX_comm_copy_data(smx_synchro_t comm);
21 static smx_synchro_t SIMIX_comm_new(e_smx_comm_type_t type);
22 static inline void SIMIX_rdv_push(smx_mailbox_t rdv, smx_synchro_t comm);
23 static smx_synchro_t SIMIX_fifo_probe_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
24 int (*match_fun)(void *, void *,smx_synchro_t),
25 void *user_data, smx_synchro_t my_synchro);
26 static smx_synchro_t SIMIX_fifo_get_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
27 int (*match_fun)(void *, void *,smx_synchro_t),
28 void *user_data, smx_synchro_t my_synchro);
29 static void SIMIX_rdv_free(void *data);
30 static void SIMIX_comm_start(smx_synchro_t synchro);
32 void SIMIX_network_init(void)
34 rdv_points = xbt_dict_new_homogeneous(SIMIX_rdv_free);
37 void SIMIX_network_exit(void)
39 xbt_dict_free(&rdv_points);
42 /******************************************************************************/
43 /* Rendez-Vous Points */
44 /******************************************************************************/
46 smx_mailbox_t SIMIX_rdv_create(const char *name)
48 /* two processes may have pushed the same rdv_create simcall at the same time */
49 smx_mailbox_t rdv = name ? (smx_mailbox_t) xbt_dict_get_or_null(rdv_points, name) : NULL;
52 rdv = xbt_new0(s_smx_rvpoint_t, 1);
53 rdv->name = name ? xbt_strdup(name) : NULL;
54 rdv->comm_fifo = xbt_fifo_new();
55 rdv->done_comm_fifo = xbt_fifo_new();
56 rdv->permanent_receiver=NULL;
58 XBT_DEBUG("Creating a mailbox at %p with name %s", rdv, name);
61 xbt_dict_set(rdv_points, rdv->name, rdv, NULL);
66 void SIMIX_rdv_destroy(smx_mailbox_t rdv)
69 xbt_dict_remove(rdv_points, rdv->name);
72 void SIMIX_rdv_free(void *data)
74 XBT_DEBUG("rdv free %p", data);
75 smx_mailbox_t rdv = (smx_mailbox_t) data;
77 xbt_fifo_free(rdv->comm_fifo);
78 xbt_fifo_free(rdv->done_comm_fifo);
83 xbt_dict_t SIMIX_get_rdv_points()
88 smx_mailbox_t SIMIX_rdv_get_by_name(const char *name)
90 return (smx_mailbox_t) xbt_dict_get_or_null(rdv_points, name);
93 int SIMIX_rdv_comm_count_by_host(smx_mailbox_t rdv, sg_host_t host)
95 smx_synchro_t comm = NULL;
96 xbt_fifo_item_t item = NULL;
99 xbt_fifo_foreach(rdv->comm_fifo, item, comm, smx_synchro_t) {
100 if (comm->comm.src_proc->host == host)
107 smx_synchro_t SIMIX_rdv_get_head(smx_mailbox_t rdv)
109 return (smx_synchro_t) xbt_fifo_get_item_content(
110 xbt_fifo_get_first_item(rdv->comm_fifo));
114 * \brief get the receiver (process associated to the mailbox)
115 * \param rdv The rendez-vous point
116 * \return process The receiving process (NULL if not set)
118 smx_process_t SIMIX_rdv_get_receiver(smx_mailbox_t rdv)
120 return rdv->permanent_receiver;
124 * \brief set the receiver of the rendez vous point to allow eager sends
125 * \param rdv The rendez-vous point
126 * \param process The receiving process
128 void SIMIX_rdv_set_receiver(smx_mailbox_t rdv, smx_process_t process)
130 rdv->permanent_receiver=process;
134 * \brief Pushes a communication synchro into a rendez-vous point
135 * \param rdv The rendez-vous point
136 * \param comm The communication synchro
138 static inline void SIMIX_rdv_push(smx_mailbox_t rdv, smx_synchro_t comm)
140 xbt_fifo_push(rdv->comm_fifo, comm);
141 comm->comm.rdv = rdv;
145 * \brief Removes a communication synchro from a rendez-vous point
146 * \param rdv The rendez-vous point
147 * \param comm The communication synchro
149 void SIMIX_rdv_remove(smx_mailbox_t rdv, smx_synchro_t comm)
151 xbt_fifo_remove(rdv->comm_fifo, comm);
152 comm->comm.rdv = NULL;
156 * \brief Checks if there is a communication synchro queued in a fifo matching our needs
157 * \param type The type of communication we are looking for (comm_send, comm_recv)
158 * \return The communication synchro if found, NULL otherwise
160 smx_synchro_t SIMIX_fifo_get_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
161 int (*match_fun)(void *, void *,smx_synchro_t),
162 void *this_user_data, smx_synchro_t my_synchro)
164 smx_synchro_t synchro;
165 xbt_fifo_item_t item;
166 void* other_user_data = NULL;
168 xbt_fifo_foreach(fifo, item, synchro, smx_synchro_t) {
169 if (synchro->comm.type == SIMIX_COMM_SEND) {
170 other_user_data = synchro->comm.src_data;
171 } else if (synchro->comm.type == SIMIX_COMM_RECEIVE) {
172 other_user_data = synchro->comm.dst_data;
174 if (synchro->comm.type == type &&
175 (!match_fun || match_fun(this_user_data, other_user_data, synchro)) &&
176 (!synchro->comm.match_fun || synchro->comm.match_fun(other_user_data, this_user_data, my_synchro))) {
177 XBT_DEBUG("Found a matching communication synchro %p", synchro);
178 xbt_fifo_remove_item(fifo, item);
179 xbt_fifo_free_item(item);
180 synchro->comm.refcount++;
182 synchro->comm.rdv_cpy = synchro->comm.rdv;
184 synchro->comm.rdv = NULL;
187 XBT_DEBUG("Sorry, communication synchro %p does not match our needs:"
188 " its type is %d but we are looking for a comm of type %d (or maybe the filtering didn't match)",
189 synchro, (int)synchro->comm.type, (int)type);
191 XBT_DEBUG("No matching communication synchro found");
197 * \brief Checks if there is a communication synchro queued in a fifo matching our needs, but leave it there
198 * \param type The type of communication we are looking for (comm_send, comm_recv)
199 * \return The communication synchro if found, NULL otherwise
201 smx_synchro_t SIMIX_fifo_probe_comm(xbt_fifo_t fifo, e_smx_comm_type_t type,
202 int (*match_fun)(void *, void *,smx_synchro_t),
203 void *this_user_data, smx_synchro_t my_synchro)
205 smx_synchro_t synchro;
206 xbt_fifo_item_t item;
207 void* other_user_data = NULL;
209 xbt_fifo_foreach(fifo, item, synchro, smx_synchro_t) {
210 if (synchro->comm.type == SIMIX_COMM_SEND) {
211 other_user_data = synchro->comm.src_data;
212 } else if (synchro->comm.type == SIMIX_COMM_RECEIVE) {
213 other_user_data = synchro->comm.dst_data;
215 if (synchro->comm.type == type &&
216 (!match_fun || match_fun(this_user_data, other_user_data, synchro)) &&
217 (!synchro->comm.match_fun || synchro->comm.match_fun(other_user_data, this_user_data, my_synchro))) {
218 XBT_DEBUG("Found a matching communication synchro %p", synchro);
219 synchro->comm.refcount++;
223 XBT_DEBUG("Sorry, communication synchro %p does not match our needs:"
224 " its type is %d but we are looking for a comm of type %d (or maybe the filtering didn't match)",
225 synchro, (int)synchro->comm.type, (int)type);
227 XBT_DEBUG("No matching communication synchro found");
230 /******************************************************************************/
231 /* Communication synchros */
232 /******************************************************************************/
235 * \brief Creates a new communicate synchro
236 * \param type The direction of communication (comm_send, comm_recv)
237 * \return The new communicate synchro
239 smx_synchro_t SIMIX_comm_new(e_smx_comm_type_t type)
241 smx_synchro_t synchro;
243 /* alloc structures */
244 synchro = (smx_synchro_t) xbt_mallocator_get(simix_global->synchro_mallocator);
246 synchro->type = SIMIX_SYNC_COMMUNICATE;
247 synchro->state = SIMIX_WAITING;
249 /* set communication */
250 synchro->comm.type = type;
251 synchro->comm.refcount = 1;
252 synchro->comm.src_data=NULL;
253 synchro->comm.dst_data=NULL;
256 #ifdef HAVE_LATENCY_BOUND_TRACKING
257 //initialize with unknown value
258 synchro->latency_limited = -1;
261 synchro->category = NULL;
263 XBT_DEBUG("Create communicate synchro %p", synchro);
270 * \brief Destroy a communicate synchro
271 * \param synchro The communicate synchro to be destroyed
273 void SIMIX_comm_destroy(smx_synchro_t synchro)
275 XBT_DEBUG("Destroy synchro %p (refcount: %d), state: %d",
276 synchro, synchro->comm.refcount, (int)synchro->state);
278 if (synchro->comm.refcount <= 0) {
279 xbt_backtrace_display_current();
280 xbt_die("The refcount of comm %p is already 0 before decreasing it. "
281 "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...", synchro);
283 synchro->comm.refcount--;
284 if (synchro->comm.refcount > 0)
286 XBT_DEBUG("Really free communication %p; refcount is now %d", synchro,
287 synchro->comm.refcount);
289 #ifdef HAVE_LATENCY_BOUND_TRACKING
290 synchro->latency_limited = SIMIX_comm_is_latency_bounded( synchro ) ;
293 xbt_free(synchro->name);
294 SIMIX_comm_destroy_internal_actions(synchro);
296 if (synchro->comm.detached && synchro->state != SIMIX_DONE) {
297 /* the communication has failed and was detached:
298 * we have to free the buffer */
299 if (synchro->comm.clean_fun) {
300 synchro->comm.clean_fun(synchro->comm.src_buff);
302 synchro->comm.src_buff = NULL;
305 if(synchro->comm.rdv)
306 SIMIX_rdv_remove(synchro->comm.rdv, synchro);
308 xbt_mallocator_release(simix_global->synchro_mallocator, synchro);
311 void SIMIX_comm_destroy_internal_actions(smx_synchro_t synchro)
313 if (synchro->comm.surf_comm){
314 #ifdef HAVE_LATENCY_BOUND_TRACKING
315 synchro->latency_limited = SIMIX_comm_is_latency_bounded(synchro);
317 synchro->comm.surf_comm->unref();
318 synchro->comm.surf_comm = NULL;
321 if (synchro->comm.src_timeout){
322 synchro->comm.src_timeout->unref();
323 synchro->comm.src_timeout = NULL;
326 if (synchro->comm.dst_timeout){
327 synchro->comm.dst_timeout->unref();
328 synchro->comm.dst_timeout = NULL;
332 void simcall_HANDLER_comm_send(smx_simcall_t simcall, smx_process_t src, smx_mailbox_t rdv,
333 double task_size, double rate,
334 void *src_buff, size_t src_buff_size,
335 int (*match_fun)(void *, void *,smx_synchro_t),
336 void (*copy_data_fun)(smx_synchro_t, void*, size_t),
337 void *data, double timeout){
338 smx_synchro_t comm = simcall_HANDLER_comm_isend(simcall, src, rdv, task_size, rate,
339 src_buff, src_buff_size, match_fun, NULL, copy_data_fun,
341 SIMCALL_SET_MC_VALUE(simcall, 0);
342 simcall_HANDLER_comm_wait(simcall, comm, timeout);
344 smx_synchro_t simcall_HANDLER_comm_isend(smx_simcall_t simcall, smx_process_t src_proc, smx_mailbox_t rdv,
345 double task_size, double rate,
346 void *src_buff, size_t src_buff_size,
347 int (*match_fun)(void *, void *,smx_synchro_t),
348 void (*clean_fun)(void *), // used to free the synchro in case of problem after a detached send
349 void (*copy_data_fun)(smx_synchro_t, void*, size_t),// used to copy data if not default one
350 void *data, int detached)
352 XBT_DEBUG("send from %p", rdv);
354 /* Prepare a synchro describing us, so that it gets passed to the user-provided filter of other side */
355 smx_synchro_t this_synchro = SIMIX_comm_new(SIMIX_COMM_SEND);
357 /* Look for communication synchro matching our needs. We also provide a description of
358 * ourself so that the other side also gets a chance of choosing if it wants to match with us.
360 * If it is not found then push our communication into the rendez-vous point */
361 smx_synchro_t other_synchro = SIMIX_fifo_get_comm(rdv->comm_fifo, SIMIX_COMM_RECEIVE, match_fun, data, this_synchro);
363 if (!other_synchro) {
364 other_synchro = this_synchro;
366 if (rdv->permanent_receiver!=NULL){
367 //this mailbox is for small messages, which have to be sent right now
368 other_synchro->state = SIMIX_READY;
369 other_synchro->comm.dst_proc=rdv->permanent_receiver;
370 other_synchro->comm.refcount++;
371 xbt_fifo_push(rdv->done_comm_fifo,other_synchro);
372 other_synchro->comm.rdv=rdv;
373 XBT_DEBUG("pushing a message into the permanent receive fifo %p, comm %p", rdv, &(other_synchro->comm));
376 SIMIX_rdv_push(rdv, this_synchro);
379 XBT_DEBUG("Receive already pushed");
381 SIMIX_comm_destroy(this_synchro);
382 --smx_total_comms; // this creation was a pure waste
384 other_synchro->state = SIMIX_READY;
385 other_synchro->comm.type = SIMIX_COMM_READY;
388 xbt_fifo_push(src_proc->comms, other_synchro);
390 /* if the communication synchro is detached then decrease the refcount
391 * by one, so it will be eliminated by the receiver's destroy call */
393 other_synchro->comm.detached = 1;
394 other_synchro->comm.refcount--;
395 other_synchro->comm.clean_fun = clean_fun;
397 other_synchro->comm.clean_fun = NULL;
400 /* Setup the communication synchro */
401 other_synchro->comm.src_proc = src_proc;
402 other_synchro->comm.task_size = task_size;
403 other_synchro->comm.rate = rate;
404 other_synchro->comm.src_buff = src_buff;
405 other_synchro->comm.src_buff_size = src_buff_size;
406 other_synchro->comm.src_data = data;
408 other_synchro->comm.match_fun = match_fun;
409 other_synchro->comm.copy_data_fun = copy_data_fun;
412 if (MC_is_active() || MC_record_replay_is_active()) {
413 other_synchro->state = SIMIX_RUNNING;
414 return (detached ? NULL : other_synchro);
417 SIMIX_comm_start(other_synchro);
418 return (detached ? NULL : other_synchro);
421 void simcall_HANDLER_comm_recv(smx_simcall_t simcall, smx_process_t receiver, smx_mailbox_t rdv,
422 void *dst_buff, size_t *dst_buff_size,
423 int (*match_fun)(void *, void *, smx_synchro_t),
424 void (*copy_data_fun)(smx_synchro_t, void*, size_t),
425 void *data, double timeout, double rate)
427 smx_synchro_t comm = SIMIX_comm_irecv(receiver, rdv, dst_buff,
428 dst_buff_size, match_fun, copy_data_fun, data, rate);
429 SIMCALL_SET_MC_VALUE(simcall, 0);
430 simcall_HANDLER_comm_wait(simcall, comm, timeout);
433 smx_synchro_t simcall_HANDLER_comm_irecv(smx_simcall_t simcall, smx_process_t receiver, smx_mailbox_t rdv,
434 void *dst_buff, size_t *dst_buff_size,
435 int (*match_fun)(void *, void *, smx_synchro_t),
436 void (*copy_data_fun)(smx_synchro_t, void*, size_t),
437 void *data, double rate)
439 return SIMIX_comm_irecv(receiver, rdv, dst_buff, dst_buff_size, match_fun, copy_data_fun, data, rate);
442 smx_synchro_t SIMIX_comm_irecv(smx_process_t dst_proc, smx_mailbox_t rdv, void *dst_buff, size_t *dst_buff_size,
443 int (*match_fun)(void *, void *, smx_synchro_t),
444 void (*copy_data_fun)(smx_synchro_t, void*, size_t), // used to copy data if not default one
445 void *data, double rate)
447 XBT_DEBUG("recv from %p %p", rdv, rdv->comm_fifo);
448 smx_synchro_t this_synchro = SIMIX_comm_new(SIMIX_COMM_RECEIVE);
450 smx_synchro_t other_synchro;
451 //communication already done, get it inside the fifo of completed comms
452 if (rdv->permanent_receiver && xbt_fifo_size(rdv->done_comm_fifo)!=0) {
454 XBT_DEBUG("We have a comm that has probably already been received, trying to match it, to skip the communication");
455 //find a match in the already received fifo
456 other_synchro = SIMIX_fifo_get_comm(rdv->done_comm_fifo, SIMIX_COMM_SEND, match_fun, data, this_synchro);
457 //if not found, assume the receiver came first, register it to the mailbox in the classical way
458 if (!other_synchro) {
459 XBT_DEBUG("We have messages in the permanent receive list, but not the one we are looking for, pushing request into fifo");
460 other_synchro = this_synchro;
461 SIMIX_rdv_push(rdv, this_synchro);
463 if(other_synchro->comm.surf_comm && SIMIX_comm_get_remains(other_synchro)==0.0) {
464 XBT_DEBUG("comm %p has been already sent, and is finished, destroy it",&(other_synchro->comm));
465 other_synchro->state = SIMIX_DONE;
466 other_synchro->comm.type = SIMIX_COMM_DONE;
467 other_synchro->comm.rdv = NULL;
469 other_synchro->comm.refcount--;
470 SIMIX_comm_destroy(this_synchro);
471 --smx_total_comms; // this creation was a pure waste
474 /* Prepare a synchro describing us, so that it gets passed to the user-provided filter of other side */
476 /* Look for communication synchro matching our needs. We also provide a description of
477 * ourself so that the other side also gets a chance of choosing if it wants to match with us.
479 * If it is not found then push our communication into the rendez-vous point */
480 other_synchro = SIMIX_fifo_get_comm(rdv->comm_fifo, SIMIX_COMM_SEND, match_fun, data, this_synchro);
482 if (!other_synchro) {
483 XBT_DEBUG("Receive pushed first %d", xbt_fifo_size(rdv->comm_fifo));
484 other_synchro = this_synchro;
485 SIMIX_rdv_push(rdv, this_synchro);
487 SIMIX_comm_destroy(this_synchro);
488 --smx_total_comms; // this creation was a pure waste
489 other_synchro->state = SIMIX_READY;
490 other_synchro->comm.type = SIMIX_COMM_READY;
491 //other_synchro->comm.refcount--;
493 xbt_fifo_push(dst_proc->comms, other_synchro);
496 /* Setup communication synchro */
497 other_synchro->comm.dst_proc = dst_proc;
498 other_synchro->comm.dst_buff = dst_buff;
499 other_synchro->comm.dst_buff_size = dst_buff_size;
500 other_synchro->comm.dst_data = data;
502 if (rate != -1.0 && (other_synchro->comm.rate == -1.0 || rate < other_synchro->comm.rate))
503 other_synchro->comm.rate = rate;
505 other_synchro->comm.match_fun = match_fun;
506 other_synchro->comm.copy_data_fun = copy_data_fun;
508 if (MC_is_active() || MC_record_replay_is_active()) {
509 other_synchro->state = SIMIX_RUNNING;
510 return other_synchro;
513 SIMIX_comm_start(other_synchro);
514 return other_synchro;
517 smx_synchro_t simcall_HANDLER_comm_iprobe(smx_simcall_t simcall, smx_mailbox_t rdv,
518 int type, int src, int tag,
519 int (*match_fun)(void *, void *, smx_synchro_t),
521 return SIMIX_comm_iprobe(simcall->issuer, rdv, type, src, tag, match_fun, data);
524 smx_synchro_t SIMIX_comm_iprobe(smx_process_t dst_proc, smx_mailbox_t rdv, int type, int src,
525 int tag, int (*match_fun)(void *, void *, smx_synchro_t), void *data)
527 XBT_DEBUG("iprobe from %p %p", rdv, rdv->comm_fifo);
528 smx_synchro_t this_synchro;
531 this_synchro=SIMIX_comm_new(SIMIX_COMM_SEND);
532 smx_type = SIMIX_COMM_RECEIVE;
534 this_synchro=SIMIX_comm_new(SIMIX_COMM_RECEIVE);
535 smx_type = SIMIX_COMM_SEND;
537 smx_synchro_t other_synchro=NULL;
538 if(rdv->permanent_receiver && xbt_fifo_size(rdv->done_comm_fifo)!=0){
539 //find a match in the already received fifo
540 XBT_DEBUG("first try in the perm recv mailbox");
542 other_synchro = SIMIX_fifo_probe_comm(
543 rdv->done_comm_fifo, (e_smx_comm_type_t) smx_type,
544 match_fun, data, this_synchro);
548 XBT_DEBUG("try in the normal mailbox");
549 other_synchro = SIMIX_fifo_probe_comm(
550 rdv->comm_fifo, (e_smx_comm_type_t) smx_type,
551 match_fun, data, this_synchro);
554 if(other_synchro)other_synchro->comm.refcount--;
556 SIMIX_comm_destroy(this_synchro);
558 return other_synchro;
561 void simcall_HANDLER_comm_wait(smx_simcall_t simcall, smx_synchro_t synchro, double timeout)
563 /* the simcall may be a wait, a send or a recv */
566 /* Associate this simcall to the wait synchro */
567 XBT_DEBUG("simcall_HANDLER_comm_wait, %p", synchro);
569 xbt_fifo_push(synchro->simcalls, simcall);
570 simcall->issuer->waiting_synchro = synchro;
572 if (MC_is_active() || MC_record_replay_is_active()) {
573 int idx = SIMCALL_GET_MC_VALUE(simcall);
575 synchro->state = SIMIX_DONE;
577 /* If we reached this point, the wait simcall must have a timeout */
578 /* Otherwise it shouldn't be enabled and executed by the MC */
582 if (synchro->comm.src_proc == simcall->issuer)
583 synchro->state = SIMIX_SRC_TIMEOUT;
585 synchro->state = SIMIX_DST_TIMEOUT;
588 SIMIX_comm_finish(synchro);
592 /* If the synchro has already finish perform the error handling, */
593 /* otherwise set up a waiting timeout on the right side */
594 if (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING) {
595 SIMIX_comm_finish(synchro);
596 } 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 */
597 sleep = surf_host_sleep(simcall->issuer->host, timeout);
598 sleep->setData(synchro);
600 if (simcall->issuer == synchro->comm.src_proc)
601 synchro->comm.src_timeout = sleep;
603 synchro->comm.dst_timeout = sleep;
607 void simcall_HANDLER_comm_test(smx_simcall_t simcall, smx_synchro_t synchro)
609 if(MC_is_active() || MC_record_replay_is_active()){
610 simcall_comm_test__set__result(simcall, synchro->comm.src_proc && synchro->comm.dst_proc);
611 if(simcall_comm_test__get__result(simcall)){
612 synchro->state = SIMIX_DONE;
613 xbt_fifo_push(synchro->simcalls, simcall);
614 SIMIX_comm_finish(synchro);
616 SIMIX_simcall_answer(simcall);
621 simcall_comm_test__set__result(simcall, (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING));
622 if (simcall_comm_test__get__result(simcall)) {
623 xbt_fifo_push(synchro->simcalls, simcall);
624 SIMIX_comm_finish(synchro);
626 SIMIX_simcall_answer(simcall);
630 void simcall_HANDLER_comm_testany(smx_simcall_t simcall, xbt_dynar_t synchros)
633 smx_synchro_t synchro;
634 simcall_comm_testany__set__result(simcall, -1);
636 if (MC_is_active() || MC_record_replay_is_active()){
637 int idx = SIMCALL_GET_MC_VALUE(simcall);
639 SIMIX_simcall_answer(simcall);
641 synchro = xbt_dynar_get_as(synchros, idx, smx_synchro_t);
642 simcall_comm_testany__set__result(simcall, idx);
643 xbt_fifo_push(synchro->simcalls, simcall);
644 synchro->state = SIMIX_DONE;
645 SIMIX_comm_finish(synchro);
650 xbt_dynar_foreach(simcall_comm_testany__get__comms(simcall), cursor,synchro) {
651 if (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING) {
652 simcall_comm_testany__set__result(simcall, cursor);
653 xbt_fifo_push(synchro->simcalls, simcall);
654 SIMIX_comm_finish(synchro);
658 SIMIX_simcall_answer(simcall);
661 void simcall_HANDLER_comm_waitany(smx_simcall_t simcall, xbt_dynar_t synchros)
663 smx_synchro_t synchro;
664 unsigned int cursor = 0;
666 if (MC_is_active() || MC_record_replay_is_active()){
667 int idx = SIMCALL_GET_MC_VALUE(simcall);
668 synchro = xbt_dynar_get_as(synchros, idx, smx_synchro_t);
669 xbt_fifo_push(synchro->simcalls, simcall);
670 simcall_comm_waitany__set__result(simcall, idx);
671 synchro->state = SIMIX_DONE;
672 SIMIX_comm_finish(synchro);
676 xbt_dynar_foreach(synchros, cursor, synchro){
677 /* associate this simcall to the the synchro */
678 xbt_fifo_push(synchro->simcalls, simcall);
680 /* see if the synchro is already finished */
681 if (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING){
682 SIMIX_comm_finish(synchro);
688 void SIMIX_waitany_remove_simcall_from_actions(smx_simcall_t simcall)
690 smx_synchro_t synchro;
691 unsigned int cursor = 0;
692 xbt_dynar_t synchros = simcall_comm_waitany__get__comms(simcall);
694 xbt_dynar_foreach(synchros, cursor, synchro) {
695 xbt_fifo_remove(synchro->simcalls, simcall);
700 * \brief Starts the simulation of a communication synchro.
701 * \param synchro the communication synchro
703 static inline void SIMIX_comm_start(smx_synchro_t synchro)
705 /* If both the sender and the receiver are already there, start the communication */
706 if (synchro->state == SIMIX_READY) {
708 sg_host_t sender = synchro->comm.src_proc->host;
709 sg_host_t receiver = synchro->comm.dst_proc->host;
711 XBT_DEBUG("Starting communication %p from '%s' to '%s'", synchro,
712 sg_host_get_name(sender), sg_host_get_name(receiver));
714 synchro->comm.surf_comm = surf_network_model_communicate(surf_network_model,
716 synchro->comm.task_size, synchro->comm.rate);
718 synchro->comm.surf_comm->setData(synchro);
720 synchro->state = SIMIX_RUNNING;
722 /* If a link is failed, detect it immediately */
723 if (synchro->comm.surf_comm->getState() == SURF_ACTION_FAILED) {
724 XBT_DEBUG("Communication from '%s' to '%s' failed to start because of a link failure",
725 sg_host_get_name(sender), sg_host_get_name(receiver));
726 synchro->state = SIMIX_LINK_FAILURE;
727 SIMIX_comm_destroy_internal_actions(synchro);
730 /* If any of the process is suspend, create the synchro but stop its execution,
731 it will be restarted when the sender process resume */
732 if (SIMIX_process_is_suspended(synchro->comm.src_proc) ||
733 SIMIX_process_is_suspended(synchro->comm.dst_proc)) {
734 /* FIXME: check what should happen with the synchro state */
736 if (SIMIX_process_is_suspended(synchro->comm.src_proc))
737 XBT_DEBUG("The communication is suspended on startup because src (%s:%s) were suspended since it initiated the communication",
738 sg_host_get_name(synchro->comm.src_proc->host), synchro->comm.src_proc->name);
740 XBT_DEBUG("The communication is suspended on startup because dst (%s:%s) were suspended since it initiated the communication",
741 sg_host_get_name(synchro->comm.dst_proc->host), synchro->comm.dst_proc->name);
743 synchro->comm.surf_comm->suspend();
750 * \brief Answers the SIMIX simcalls associated to a communication synchro.
751 * \param synchro a finished communication synchro
753 void SIMIX_comm_finish(smx_synchro_t synchro)
755 unsigned int destroy_count = 0;
756 smx_simcall_t simcall;
758 while ((simcall = (smx_simcall_t) xbt_fifo_shift(synchro->simcalls))) {
760 /* If a waitany simcall is waiting for this synchro to finish, then remove
761 it from the other synchros in the waitany list. Afterwards, get the
762 position of the actual synchro in the waitany dynar and
763 return it as the result of the simcall */
765 if (simcall->call == SIMCALL_NONE) //FIXME: maybe a better way to handle this case
766 continue; // if process handling comm is killed
767 if (simcall->call == SIMCALL_COMM_WAITANY) {
768 SIMIX_waitany_remove_simcall_from_actions(simcall);
769 if (!MC_is_active() && !MC_record_replay_is_active())
770 simcall_comm_waitany__set__result(simcall, xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &synchro));
773 /* If the synchro is still in a rendez-vous point then remove from it */
774 if (synchro->comm.rdv)
775 SIMIX_rdv_remove(synchro->comm.rdv, synchro);
777 XBT_DEBUG("SIMIX_comm_finish: synchro state = %d", (int)synchro->state);
779 /* Check out for errors */
781 if (simcall->issuer->host->isOff()) {
782 simcall->issuer->context->iwannadie = 1;
783 SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
786 switch (synchro->state) {
789 XBT_DEBUG("Communication %p complete!", synchro);
790 SIMIX_comm_copy_data(synchro);
793 case SIMIX_SRC_TIMEOUT:
794 SMX_EXCEPTION(simcall->issuer, timeout_error, 0,
795 "Communication timeouted because of sender");
798 case SIMIX_DST_TIMEOUT:
799 SMX_EXCEPTION(simcall->issuer, timeout_error, 0,
800 "Communication timeouted because of receiver");
803 case SIMIX_SRC_HOST_FAILURE:
804 if (simcall->issuer == synchro->comm.src_proc)
805 simcall->issuer->context->iwannadie = 1;
806 // SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
808 SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
811 case SIMIX_DST_HOST_FAILURE:
812 if (simcall->issuer == synchro->comm.dst_proc)
813 simcall->issuer->context->iwannadie = 1;
814 // SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
816 SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
819 case SIMIX_LINK_FAILURE:
821 XBT_DEBUG("Link failure in synchro %p between '%s' and '%s': posting an exception to the issuer: %s (%p) detached:%d",
823 synchro->comm.src_proc ? sg_host_get_name(synchro->comm.src_proc->host) : NULL,
824 synchro->comm.dst_proc ? sg_host_get_name(synchro->comm.dst_proc->host) : NULL,
825 simcall->issuer->name, simcall->issuer, synchro->comm.detached);
826 if (synchro->comm.src_proc == simcall->issuer) {
827 XBT_DEBUG("I'm source");
828 } else if (synchro->comm.dst_proc == simcall->issuer) {
829 XBT_DEBUG("I'm dest");
831 XBT_DEBUG("I'm neither source nor dest");
833 SMX_EXCEPTION(simcall->issuer, network_error, 0, "Link failure");
837 if (simcall->issuer == synchro->comm.dst_proc)
838 SMX_EXCEPTION(simcall->issuer, cancel_error, 0,
839 "Communication canceled by the sender");
841 SMX_EXCEPTION(simcall->issuer, cancel_error, 0,
842 "Communication canceled by the receiver");
846 xbt_die("Unexpected synchro state in SIMIX_comm_finish: %d", (int)synchro->state);
849 /* if there is an exception during a waitany or a testany, indicate the position of the failed communication */
850 if (simcall->issuer->doexception) {
851 if (simcall->call == SIMCALL_COMM_WAITANY) {
852 simcall->issuer->running_ctx->exception.value = xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &synchro);
854 else if (simcall->call == SIMCALL_COMM_TESTANY) {
855 simcall->issuer->running_ctx->exception.value = xbt_dynar_search(simcall_comm_testany__get__comms(simcall), &synchro);
859 if (simcall->issuer->host->isOff()) {
860 simcall->issuer->context->iwannadie = 1;
863 simcall->issuer->waiting_synchro = NULL;
864 xbt_fifo_remove(simcall->issuer->comms, synchro);
865 if(synchro->comm.detached){
866 if(simcall->issuer == synchro->comm.src_proc){
867 if(synchro->comm.dst_proc)
868 xbt_fifo_remove(synchro->comm.dst_proc->comms, synchro);
870 if(simcall->issuer == synchro->comm.dst_proc){
871 if(synchro->comm.src_proc)
872 xbt_fifo_remove(synchro->comm.src_proc->comms, synchro);
875 SIMIX_simcall_answer(simcall);
879 while (destroy_count-- > 0)
880 SIMIX_comm_destroy(synchro);
884 * \brief This function is called when a Surf communication synchro is finished.
885 * \param synchro the corresponding Simix communication
887 void SIMIX_post_comm(smx_synchro_t synchro)
889 /* Update synchro state */
890 if (synchro->comm.src_timeout &&
891 synchro->comm.src_timeout->getState() == SURF_ACTION_DONE)
892 synchro->state = SIMIX_SRC_TIMEOUT;
893 else if (synchro->comm.dst_timeout &&
894 synchro->comm.dst_timeout->getState() == SURF_ACTION_DONE)
895 synchro->state = SIMIX_DST_TIMEOUT;
896 else if (synchro->comm.src_timeout &&
897 synchro->comm.src_timeout->getState() == SURF_ACTION_FAILED)
898 synchro->state = SIMIX_SRC_HOST_FAILURE;
899 else if (synchro->comm.dst_timeout &&
900 synchro->comm.dst_timeout->getState() == SURF_ACTION_FAILED)
901 synchro->state = SIMIX_DST_HOST_FAILURE;
902 else if (synchro->comm.surf_comm &&
903 synchro->comm.surf_comm->getState() == SURF_ACTION_FAILED) {
904 XBT_DEBUG("Puta madre. Surf says that the link broke");
905 synchro->state = SIMIX_LINK_FAILURE;
907 synchro->state = SIMIX_DONE;
909 XBT_DEBUG("SIMIX_post_comm: comm %p, state %d, src_proc %p, dst_proc %p, detached: %d",
910 synchro, (int)synchro->state, synchro->comm.src_proc, synchro->comm.dst_proc, synchro->comm.detached);
912 /* destroy the surf actions associated with the Simix communication */
913 SIMIX_comm_destroy_internal_actions(synchro);
915 /* if there are simcalls associated with the synchro, then answer them */
916 if (xbt_fifo_size(synchro->simcalls)) {
917 SIMIX_comm_finish(synchro);
921 void SIMIX_comm_cancel(smx_synchro_t synchro)
923 /* if the synchro is a waiting state means that it is still in a rdv */
924 /* so remove from it and delete it */
925 if (synchro->state == SIMIX_WAITING) {
926 SIMIX_rdv_remove(synchro->comm.rdv, synchro);
927 synchro->state = SIMIX_CANCELED;
929 else if (!MC_is_active() /* when running the MC there are no surf actions */
930 && !MC_record_replay_is_active()
931 && (synchro->state == SIMIX_READY || synchro->state == SIMIX_RUNNING)) {
933 synchro->comm.surf_comm->cancel();
937 void SIMIX_comm_suspend(smx_synchro_t synchro)
939 /*FIXME: shall we suspend also the timeout synchro? */
940 if (synchro->comm.surf_comm)
941 synchro->comm.surf_comm->suspend();
942 /* in the other case, the action will be suspended on creation, in SIMIX_comm_start() */
945 void SIMIX_comm_resume(smx_synchro_t synchro)
947 /*FIXME: check what happen with the timeouts */
948 if (synchro->comm.surf_comm)
949 synchro->comm.surf_comm->resume();
950 /* in the other case, the synchro were not really suspended yet, see SIMIX_comm_suspend() and SIMIX_comm_start() */
954 /************* synchro Getters **************/
957 * \brief get the amount remaining from the communication
958 * \param synchro The communication
960 double SIMIX_comm_get_remains(smx_synchro_t synchro)
968 switch (synchro->state) {
971 remains = synchro->comm.surf_comm->getRemains();
976 remains = 0; /*FIXME: check what should be returned */
980 remains = 0; /*FIXME: is this correct? */
986 e_smx_state_t SIMIX_comm_get_state(smx_synchro_t synchro)
988 return synchro->state;
992 * \brief Return the user data associated to the sender of the communication
993 * \param synchro The communication
994 * \return the user data
996 void* SIMIX_comm_get_src_data(smx_synchro_t synchro)
998 return synchro->comm.src_data;
1002 * \brief Return the user data associated to the receiver of the communication
1003 * \param synchro The communication
1004 * \return the user data
1006 void* SIMIX_comm_get_dst_data(smx_synchro_t synchro)
1008 return synchro->comm.dst_data;
1011 smx_process_t SIMIX_comm_get_src_proc(smx_synchro_t synchro)
1013 return synchro->comm.src_proc;
1016 smx_process_t SIMIX_comm_get_dst_proc(smx_synchro_t synchro)
1018 return synchro->comm.dst_proc;
1021 #ifdef HAVE_LATENCY_BOUND_TRACKING
1023 * \brief verify if communication is latency bounded
1024 * \param comm The communication
1026 int SIMIX_comm_is_latency_bounded(smx_synchro_t synchro)
1031 if (synchro->comm.surf_comm){
1032 XBT_DEBUG("Getting latency limited for surf_action (%p)", synchro->comm.surf_comm);
1033 synchro->latency_limited = surf_network_action_get_latency_limited(synchro->comm.surf_comm);
1034 XBT_DEBUG("synchro limited is %d", synchro->latency_limited);
1036 return synchro->latency_limited;
1040 /******************************************************************************/
1041 /* SIMIX_comm_copy_data callbacks */
1042 /******************************************************************************/
1043 static void (*SIMIX_comm_copy_data_callback) (smx_synchro_t, void*, size_t) =
1044 &SIMIX_comm_copy_pointer_callback;
1047 SIMIX_comm_set_copy_data_callback(void (*callback) (smx_synchro_t, void*, size_t))
1049 SIMIX_comm_copy_data_callback = callback;
1052 void SIMIX_comm_copy_pointer_callback(smx_synchro_t comm, void* buff, size_t buff_size)
1054 xbt_assert((buff_size == sizeof(void *)),
1055 "Cannot copy %zu bytes: must be sizeof(void*)", buff_size);
1056 *(void **) (comm->comm.dst_buff) = buff;
1059 void SIMIX_comm_copy_buffer_callback(smx_synchro_t comm, void* buff, size_t buff_size)
1061 XBT_DEBUG("Copy the data over");
1062 memcpy(comm->comm.dst_buff, buff, buff_size);
1063 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
1065 comm->comm.src_buff = NULL;
1071 * \brief Copy the communication data from the sender's buffer to the receiver's one
1072 * \param comm The communication
1074 void SIMIX_comm_copy_data(smx_synchro_t comm)
1076 size_t buff_size = comm->comm.src_buff_size;
1077 /* If there is no data to be copy then return */
1078 if (!comm->comm.src_buff || !comm->comm.dst_buff || comm->comm.copied)
1081 XBT_DEBUG("Copying comm %p data from %s (%p) -> %s (%p) (%zu bytes)",
1083 comm->comm.src_proc ? sg_host_get_name(comm->comm.src_proc->host) : "a finished process",
1084 comm->comm.src_buff,
1085 comm->comm.dst_proc ? sg_host_get_name(comm->comm.dst_proc->host) : "a finished process",
1086 comm->comm.dst_buff, buff_size);
1088 /* Copy at most dst_buff_size bytes of the message to receiver's buffer */
1089 if (comm->comm.dst_buff_size)
1090 buff_size = MIN(buff_size, *(comm->comm.dst_buff_size));
1092 /* Update the receiver's buffer size to the copied amount */
1093 if (comm->comm.dst_buff_size)
1094 *comm->comm.dst_buff_size = buff_size;
1097 if(comm->comm.copy_data_fun)
1098 comm->comm.copy_data_fun (comm, comm->comm.src_buff, buff_size);
1100 SIMIX_comm_copy_data_callback (comm, comm->comm.src_buff, buff_size);
1104 /* Set the copied flag so we copy data only once */
1105 /* (this function might be called from both communication ends) */
1106 comm->comm.copied = 1;