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 // Cannot include "options.h" without error, so only declare the
  13 // needed functions.
  14 namespace opt {
  15     bool parse_args(int* argc, char* argv[]);
  16     void print();
  17     void usage();
  18 }
  19
  20 class message {
  21 public:
  22     enum message_type { INFO, CREDIT, LOAD, CTRL_CLOSE, DATA_CLOSE };
  23
  24     message(message_type t, double a): type(t), amount(a) { }
  25
  26     message_type get_type() const       { return type;   }
  27     double get_amount() const           { return amount; }
  28
  29     std::string to_string();
  30
  31 private:
  32     message_type type;
  33     double amount;
  34 };
  35
  36 class communicator {
  37 public:
  38     communicator();
  39     ~communicator();
  40
  41     // Send a message to the "dest" mailbox
  42     void send(const char* dest, message* msg);
  43
  44     // Try to get a message.  Returns true on success.
  45     // If "wait" is true, blocks until success.
  46     bool recv(message*& msg, m_host_t& from, bool wait);
  47
  48     // Try to flush pending sending communications.
  49     // If "wait" is true, blocks until success.
  50     void flush(bool wait);
  51
  52 private:
  53     // Myself
  54     const hostdata* host;
  55
  56     // Used to synchronize main and receiver thread
  57     xbt_mutex_t mutex;
  58     xbt_cond_t cond;
  59
  60     // List of pending send communications
  61     std::list<msg_comm_t> sent_comm;
  62     static int send_count_before_flush;
  63     int send_counter;
  64
  65     // Queue of received messages
  66     std::queue<m_task_t> received;
  67
  68     // Control channel for receiving
  69     m_task_t    ctrl_task;          // receive buffer
  70     msg_comm_t  ctrl_comm;          // receive communication
  71
  72     // Data channel for receiving
  73     m_task_t    data_task;          // receive buffer
  74     msg_comm_t  data_comm;          // receive communication
  75
  76     const char* get_ctrl_mbox() const   { return host->get_ctrl_mbox(); }
  77     const char* get_data_mbox() const   { return host->get_data_mbox(); }
  78
  79     // Handling of receiving thread
  80     m_process_t receiver_process;
  81     static int receiver_wrapper(int, char* []);
  82     int receiver();
  83
  84     // Used to test if a communication is over, and to destroy it if it is
  85     static bool comm_test_n_destroy(msg_comm_t comm);
  86
  87     // Make opt::* functions our friends to provide them an access to
  88     // send_count_before_flush
  89     friend bool opt::parse_args(int*, char* []);
  90     friend void opt::print();
  91     friend void opt::usage();
  92 };
  93
  94 #endif // !COMMUNICATOR_H
  95
  96 // Local variables:
  97 // mode: c++
  98 // End: