Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Remove SIMIX requests that get src and dst buffers and their sizes
[simgrid.git] / src / simix / smx_network.c
1 /* Copyright (c) 2009, 2010. The SimGrid Team.
2  * All rights reserved.                                                     */
3
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. */
6
7 #include "private.h"
8 #include "xbt/log.h"
9 #include "mc/mc.h"
10 #include "xbt/dict.h"
11
12 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_network, simix,
13                                 "Logging specific to SIMIX (network)");
14
15 static xbt_dict_t rdv_points = NULL;
16
17 static XBT_INLINE void SIMIX_comm_start(smx_action_t action);
18 static void SIMIX_comm_finish(smx_action_t action);
19 static void SIMIX_waitany_req_remove_from_actions(smx_req_t req);
20 static void SIMIX_comm_copy_data(smx_action_t comm);
21 static smx_action_t SIMIX_comm_new(e_smx_comm_type_t type);
22 static XBT_INLINE void SIMIX_rdv_push(smx_rdv_t rdv, smx_action_t comm);
23 static XBT_INLINE void SIMIX_rdv_remove(smx_rdv_t rdv, smx_action_t comm);
24 static smx_action_t SIMIX_rdv_get_request(smx_rdv_t rdv, e_smx_comm_type_t type,
25                                           int (*match_fun)(void *, void *), void *);
26 static void SIMIX_rdv_free(void *data);
27
28 void SIMIX_network_init(void)
29 {
30   rdv_points = xbt_dict_new();
31 }
32
33 void SIMIX_network_exit(void)
34 {
35   xbt_dict_free(&rdv_points);
36 }
37
38 /******************************************************************************/
39 /*                           Rendez-Vous Points                               */
40 /******************************************************************************/
41
42 smx_rdv_t SIMIX_rdv_create(const char *name)
43 {
44   /* two processes may have pushed the same rdv_create request at the same time */
45   smx_rdv_t rdv = name ? xbt_dict_get_or_null(rdv_points, name) : NULL;
46
47   if (!rdv) {
48     rdv = xbt_new0(s_smx_rvpoint_t, 1);
49     rdv->name = name ? xbt_strdup(name) : NULL;
50     rdv->comm_fifo = xbt_fifo_new();
51
52     if (rdv->name)
53       xbt_dict_set(rdv_points, rdv->name, rdv, SIMIX_rdv_free);
54   }
55   return rdv;
56 }
57
58 void SIMIX_rdv_destroy(smx_rdv_t rdv)
59 {
60   if (rdv->name)
61     xbt_dict_remove(rdv_points, rdv->name);
62 }
63
64 void SIMIX_rdv_free(void *data)
65 {
66   smx_rdv_t rdv = (smx_rdv_t) data;
67   if (rdv->name)
68     xbt_free(rdv->name);
69   xbt_fifo_free(rdv->comm_fifo);
70   xbt_free(rdv);  
71 }
72
73 smx_rdv_t SIMIX_rdv_get_by_name(const char *name)
74 {
75   return xbt_dict_get_or_null(rdv_points, name);
76 }
77
78 int SIMIX_rdv_comm_count_by_host(smx_rdv_t rdv, smx_host_t host)
79 {
80   smx_action_t comm = NULL;
81   xbt_fifo_item_t item = NULL;
82   int count = 0;
83
84   xbt_fifo_foreach(rdv->comm_fifo, item, comm, smx_action_t) {
85     if (comm->comm.src_proc->smx_host == host)
86       count++;
87   }
88
89   return count;
90 }
91
92 smx_action_t SIMIX_rdv_get_head(smx_rdv_t rdv)
93 {
94   return xbt_fifo_get_item_content(xbt_fifo_get_first_item(rdv->comm_fifo));
95 }
96
97 /**
98  *  \brief Push a communication request into a rendez-vous point
99  *  \param rdv The rendez-vous point
100  *  \param comm The communication request
101  */
102 static XBT_INLINE void SIMIX_rdv_push(smx_rdv_t rdv, smx_action_t comm)
103 {
104   xbt_fifo_push(rdv->comm_fifo, comm);
105   comm->comm.rdv = rdv;
106 }
107
108 /**
109  *  \brief Remove a communication request from a rendez-vous point
110  *  \param rdv The rendez-vous point
111  *  \param comm The communication request
112  */
113 static XBT_INLINE void SIMIX_rdv_remove(smx_rdv_t rdv, smx_action_t comm)
114 {
115   xbt_fifo_remove(rdv->comm_fifo, comm);
116   comm->comm.rdv = NULL;
117 }
118
119 /**
120  *  \brief Checks if there is a communication action queued in a rendez-vous matching our needs
121  *  \param type The type of communication we are looking for (comm_send, comm_recv)
122  *  \return The communication action if found, NULL otherwise
123  */
124 smx_action_t SIMIX_rdv_get_request(smx_rdv_t rdv, e_smx_comm_type_t type,
125                                    int (*match_fun)(void *, void *), void *data)
126 {
127   smx_action_t action;
128   xbt_fifo_item_t item;
129   void* req_data = NULL;
130
131   xbt_fifo_foreach(rdv->comm_fifo, item, action, smx_action_t){
132     if (action->comm.type == SIMIX_COMM_SEND) {
133       req_data = action->comm.src_data;
134     } else if (action->comm.type == SIMIX_COMM_RECEIVE) {
135       req_data = action->comm.dst_data;
136     }
137     if (action->comm.type == type && (!match_fun || match_fun(data, req_data))) {
138       DEBUG1("Found a matching communication action %p", action);
139       xbt_fifo_remove_item(rdv->comm_fifo, item);
140       xbt_fifo_free_item(item);
141       action->comm.refcount++;
142       action->comm.rdv = NULL;
143       return action;
144     }
145     DEBUG3("Sorry, communication action %p does not match our needs:"
146            " its type is %d but we are looking for a comm of type %d",
147            action, action->comm.type, type);
148   }
149   DEBUG0("No matching communication action found");
150   return NULL;
151 }
152
153 /******************************************************************************/
154 /*                            Comunication Actions                            */
155 /******************************************************************************/
156
157 /**
158  *  \brief Creates a new comunicate action
159  *  \param type The type of request (comm_send, comm_recv)
160  *  \return The new comunicate action
161  */
162 smx_action_t SIMIX_comm_new(e_smx_comm_type_t type)
163 {
164   smx_action_t act;
165
166   /* alloc structures */
167   act = xbt_new0(s_smx_action_t, 1);
168   act->type = SIMIX_ACTION_COMMUNICATE;
169   act->state = SIMIX_WAITING;
170   act->request_list = xbt_fifo_new();
171
172   /* set communication */
173   act->comm.type = type;
174   act->comm.refcount = 1;
175
176 #ifdef HAVE_LATENCY_BOUND_TRACKING
177   //initialize with unknown value
178   act->latency_limited = -1;
179 #endif
180
181 #ifdef HAVE_TRACING
182   act->category = NULL;
183 #endif
184
185   DEBUG1("Create communicate action %p", act);
186
187   return act;
188 }
189
190 /**
191  *  \brief Destroy a communicate action
192  *  \param action The communicate action to be destroyed
193  */
194 void SIMIX_comm_destroy(smx_action_t action)
195 {
196   DEBUG2("Destroy action %p (refcount:%d)", action, action->comm.refcount);
197
198   if (action->comm.refcount <= 0)
199     xbt_die(bprintf("the refcount of comm %p is already 0 before decreasing it. That's a bug!",action));
200
201   action->comm.refcount--;
202   if (action->comm.refcount > 0)
203     return;
204   DEBUG2("Really free communication %p; refcount is now %d", action,
205         action->comm.refcount);
206
207 #ifdef HAVE_LATENCY_BOUND_TRACKING
208     action->latency_limited = SIMIX_comm_is_latency_bounded( action ) ;
209 #endif
210
211 #ifdef HAVE_TRACING
212   TRACE_smx_action_destroy(action);
213 #endif
214
215   if (action->name)
216     xbt_free(action->name);
217
218   xbt_fifo_free(action->request_list);
219
220   SIMIX_comm_destroy_internal_actions(action);
221
222   if (action->comm.detached && action->state != SIMIX_DONE) {
223     /* the communication has failed and was detached:
224      * we have to free the buffer */
225     ((void_f_pvoid_t) action->comm.src_data)(action->comm.src_buff);
226   }
227
228   xbt_free(action);
229 }
230
231 void SIMIX_comm_destroy_internal_actions(smx_action_t action)
232 {
233   if (action->comm.surf_comm){
234 #ifdef HAVE_LATENCY_BOUND_TRACKING
235     action->latency_limited = SIMIX_comm_is_latency_bounded(action);
236 #endif
237     action->comm.surf_comm->model_type->action_unref(action->comm.surf_comm);
238     action->comm.surf_comm = NULL;
239   }
240
241   if (action->comm.src_timeout){
242     action->comm.src_timeout->model_type->action_unref(action->comm.src_timeout);
243     action->comm.src_timeout = NULL;
244   }
245
246   if (action->comm.dst_timeout){
247     action->comm.dst_timeout->model_type->action_unref(action->comm.dst_timeout);
248     action->comm.dst_timeout = NULL;
249   }
250 }
251
252 smx_action_t SIMIX_comm_isend(smx_process_t src_proc, smx_rdv_t rdv,
253                               double task_size, double rate,
254                               void *src_buff, size_t src_buff_size,
255                               int (*match_fun)(void *, void *), void *data,
256                               int detached)
257 {
258   smx_action_t action;
259
260   /* Look for communication request matching our needs.
261      If it is not found then create it and push it into the rendez-vous point */
262   action = SIMIX_rdv_get_request(rdv, SIMIX_COMM_RECEIVE, match_fun, data);
263
264   if (!action) {
265     action = SIMIX_comm_new(SIMIX_COMM_SEND);
266     SIMIX_rdv_push(rdv, action);
267   } else {
268     action->state = SIMIX_READY;
269     action->comm.type = SIMIX_COMM_READY;
270   }
271
272   /* If the communication action is detached then decrease the refcount
273    * by one, so it will be eliminated by the receivers destroy call */
274   if (detached) {
275     action->comm.detached = 1;
276     action->comm.refcount--;
277   }
278
279   /* Setup the communication request */
280   action->comm.src_proc = src_proc;
281   action->comm.task_size = task_size;
282   action->comm.rate = rate;
283   action->comm.src_buff = src_buff;
284   action->comm.src_buff_size = src_buff_size;
285   action->comm.src_data = data;
286
287   if (MC_IS_ENABLED) {
288     action->state = SIMIX_RUNNING;
289     return action;
290   }
291
292   SIMIX_comm_start(action);
293   return action;
294 }
295
296 smx_action_t SIMIX_comm_irecv(smx_process_t dst_proc, smx_rdv_t rdv,
297                       void *dst_buff, size_t *dst_buff_size,
298                       int (*match_fun)(void *, void *), void *data)
299 {
300   smx_action_t action;
301
302   /* Look for communication request matching our needs.
303    * If it is not found then create it and push it into the rendez-vous point
304    */
305   action = SIMIX_rdv_get_request(rdv, SIMIX_COMM_SEND, match_fun, data);
306
307   if (!action) {
308     action = SIMIX_comm_new(SIMIX_COMM_RECEIVE);
309     SIMIX_rdv_push(rdv, action);
310   } else {
311     action->state = SIMIX_READY;
312     action->comm.type = SIMIX_COMM_READY;
313   }
314
315   /* Setup communication request */
316   action->comm.dst_proc = dst_proc;
317   action->comm.dst_buff = dst_buff;
318   action->comm.dst_buff_size = dst_buff_size;
319   action->comm.dst_data = data;
320
321   if (MC_IS_ENABLED) {
322     action->state = SIMIX_RUNNING;
323     return action;
324   }
325
326   SIMIX_comm_start(action);
327   return action;
328 }
329
330 void SIMIX_pre_comm_wait(smx_req_t req, int idx)
331 {
332   smx_action_t action = req->comm_wait.comm;
333   double timeout = req->comm_wait.timeout;
334   surf_action_t sleep;
335
336   /* Associate this request to the action */
337   xbt_fifo_push(action->request_list, req);
338   req->issuer->waiting_action = action;
339
340   if (MC_IS_ENABLED){
341     if(idx == 0){
342       action->state = SIMIX_DONE;
343     }else{
344       /* If we reached this point, the wait request must have a timeout */
345       /* Otherwise it shouldn't be enabled and executed by the MC */
346       if(timeout == -1)
347         THROW_IMPOSSIBLE;
348
349       if(action->comm.src_proc == req->issuer)
350         action->state = SIMIX_SRC_TIMEOUT;
351       else
352         action->state = SIMIX_DST_TIMEOUT;
353     }
354
355     SIMIX_comm_finish(action);
356     return;
357   }
358
359   /* If the action has already finish perform the error handling, */
360   /* otherwise set up a waiting timeout on the right side         */
361   if (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING) {
362     SIMIX_comm_finish(action);
363   } 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 */
364     sleep = surf_workstation_model->extension.workstation.sleep(req->issuer->smx_host->host, timeout);
365     surf_workstation_model->action_data_set(sleep, action);
366
367     if (req->issuer == action->comm.src_proc)
368       action->comm.src_timeout = sleep;
369     else
370       action->comm.dst_timeout = sleep;
371   }
372 }
373
374 void SIMIX_pre_comm_test(smx_req_t req)
375 {
376   smx_action_t action = req->comm_test.comm;
377
378   if(MC_IS_ENABLED){
379     req->comm_test.result = action->comm.src_proc && action->comm.dst_proc;
380     if(req->comm_test.result){
381       action->state = SIMIX_DONE;
382       xbt_fifo_push(action->request_list, req);
383       SIMIX_comm_finish(action);
384     }else{
385       SIMIX_request_answer(req);
386     }
387     return;
388   }
389
390   req->comm_test.result = (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING);
391   if (req->comm_test.result) {
392     xbt_fifo_push(action->request_list, req);
393     SIMIX_comm_finish(action);
394   } else {
395     SIMIX_request_answer(req);
396   }
397 }
398
399 void SIMIX_pre_comm_testany(smx_req_t req, int idx)
400 {
401   unsigned int cursor;
402   smx_action_t action;
403   xbt_dynar_t actions = req->comm_testany.comms;
404   req->comm_testany.result = -1;
405
406   if (MC_IS_ENABLED){
407     if(idx == -1){
408       SIMIX_request_answer(req);
409     }else{
410       action = xbt_dynar_get_as(actions, idx, smx_action_t);
411       req->comm_testany.result = idx;
412       xbt_fifo_push(action->request_list, req);
413       action->state = SIMIX_DONE;
414       SIMIX_comm_finish(action);
415     }
416     return;
417   }
418
419   xbt_dynar_foreach(req->comm_testany.comms,cursor,action) {
420     if (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING) {
421       req->comm_testany.result = cursor;
422       xbt_fifo_push(action->request_list, req);
423       SIMIX_comm_finish(action);
424       return;
425     }
426   }
427   SIMIX_request_answer(req);
428 }
429
430 void SIMIX_pre_comm_waitany(smx_req_t req, int idx)
431 {
432   smx_action_t action;
433   unsigned int cursor = 0;
434   xbt_dynar_t actions = req->comm_waitany.comms;
435
436   if (MC_IS_ENABLED){
437     action = xbt_dynar_get_as(actions, idx, smx_action_t);
438     xbt_fifo_push(action->request_list, req);
439     req->comm_waitany.result = idx;
440     action->state = SIMIX_DONE;
441     SIMIX_comm_finish(action);
442     return;
443   }
444
445   xbt_dynar_foreach(actions, cursor, action){
446     /* Associate this request to the action */
447     xbt_fifo_push(action->request_list, req);
448     if (action->state != SIMIX_WAITING && action->state != SIMIX_RUNNING){
449       SIMIX_comm_finish(action);
450       break;
451     }
452   }
453 }
454
455 void SIMIX_waitany_req_remove_from_actions(smx_req_t req)
456 {
457   smx_action_t action;
458   unsigned int cursor = 0;
459   xbt_dynar_t actions = req->comm_waitany.comms;
460
461   xbt_dynar_foreach(actions, cursor, action){
462     xbt_fifo_remove(action->request_list, req);
463   }
464 }
465
466 /**
467  *  \brief Start the simulation of a communication request
468  *  \param action The communication action
469  */
470 static XBT_INLINE void SIMIX_comm_start(smx_action_t action)
471 {
472   /* If both the sender and the receiver are already there, start the communication */
473   if (action->state == SIMIX_READY) {
474     smx_host_t sender = action->comm.src_proc->smx_host;
475     smx_host_t receiver = action->comm.dst_proc->smx_host;
476
477     DEBUG3("Starting communication %p from '%s' to '%s'", action,
478            SIMIX_host_get_name(sender), SIMIX_host_get_name(receiver));
479
480     action->comm.surf_comm = surf_workstation_model->extension.workstation.
481         communicate(sender->host, receiver->host, action->comm.task_size, action->comm.rate);
482
483     surf_workstation_model->action_data_set(action->comm.surf_comm, action);
484
485     action->state = SIMIX_RUNNING;
486
487 #ifdef HAVE_TRACING
488     TRACE_smx_action_communicate(action, action->comm.src_proc);
489 #endif
490
491     /* If a link is failed, detect it immediately */
492     if (surf_workstation_model->action_state_get(action->comm.surf_comm) == SURF_ACTION_FAILED) {
493       DEBUG2("Communication from '%s' to '%s' failed to start because of a link failure",
494           SIMIX_host_get_name(sender), SIMIX_host_get_name(receiver));
495       action->state = SIMIX_LINK_FAILURE;
496       SIMIX_comm_destroy_internal_actions(action);
497     }
498
499     /* If any of the process is suspend, create the action but stop its execution,
500        it will be restarted when the sender process resume */
501     if (SIMIX_process_is_suspended(action->comm.src_proc) ||
502         SIMIX_process_is_suspended(action->comm.dst_proc)) {
503       /* FIXME: check what should happen with the action state */
504       surf_workstation_model->suspend(action->comm.surf_comm);
505     }
506   }
507 }
508
509 void SIMIX_comm_finish(smx_action_t action)
510 {
511   unsigned int destroy_count = 0;
512   smx_req_t req;
513
514   while ((req = xbt_fifo_shift(action->request_list))) {
515
516     /* If a waitany request is waiting for this action to finish, then remove
517        it from the other actions in the waitany list. Afterwards, get the
518        position of the actual action in the waitany request's actions dynar and
519        return it as the result of the call */
520     if (req->call == REQ_COMM_WAITANY) {
521       SIMIX_waitany_req_remove_from_actions(req);
522       if(!MC_IS_ENABLED)
523         req->comm_waitany.result = xbt_dynar_search(req->comm_waitany.comms, &action);
524     }
525
526     /* If the action is still in a rendez-vous point then remove from it */
527     if (action->comm.rdv)
528       SIMIX_rdv_remove(action->comm.rdv, action);
529
530     DEBUG1("SIMIX_comm_finish: action state = %d", action->state);
531
532     /* Check out for errors */
533     switch (action->state) {
534
535       case SIMIX_DONE:
536         DEBUG1("Communication %p complete!", action);
537         SIMIX_comm_copy_data(action);
538         break;
539
540       case SIMIX_SRC_TIMEOUT:
541         TRY {
542           THROW0(timeout_error, 0, "Communication timeouted because of sender");
543         }
544         CATCH(req->issuer->running_ctx->exception) {
545           req->issuer->doexception = 1;
546         }
547         break;
548
549       case SIMIX_DST_TIMEOUT:
550         TRY {
551           THROW0(timeout_error, 0, "Communication timeouted because of receiver");
552         }
553         CATCH(req->issuer->running_ctx->exception) {
554           req->issuer->doexception = 1;
555         }
556         break;
557
558       case SIMIX_SRC_HOST_FAILURE:
559         TRY {
560           if (req->issuer == action->comm.src_proc)
561             THROW0(host_error, 0, "Host failed");
562           else
563             THROW0(network_error, 0, "Remote peer failed");
564         }
565         CATCH(req->issuer->running_ctx->exception) {
566           req->issuer->doexception = 1;
567         }
568         break;
569
570       case SIMIX_DST_HOST_FAILURE:
571         TRY {
572           if (req->issuer == action->comm.dst_proc)
573             THROW0(host_error, 0, "Host failed");
574           else
575             THROW0(network_error, 0, "Remote peer failed");
576         }
577         CATCH(req->issuer->running_ctx->exception) {
578           req->issuer->doexception = 1;
579         }
580         break;
581
582       case SIMIX_LINK_FAILURE:
583         TRY {
584           DEBUG5("Link failure in action %p between '%s' and '%s': posting an exception to the issuer: %s (%p)",
585               action, action->comm.src_proc->smx_host->name, action->comm.dst_proc->smx_host->name,
586               req->issuer->name, req->issuer);
587           THROW0(network_error, 0, "Link failure");
588         }
589         CATCH(req->issuer->running_ctx->exception) {
590           req->issuer->doexception = 1;
591         }
592         break;
593
594       default:
595         THROW_IMPOSSIBLE;
596     }
597
598     /* if there is an exception during a waitany or a testany, indicate the position of the failed communication */
599     if (req->issuer->doexception) {
600       if (req->call == REQ_COMM_WAITANY) {
601         req->issuer->running_ctx->exception.value = xbt_dynar_search(req->comm_waitany.comms, &action);
602       }
603       else if (req->call == REQ_COMM_TESTANY) {
604         req->issuer->running_ctx->exception.value = xbt_dynar_search(req->comm_testany.comms, &action);
605       }
606     }
607
608     req->issuer->waiting_action = NULL;
609     SIMIX_request_answer(req);
610     destroy_count++;
611   }
612
613   while(destroy_count-- > 0)
614     SIMIX_comm_destroy(action);
615 }
616
617 void SIMIX_post_comm(smx_action_t action)
618 {
619   /* Update action state */
620   if (action->comm.src_timeout &&
621      surf_workstation_model->action_state_get(action->comm.src_timeout) == SURF_ACTION_DONE)
622      action->state = SIMIX_SRC_TIMEOUT;
623   else if (action->comm.dst_timeout &&
624           surf_workstation_model->action_state_get(action->comm.dst_timeout) == SURF_ACTION_DONE)
625      action->state = SIMIX_DST_TIMEOUT;
626   else if (action->comm.src_timeout &&
627           surf_workstation_model->action_state_get(action->comm.src_timeout) == SURF_ACTION_FAILED)
628      action->state = SIMIX_SRC_HOST_FAILURE;
629   else if (action->comm.dst_timeout &&
630           surf_workstation_model->action_state_get(action->comm.dst_timeout) == SURF_ACTION_FAILED)
631      action->state = SIMIX_DST_HOST_FAILURE;
632   else if (action->comm.surf_comm &&
633           surf_workstation_model->action_state_get(action->comm.surf_comm) == SURF_ACTION_FAILED)
634      action->state = SIMIX_LINK_FAILURE;
635   else
636     action->state = SIMIX_DONE;
637
638   DEBUG1("SIMIX_post_comm: action state = %d", action->state);
639
640   /* After this point the surf actions associated with the simix communicate
641      action are no longer needed, thus we delete them. */
642   SIMIX_comm_destroy_internal_actions(action);
643
644   /* If there are requests associated with the action, then answer them */
645   if (xbt_fifo_size(action->request_list))
646     SIMIX_comm_finish(action);
647 }
648
649 void SIMIX_comm_cancel(smx_action_t action)
650 {
651   /* If the action is a waiting state means that it is still in a rdv */
652   /* so remove from it and delete it */
653   if (action->state == SIMIX_WAITING) {
654     SIMIX_rdv_remove(action->comm.rdv, action);
655     action->state = SIMIX_FAILED;
656   } else {
657     /* When running the MC there are no surf actions */
658     if(!MC_IS_ENABLED)
659       surf_workstation_model->action_cancel(action->comm.surf_comm);
660   }
661 }
662
663 void SIMIX_comm_suspend(smx_action_t action)
664 {
665   /*FIXME: shall we suspend also the timeout actions? */
666   surf_workstation_model->suspend(action->comm.surf_comm);
667 }
668
669 void SIMIX_comm_resume(smx_action_t action)
670 {
671   /*FIXME: check what happen with the timeouts */
672   surf_workstation_model->resume(action->comm.surf_comm);
673 }
674
675
676 /************* Action Getters **************/
677
678 /**
679  *  \brief get the amount remaining from the communication
680  *  \param action The communication
681  */
682 double SIMIX_comm_get_remains(smx_action_t action)
683 {
684   double remains;
685
686   switch (action->state) {
687
688     case SIMIX_RUNNING:
689       remains = surf_workstation_model->get_remains(action->comm.surf_comm);
690       break;
691
692     case SIMIX_WAITING:
693     case SIMIX_READY:
694       remains = 0; /*FIXME: check what should be returned */
695       break;
696
697     default:
698       remains = 0; /*FIXME: is this correct? */
699       break;
700   }
701   return remains;
702 }
703
704 e_smx_state_t SIMIX_comm_get_state(smx_action_t action)
705 {
706   return action->state;
707 }
708
709 /**
710  *  \brief Return the user data associated to the sender of the communication
711  *  \param action The communication
712  *  \return the user data
713  */
714 void* SIMIX_comm_get_src_data(smx_action_t action)
715 {
716   return action->comm.src_data;
717 }
718
719 /**
720  *  \brief Return the user data associated to the receiver of the communication
721  *  \param action The communication
722  *  \return the user data
723  */
724 void* SIMIX_comm_get_dst_data(smx_action_t action)
725 {
726   return action->comm.dst_data;
727 }
728
729 smx_process_t SIMIX_comm_get_src_proc(smx_action_t action)
730 {
731   return action->comm.src_proc;
732 }
733
734 smx_process_t SIMIX_comm_get_dst_proc(smx_action_t action)
735 {
736   return action->comm.dst_proc;
737 }
738
739 #ifdef HAVE_LATENCY_BOUND_TRACKING
740 /**
741  *  \brief verify if communication is latency bounded
742  *  \param comm The communication
743  */
744 XBT_INLINE int SIMIX_comm_is_latency_bounded(smx_action_t action)
745 {
746   if (action->comm.surf_comm){
747       DEBUG1("Getting latency limited for surf_action (%p)", action->comm.surf_comm);
748       action->latency_limited = surf_workstation_model->get_latency_limited(action->comm.surf_comm);
749       DEBUG1("Action limited is %d", action->latency_limited);
750   }
751   return action->latency_limited;
752 }
753 #endif
754
755 /******************************************************************************/
756 /*                    SIMIX_comm_copy_data callbacks                       */
757 /******************************************************************************/
758 static void (*SIMIX_comm_copy_data_callback) (smx_action_t, size_t) =
759     &SIMIX_comm_copy_pointer_callback;
760
761 void
762 SIMIX_comm_set_copy_data_callback(void (*callback) (smx_action_t, size_t))
763 {
764   SIMIX_comm_copy_data_callback = callback;
765 }
766
767 void SIMIX_comm_copy_pointer_callback(smx_action_t comm, size_t buff_size)
768 {
769   xbt_assert1((buff_size == sizeof(void *)),
770               "Cannot copy %zu bytes: must be sizeof(void*)", buff_size);
771   *(void **) (comm->comm.dst_buff) = comm->comm.src_buff;
772 }
773
774 void SIMIX_comm_copy_buffer_callback(smx_action_t comm, size_t buff_size)
775 {
776   memcpy(comm->comm.dst_buff, comm->comm.src_buff, buff_size);
777 }
778
779 /**
780  *  \brief Copy the communication data from the sender's buffer to the receiver's one
781  *  \param comm The communication
782  */
783 void SIMIX_comm_copy_data(smx_action_t comm)
784 {
785   size_t buff_size = comm->comm.src_buff_size;
786   /* If there is no data to be copy then return */
787   if (!comm->comm.src_buff || !comm->comm.dst_buff || comm->comm.copied == 1)
788     return;
789
790   DEBUG6("Copying comm %p data from %s (%p) -> %s (%p) (%zu bytes)",
791          comm,
792          comm->comm.src_proc->smx_host->name, comm->comm.src_buff,
793          comm->comm.dst_proc->smx_host->name, comm->comm.dst_buff, buff_size);
794
795   /* Copy at most dst_buff_size bytes of the message to receiver's buffer */
796   if (comm->comm.dst_buff_size)
797     buff_size = MIN(buff_size, *(comm->comm.dst_buff_size));
798
799   /* Update the receiver's buffer size to the copied amount */
800   if (comm->comm.dst_buff_size)
801     *comm->comm.dst_buff_size = buff_size;
802
803   if (buff_size == 0)
804     return;
805
806   (*SIMIX_comm_copy_data_callback) (comm, buff_size);
807
808   /* Set the copied flag so we copy data only once */
809   /* (this function might be called from both communication ends) */
810   comm->comm.copied = 1;
811 }