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 CTRL:
  18         size = opt::bookkeeping ? 24 : 16; // type + amount + (credit)?
  19         break;
  20     case 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     static const char* str[DATA_CLOSE + 1] = { "CTRL", "DATA",
  32                                                "CTRL_CLOSE", "DATA_CLOSE" };
  33     std::ostringstream oss;
  34     oss << str[type] << ": " << amount;
  35     return oss.str();
  36 }
  37
  38 void message_queue::push(m_task_t task)
  39 {
  40     if (queue.push(task)) {
  41          // list was empty, the push must be signaled
  42         mutex.acquire();
  43         cond.signal();
  44         mutex.release();
  45     }
  46 }
  47
  48 bool message_queue::pop(message*& msg, m_host_t& from, double timeout)
  49 {
  50     m_task_t task;
  51     if (!queue.try_pop(task)) {
  52         if (timeout == 0.0)
  53             return false;
  54
  55         mutex.acquire();
  56         if (!queue.try_pop(task)) {
  57             xbt_ex_t e;
  58             XBT_DEBUG("waiting for a message to come");
  59             TRY {
  60                 if (timeout > 0)
  61                     cond.timedwait(mutex, timeout);
  62                 else
  63                     cond.wait(mutex);
  64             }
  65             TRY_CLEANUP {
  66                 mutex.release();
  67             }
  68             CATCH (e) {
  69                 if (e.category != timeout_error)
  70                     RETHROW;
  71                 xbt_ex_free(e);
  72                 return false;   // got a timeout
  73             }
  74             bool pop_was_successful = queue.try_pop(task);
  75             xbt_assert(pop_was_successful);
  76         } else {
  77             mutex.release();
  78         }
  79     }
  80     msg = static_cast<message*>(MSG_task_get_data(task));
  81     from = MSG_task_get_source(task);
  82     MSG_task_destroy(task);
  83
  84     XBT_DEBUG("received %s from %s",
  85               msg->to_string().c_str(), MSG_host_get_name(from));
  86
  87     return true;
  88 }