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: