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

Private GIT Repository
0d47ca0a36305f7035dc1fc36f1de68dfd90134b
[loba.git] / communicator.h
1 // Asynchronous communicator
2
3 #ifndef COMMUNICATOR_H
4 #define COMMUNICATOR_H
5
6 #include <list>
7 #include <queue>
8 #include <string>
9 #include <msg/msg.h>
10 #include "hostdata.h"
11
12 class message {
13 public:
14     enum message_type { INFO, CREDIT, LOAD, CTRL_CLOSE, DATA_CLOSE };
15
16     message(message_type t, double a): type(t), amount(a) { }
17
18     message_type get_type() const       { return type;   }
19     double get_amount() const           { return amount; }
20
21     std::string to_string();
22
23 private:
24     message_type type;
25     double amount;
26 };
27
28 class communicator {
29 public:
30     communicator();
31     ~communicator();
32
33     // Send a message to the "dest" mailbox
34     void send(const char* dest, message* msg);
35
36     // Try to get a message.  Returns true on success.
37     // Parameter "timeout" may be 0 for non-blocking operation, -1 for
38     // infinite waiting, or any positive timeout.
39     bool recv(message*& msg, m_host_t& from, double timeout);
40
41     // Try to flush pending sending communications.
42     // If "wait" is true, blocks until success.
43     void flush(bool wait);
44
45 private:
46     // Myself
47     const hostdata* host;
48
49     // Used to synchronize main and receiver thread 
50     xbt_mutex_t mutex;
51     xbt_cond_t cond;
52
53     // List of pending send communications
54     std::list<msg_comm_t> sent_comm;
55
56     // Queue of received messages
57     std::queue<m_task_t> received;
58
59     // Control channel for receiving
60     m_task_t    ctrl_task;          // receive buffer
61     msg_comm_t  ctrl_comm;          // receive communication
62
63     // Data channel for receiving
64     m_task_t    data_task;          // receive buffer
65     msg_comm_t  data_comm;          // receive communication
66
67     const char* get_ctrl_mbox() const   { return host->get_ctrl_mbox(); }
68     const char* get_data_mbox() const   { return host->get_data_mbox(); }
69
70     // Handling of receiving thread
71     m_process_t receiver_process;
72     static int receiver_wrapper(int, char* []);
73     void receiver1(msg_comm_t& comm, m_task_t& task, const char* mbox);
74     int receiver();
75
76     // Used to test if a communication is over, and to destroy it if it is
77     static bool comm_test_n_destroy(msg_comm_t comm);
78 };
79
80 #endif // !COMMUNICATOR_H
81
82 // Local variables:
83 // mode: c++
84 // End: