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 }