process.cpp

   1 #include <algorithm>
   2 #include <tr1/functional>
   3 #include <iterator>
   4 #include <stdexcept>
   5 #include <sstream>
   6 #include <xbt/log.h>
   7 #include <xbt/time.h>
   8 #include "misc.h"
   9 #include "options.h"
  10 #include "process.h"
  11
  12 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(simu);
  13
  14 process::process(int argc, char* argv[])
  15 {
  16     if (argc < 2 || !(std::istringstream(argv[1]) >> load))
  17         throw std::invalid_argument("bad or missing initial load");
  18
  19     neigh.assign(argv + 2, argv + argc);
  20     expected_load = load;
  21     ctrl_close_pending = data_close_pending = neigh.size();
  22
  23     e_xbt_log_priority_t logp = xbt_log_priority_verbose;
  24     if (!LOG_ISENABLED(logp))
  25         return;
  26     LOG1(logp, "My initial load is: %g", load);
  27     std::ostringstream oss;
  28     oss << neigh.size() << " neighbor";
  29     if (!neigh.empty()) {
  30         oss << ESSE(neigh.size()) << ": ";
  31         std::transform(neigh.begin(), neigh.end() - 1,
  32                        std::ostream_iterator<const char*>(oss, ", "),
  33                        std::tr1::mem_fn(&neighbor::get_name));
  34         oss << neigh.back().get_name();
  35     }
  36     LOG1(logp, "Got %s.", oss.str().c_str());
  37     print_loads(logp);
  38 }
  39
  40 int process::run()
  41 {
  42     INFO0("Coucou !");
  43
  44     int n = 100;
  45     while (n--) {
  46         if (opt::bookkeeping)
  47             INFO2("current load: %g ; expected: %g", load, expected_load);
  48         else
  49             INFO1("current load: %g", load);
  50
  51         if (load > 0)
  52             compute();
  53         else
  54             xbt_sleep(100);        // fixme
  55         if (!receive(false))
  56             n = 0;
  57     }
  58     DEBUG0("going to finalize.");
  59     finalize();
  60
  61     //    MSG_process_sleep(100.0);   // xxx
  62     /* xxx:
  63      * while (there is something to do) {
  64      *    compute some task;
  65      *    get received tasks;
  66      *    compute load balancing;
  67      *    send tasks to neighbors;
  68      * }
  69      * finalize;
  70      * wait for pending messages;
  71      */
  72
  73     /* Open Questions :
  74      * - definition of load on heterogeneous hosts ?
  75      * - how to detect convergence ?
  76      * - how to manage link failures ?
  77      */
  78
  79     DEBUG0("done.");
  80     return 0;
  81 }
  82
  83 void process::compute()
  84 {
  85     double duration = opt::comp_cost(load);
  86     m_task_t task = MSG_task_create("computation", duration, 0.0, NULL);
  87     DEBUG2("compute %g flop%s.", duration, ESSE(duration));
  88     MSG_task_execute(task);
  89     MSG_task_destroy(task);
  90 }
  91
  92 bool process::receive(bool wait_for_close)
  93 {
  94     bool result = true;
  95     message* msg;
  96     m_host_t from;
  97     while ((ctrl_close_pending ||
  98             data_close_pending) && comm.recv(msg, from, wait_for_close)) {
  99         switch (msg->get_type()) {
 100         case message::INFO:
 101             DEBUG0("received INFO");
 102             // fixme: update neighbor
 103             // need a map m_host_t -> neighbor&
 104             break;
 105         case message::CREDIT:
 106             DEBUG0("received CREDIT");
 107             expected_load += msg->get_amount();
 108             break;
 109         case message::LOAD:
 110             DEBUG0("received LOAD");
 111             load += msg->get_amount();
 112             break;
 113         case message::CTRL_CLOSE:
 114             DEBUG0("received CTRL_CLOSE");
 115             if (--ctrl_close_pending == 1)
 116                 comm.next_close_on_ctrl_is_last();
 117             result = false;
 118             break;
 119         case message::DATA_CLOSE:
 120             DEBUG0("received DATA_CLOSE");
 121             if (--data_close_pending == 1)
 122                 comm.next_close_on_data_is_last();
 123             result = false;
 124             break;
 125         }
 126         delete msg;
 127     }
 128     return result;
 129 }
 130
 131 void process::finalize()
 132 {
 133     DEBUG2("send CLOSE to %d neighbor%s.",
 134            (int )neigh.size(), ESSE(neigh.size()));
 135     std::vector<neighbor>::iterator n;
 136     for (n = neigh.begin() ; n != neigh.end() ; ++n) {
 137         comm.send(n->get_ctrl_mbox(), new message(message::CTRL_CLOSE, 0.0));
 138         comm.send(n->get_data_mbox(), new message(message::DATA_CLOSE, 0.0));
 139     }
 140
 141     DEBUG2("wait for CLOSE from %d neighbor%s.",
 142            (int )neigh.size(), ESSE(neigh.size()));
 143     receive(true);
 144
 145     comm.wait_for_sent();
 146 }
 147
 148 void process::print_loads(e_xbt_log_priority_t logp)
 149 {
 150     if (!LOG_ISENABLED(logp))
 151         return;
 152
 153     std::ostringstream oss;
 154     if (neigh.empty()) {
 155         oss << "no neighbor!";
 156     } else {
 157         std::transform(neigh.begin(), neigh.end() - 1,
 158                        std::ostream_iterator<double>(oss, ", "),
 159                        std::tr1::mem_fn(&neighbor::get_load));
 160         oss << neigh.back().get_load();
 161     }
 162     LOG1(logp, "Neighbor loads: %s", oss.str().c_str());
 163 }
 164
 165 // Local variables:
 166 // mode: c++
 167 // End: