Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Merge remote-tracking branch 'adfaure/master'
[simgrid.git] / src / s4u / s4u_comm.cpp
1 /* Copyright (c) 2006-2015. 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 "xbt/log.h"
8 #include "src/msg/msg_private.h"
9
10 #include "simgrid/s4u/comm.hpp"
11
12 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(s4u_comm,s4u_activity,"S4U asynchronous communications");
13
14 namespace simgrid {
15 namespace s4u {
16
17 Comm::~Comm() {
18
19 }
20
21 s4u::Comm &Comm::send_init(s4u::Mailbox &chan) {
22   s4u::Comm *res = new s4u::Comm();
23   res->sender_ = SIMIX_process_self();
24   res->mailbox_ = &chan;
25   return *res;
26 }
27
28 s4u::Comm &Comm::recv_init(s4u::Mailbox &chan) {
29   s4u::Comm *res = new s4u::Comm();
30   res->receiver_ = SIMIX_process_self();
31   res->mailbox_ = &chan;
32   return *res;
33 }
34
35 void Comm::setRate(double rate) {
36   xbt_assert(state_==inited);
37   rate_ = rate;
38 }
39
40 void Comm::setSrcData(void * buff) {
41   xbt_assert(state_==inited);
42   xbt_assert(dstBuff_ == nullptr, "Cannot set the src and dst buffers at the same time");
43   srcBuff_ = buff;
44 }
45 void Comm::setSrcDataSize(size_t size){
46   xbt_assert(state_==inited);
47   srcBuffSize_ = size;
48 }
49 void Comm::setSrcData(void * buff, size_t size) {
50   xbt_assert(state_==inited);
51
52   xbt_assert(dstBuff_ == nullptr, "Cannot set the src and dst buffers at the same time");
53   srcBuff_ = buff;
54   srcBuffSize_ = size;
55 }
56 void Comm::setDstData(void ** buff) {
57   xbt_assert(state_==inited);
58   xbt_assert(srcBuff_ == nullptr, "Cannot set the src and dst buffers at the same time");
59   dstBuff_ = buff;
60 }
61 size_t Comm::getDstDataSize(){
62   xbt_assert(state_==finished);
63   return dstBuffSize_;
64 }
65 void Comm::setDstData(void ** buff, size_t size) {
66   xbt_assert(state_==inited);
67
68   xbt_assert(srcBuff_ == nullptr, "Cannot set the src and dst buffers at the same time");
69   dstBuff_ = buff;
70   dstBuffSize_ = size;
71 }
72
73 void Comm::start() {
74   xbt_assert(state_ == inited);
75
76   if (srcBuff_ != nullptr) { // Sender side
77     pimpl_ = simcall_comm_isend(sender_, mailbox_->getInferior(), remains_, rate_,
78         srcBuff_, srcBuffSize_,
79         matchFunction_, cleanFunction_, copyDataFunction_,
80         userData_, detached_);
81   } else if (dstBuff_ != nullptr) { // Receiver side
82     pimpl_ = simcall_comm_irecv(receiver_, mailbox_->getInferior(), dstBuff_, &dstBuffSize_,
83         matchFunction_, copyDataFunction_,
84         userData_, rate_);
85
86   } else {
87     xbt_die("Cannot start a communication before specifying whether we are the sender or the receiver");
88   }
89   state_ = started;
90 }
91 void Comm::wait() {
92   xbt_assert(state_ == started || state_ == inited);
93
94   if (state_ == started)
95     simcall_comm_wait(pimpl_, -1/*timeout*/);
96   else {// p_state == inited. Save a simcall and do directly a blocking send/recv
97     if (srcBuff_ != nullptr) {
98       simcall_comm_send(sender_, mailbox_->getInferior(), remains_, rate_,
99           srcBuff_, srcBuffSize_,
100           matchFunction_, copyDataFunction_,
101           userData_, -1 /*timeout*/);
102     } else {
103       simcall_comm_recv(receiver_, mailbox_->getInferior(), dstBuff_, &dstBuffSize_,
104           matchFunction_, copyDataFunction_,
105           userData_, -1/*timeout*/, rate_);
106     }
107   }
108   state_ = finished;
109 }
110 void Comm::wait(double timeout) {
111   xbt_assert(state_ == started || state_ == inited);
112
113   if (state_ == started) {
114     simcall_comm_wait(pimpl_, timeout);
115     state_ = finished;
116     return;
117   }
118
119   // It's not started yet. Do it in one simcall
120   if (srcBuff_ != nullptr) {
121     simcall_comm_send(sender_, mailbox_->getInferior(), remains_, rate_,
122         srcBuff_, srcBuffSize_,
123         matchFunction_, copyDataFunction_,
124         userData_, timeout);
125   } else { // Receiver
126     simcall_comm_recv(receiver_, mailbox_->getInferior(), dstBuff_, &dstBuffSize_,
127         matchFunction_, copyDataFunction_,
128         userData_, timeout, rate_);
129   }
130   state_ = finished;
131 }
132
133 s4u::Comm &Comm::send_async(Mailbox &dest, void *data, int simulatedSize) {
134   s4u::Comm &res = s4u::Comm::send_init(dest);
135   res.setRemains(simulatedSize);
136   res.srcBuff_ = data;
137   res.srcBuffSize_ = sizeof(void*);
138   res.start();
139   return res;
140 }
141
142 s4u::Comm &Comm::recv_async(Mailbox &dest, void **data) {
143   s4u::Comm &res = s4u::Comm::recv_init(dest);
144   res.setDstData(data);
145   res.start();
146   return res;
147 }
148
149 bool Comm::test() {
150   xbt_assert(state_ == inited || state_ == started || state_ == finished);
151   
152   if (state_ == finished) 
153     xbt_die("Don't call test on a finished comm.");
154   
155   if (state_ == inited) {
156     this->start();
157   }
158   
159   if(simcall_comm_test(pimpl_)){
160     state_ = finished;
161     return true;
162   }
163   return false;
164 }
165
166 }
167 }