Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
further snake_case in routing
[simgrid.git] / src / s4u / s4u_comm.cpp
1 /* Copyright (c) 2006-2017. The SimGrid Team. All rights reserved.          */
2
3 /* This program is free software; you can redistribute it and/or modify it
4  * under the terms of the license (GNU LGPL) which comes with this package. */
5
6 #include "src/msg/msg_private.hpp"
7 #include "xbt/log.h"
8
9 #include "simgrid/s4u/Comm.hpp"
10 #include "simgrid/s4u/Mailbox.hpp"
11
12 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(s4u_comm,s4u_activity,"S4U asynchronous communications");
13
14 namespace simgrid {
15 namespace s4u {
16 Comm::~Comm()
17 {
18   if (state_ == started && not detached_ && (pimpl_ == nullptr || pimpl_->state == SIMIX_RUNNING)) {
19     XBT_INFO("Comm %p freed before its completion. Detached: %d, State: %d", this, detached_, state_);
20     if (pimpl_ != nullptr)
21       XBT_INFO("pimpl_->state: %d", pimpl_->state);
22     else
23       XBT_INFO("pimpl_ is null");
24     xbt_backtrace_display_current();
25   }
26 }
27
28 Activity* Comm::setRate(double rate)
29 {
30   xbt_assert(state_==inited);
31   rate_ = rate;
32   return this;
33 }
34
35 Activity* Comm::setSrcData(void* buff)
36 {
37   xbt_assert(state_==inited);
38   xbt_assert(dstBuff_ == nullptr, "Cannot set the src and dst buffers at the same time");
39   srcBuff_ = buff;
40   return this;
41 }
42 Activity* Comm::setSrcDataSize(size_t size)
43 {
44   xbt_assert(state_==inited);
45   srcBuffSize_ = size;
46   return this;
47 }
48 Activity* Comm::setSrcData(void* buff, size_t size)
49 {
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   return this;
56 }
57 Activity* Comm::setDstData(void** buff)
58 {
59   xbt_assert(state_==inited);
60   xbt_assert(srcBuff_ == nullptr, "Cannot set the src and dst buffers at the same time");
61   dstBuff_ = buff;
62   return this;
63 }
64 size_t Comm::getDstDataSize(){
65   xbt_assert(state_==finished);
66   return dstBuffSize_;
67 }
68 Activity* Comm::setDstData(void** buff, size_t size)
69 {
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   return this;
76 }
77
78 Activity* Comm::start()
79 {
80   xbt_assert(state_ == inited);
81
82   if (srcBuff_ != nullptr) { // Sender side
83     pimpl_ = simcall_comm_isend(sender_, mailbox_->getImpl(), remains_, rate_, srcBuff_, srcBuffSize_, matchFunction_,
84                                 cleanFunction_, copyDataFunction_, user_data_, detached_);
85   } else if (dstBuff_ != nullptr) { // Receiver side
86     xbt_assert(not detached_, "Receive cannot be detached");
87     pimpl_ = simcall_comm_irecv(receiver_, mailbox_->getImpl(), dstBuff_, &dstBuffSize_, matchFunction_,
88                                 copyDataFunction_, user_data_, 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   return this;
95 }
96
97 /** @brief Block the calling actor until the communication is finished */
98 Activity* Comm::wait()
99 {
100   return this->wait(-1);
101 }
102
103 /** @brief Block the calling actor until the communication is finished, or until timeout
104  *
105  * On timeout, an exception is thrown.
106  *
107  * @param timeout the amount of seconds to wait for the comm termination.
108  *                Negative values denote infinite wait times. 0 as a timeout returns immediately. */
109 Activity* Comm::wait(double timeout)
110 {
111   switch (state_) {
112     case finished:
113       return this;
114
115     case inited: // It's not started yet. Do it in one simcall
116       if (srcBuff_ != nullptr) {
117         simcall_comm_send(sender_, mailbox_->getImpl(), remains_, rate_, srcBuff_, srcBuffSize_, matchFunction_,
118                           copyDataFunction_, user_data_, timeout);
119       } else { // Receiver
120         simcall_comm_recv(receiver_, mailbox_->getImpl(), dstBuff_, &dstBuffSize_, matchFunction_, copyDataFunction_,
121                           user_data_, timeout, rate_);
122       }
123       state_ = finished;
124       return this;
125
126     case started:
127       simcall_comm_wait(pimpl_, timeout);
128       state_ = finished;
129       return this;
130
131     default:
132       THROW_IMPOSSIBLE;
133   }
134   return this;
135 }
136 int Comm::test_any(std::vector<CommPtr>* comms)
137 {
138   smx_activity_t* array = new smx_activity_t[comms->size()];
139   for (unsigned int i = 0; i < comms->size(); i++) {
140     array[i] = comms->at(i)->pimpl_;
141   }
142   int res = simcall_comm_testany(array, comms->size());
143   delete[] array;
144   return res;
145 }
146
147 Activity* Comm::detach()
148 {
149   xbt_assert(state_ == inited, "You cannot detach communications once they are started.");
150   xbt_assert(srcBuff_ != nullptr && srcBuffSize_ != 0, "You can only detach sends, not recvs");
151   detached_ = true;
152   return start();
153 }
154
155 Activity* Comm::cancel()
156 {
157   simgrid::kernel::activity::CommImplPtr commPimpl =
158       boost::static_pointer_cast<simgrid::kernel::activity::CommImpl>(pimpl_);
159   commPimpl->cancel();
160   return this;
161 }
162
163 bool Comm::test()
164 {
165   xbt_assert(state_ == inited || state_ == started || state_ == finished);
166
167   if (state_ == finished) {
168     return true;
169   }
170
171   if (state_ == inited) {
172     this->start();
173   }
174
175   if(simcall_comm_test(pimpl_)){
176     state_ = finished;
177     return true;
178   }
179   return false;
180 }
181
182 MailboxPtr Comm::getMailbox()
183 {
184   return mailbox_;
185 }
186
187 void intrusive_ptr_release(simgrid::s4u::Comm* c)
188 {
189   if (c->refcount_.fetch_sub(1, std::memory_order_release) == 1) {
190     std::atomic_thread_fence(std::memory_order_acquire);
191     delete c;
192   }
193 }
194 void intrusive_ptr_add_ref(simgrid::s4u::Comm* c)
195 {
196   c->refcount_.fetch_add(1, std::memory_order_relaxed);
197 }
198 }
199 } // namespaces