Logo AND Algorithmique Numérique Distribuée

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