messages.cpp

   1 #include <sstream>
   2 #include <xbt/log.h>
   3
   4 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(comm);
   5
   6 #include "misc.h"
   7 #include "options.h"
   8
   9 #include "messages.h"
  10
  11 message::message(message_type t, double a, double c)
  12     : type(t), amount(a) , credit(c)
  13 {
  14     // compute message size
  15     // arbitrary: 8 for type, and 8 for each double
  16     switch (type) {
  17     case message_type::CTRL:
  18         size = opt::bookkeeping ? 24 : 16; // type + amount + (credit)?
  19         break;
  20     case message_type::DATA:
  21         size = 16 + opt::comm_cost(amount); // type + amount + data size
  22         break;
  23     default:
  24         size = 8;               // type
  25         break;
  26     }
  27 }
  28
  29 std::string message::to_string()
  30 {
  31     std::ostringstream oss;
  32     switch (type) {
  33     case message_type::CTRL:
  34         oss << "CTRL: " << amount << " (info)";
  35         if (opt::bookkeeping)
  36             oss << "; " << credit << " (credit)";
  37         break;
  38     case message_type::DATA:
  39         oss << "DATA: " << amount << " (load)";
  40         break;
  41     case message_type::CTRL_CLOSE:
  42         oss << "CTRL_CLOSE";
  43         break;
  44     case message_type::DATA_CLOSE:
  45         oss << "DATA_CLOSE";
  46         break;
  47     default:
  48         oss << "UNKNOWN MESSAGE TYPE: " << static_cast<int>(type);
  49         break;
  50     }
  51     return oss.str();
  52 }
  53
  54 void message_queue::push(msg_task_t task)
  55 {
  56     if (queue.push(task)) {
  57          // list was empty, the push must be signaled
  58         mutex.acquire();
  59         cond.signal();
  60         mutex.release();
  61     }
  62 }
  63
  64 bool message_queue::pop(message*& msg, msg_host_t& from, double timeout)
  65 {
  66     msg_task_t task = nullptr;
  67     if (!queue.try_pop(task)) {
  68         if (timeout == 0.0)
  69             return false;
  70
  71         mutex.acquire();
  72         if (!queue.try_pop(task)) {
  73             XBT_DEBUG("waiting for a message to come");
  74             bool hit_timeout;
  75             if (timeout > 0) {
  76                 hit_timeout = !cond.timedwait(mutex, timeout);
  77             } else {
  78                 cond.wait(mutex);
  79                 hit_timeout = false;
  80             }
  81             bool pop_was_successful = queue.try_pop(task);
  82             xbt_assert(hit_timeout || pop_was_successful);
  83         }
  84         mutex.release();
  85     }
  86     if (task == nullptr)
  87         return false;
  88
  89     msg = static_cast<message*>(MSG_task_get_data(task));
  90     from = MSG_task_get_source(task);
  91     MSG_task_destroy(task);
  92
  93     XBT_DEBUG("received %s from %s",
  94               msg->to_string().c_str(), MSG_host_get_name(from));
  95
  96     return true;
  97 }