]> AND Private Git Repository - loba.git/blob - messages.cpp
Logo AND Algorithmique Numérique Distribuée

Private GIT Repository
MSG type names got renamed in SimGrid 3.8.
[loba.git] / 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(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;
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 }