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     std::ostringstream oss;
  32     switch (type) {
  33     case CTRL:
  34         oss << "CTRL: " << amount << " (info)";
  35         if (opt::bookkeeping)
  36             oss << "; " << credit << " (credit)";
  37         break;
  38     case DATA:
  39         oss << "DATA: " << amount << " (load)";
  40         break;
  41     case CTRL_CLOSE:
  42         oss << "CTRL_CLOSE";
  43         break;
  44     case DATA_CLOSE:
  45         oss << "DATA_CLOSE";
  46         break;
  47     default:
  48         oss << "UNKNOWN MESSAGE TYPE: " << type;
  49         break;
  50     }
  51     return oss.str();
  52 }
  53
  54 void message_queue::push(m_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, m_host_t& from, double timeout)
  65 {
  66     m_task_t task;
  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_ex_t e;
  74             XBT_DEBUG("waiting for a message to come");
  75             TRY_FAST {
  76                 if (timeout > 0)
  77                     cond.timedwait(mutex, timeout);
  78                 else
  79                     cond.wait(mutex);
  80             }
  81             TRY_CLEANUP {
  82                 mutex.release();
  83             }
  84             CATCH (e) {
  85                 if (e.category != timeout_error)
  86                     RETHROW;
  87                 xbt_ex_free(e);
  88                 return false;   // got a timeout
  89             }
  90             bool pop_was_successful = queue.try_pop(task);
  91             xbt_assert(pop_was_successful);
  92         } else {
  93             mutex.release();
  94         }
  95     }
  96     msg = static_cast<message*>(MSG_task_get_data(task));
  97     from = MSG_task_get_source(task);
  98     MSG_task_destroy(task);
  99
 100     XBT_DEBUG("received %s from %s",
 101               msg->to_string().c_str(), MSG_host_get_name(from));
 102
 103     return true;
 104 }