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 #include "synchro.h"
  12
  13 class message {
  14 public:
  15     enum message_type { INFO, CREDIT, LOAD, CTRL_CLOSE, DATA_CLOSE };
  16
  17     message(message_type t, double a): type(t), amount(a) { }
  18
  19     message_type get_type() const       { return type;   }
  20     double get_amount() const           { return amount; }
  21
  22     std::string to_string();
  23
  24 private:
  25     message_type type;
  26     double amount;
  27 };
  28
  29 class communicator {
  30 public:
  31     communicator();
  32     ~communicator();
  33
  34     // Send a message to the "dest" mailbox
  35     void send(const char* dest, message* msg);
  36
  37     // Try to get a message.  Returns true on success.
  38     // Parameter "timeout" may be 0 for non-blocking operation, -1 for
  39     // infinite waiting, or any positive timeout.
  40     bool recv(message*& msg, m_host_t& from, double timeout);
  41
  42     // Try to flush pending sending communications.
  43     // If "wait" is true, blocks until success.
  44     void flush(bool wait);
  45
  46 private:
  47     // Myself
  48     const hostdata* host;
  49
  50     // List of pending send communications
  51     std::list<msg_comm_t> sent_comm;
  52
  53     // Queue of received messages
  54     std::queue<m_task_t> received;
  55
  56     // Handling of receiving thread
  57     mutex_t receiver_mutex;
  58     condition_t receiver_cond;
  59     m_process_t receiver_thread;
  60     static int receiver_wrapper(int, char* []);
  61     void receiver();
  62
  63     // Used to chek if a communication is successfull before destroying it
  64     static void comm_check_n_destroy(msg_comm_t comm);
  65     // If comm is over, call comm_check_n_destroy(comm), and return true
  66     static bool comm_test_n_destroy(msg_comm_t comm);
  67 };
  68
  69 #endif // !COMMUNICATOR_H
  70
  71 // Local variables:
  72 // mode: c++
  73 // End: