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 }