process.cpp

   1 #include "process.h"
   2
   3 #include <algorithm>
   4 #include <tr1/functional>
   5 #include <iterator>
   6 #include <numeric>
   7 #include <stdexcept>
   8 #include <sstream>
   9 #include <xbt/log.h>
  10 #include <xbt/time.h>
  11 #include "misc.h"
  12 #include "options.h"
  13
  14 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(proc);
  15
  16 process::process(int argc, char* argv[])
  17 {
  18     using namespace std::tr1;
  19     using namespace std::tr1::placeholders;
  20
  21     if (argc < 2 || !(std::istringstream(argv[1]) >> load))
  22         throw std::invalid_argument("bad or missing initial load");
  23
  24     neigh.assign(argv + 2, argv + argc);
  25
  26     std::for_each(neigh.begin(), neigh.end(),
  27                   bind(&process::insert_neighbor_in_map, this, _1));
  28
  29     pneigh.resize(neigh.size());
  30     std::transform(neigh.begin(), neigh.end(), pneigh.begin(),
  31                    misc::address<neighbor>());
  32
  33     expected_load = load;
  34
  35     ctrl_close_pending = data_close_pending = neigh.size();
  36     if (neigh.size() == 1) {
  37         comm.next_close_on_ctrl_is_last();
  38         comm.next_close_on_data_is_last();
  39     }
  40     if (neigh.size() > 0)
  41         comm.listen();
  42
  43     e_xbt_log_priority_t logp = xbt_log_priority_verbose;
  44     if (!LOG_ISENABLED(logp))
  45         return;
  46     std::ostringstream oss;
  47     oss << neigh.size() << " neighbor";
  48     if (!neigh.empty()) {
  49         oss << ESSE(neigh.size()) << ": ";
  50         std::transform(neigh.begin(), neigh.end() - 1,
  51                        std::ostream_iterator<const char*>(oss, ", "),
  52                        mem_fn(&neighbor::get_name));
  53         oss << neigh.back().get_name();
  54     }
  55     LOG1(logp, "Got %s.", oss.str().c_str());
  56     print_loads(logp);
  57 }
  58
  59 process::~process()
  60 {
  61 }
  62
  63 int process::run()
  64 {
  65     bool one_more = true;
  66
  67     INFO1("Initial load: %g", load);
  68     VERB0("Starting...");
  69     // first send to inform neighbors about our load
  70     send();
  71     iter = 0;
  72     while (one_more) {
  73         ++iter;
  74
  75         if (opt::log_rate && iter % opt::log_rate == 0) {
  76             if (opt::bookkeeping)
  77                 INFO3("(%u) current load: %g ; expected: %g",
  78                       iter, load, expected_load);
  79             else
  80                 INFO2("(%u) current load: %g",
  81                       iter, load);
  82         }
  83
  84         compute();
  85
  86         bool close_received = !receive(false);
  87
  88         if (opt::bookkeeping)
  89             expected_load -= load_balance(expected_load);
  90         else
  91             load -= load_balance(load);
  92
  93         send();
  94
  95         if (opt::exit_on_close && close_received)
  96             one_more = false;
  97         if (opt::maxiter && iter >= opt::maxiter)
  98             one_more = false;
  99     }
 100     VERB0("Going to finalize...");
 101     finalize();
 102
 103     /* Open Questions :
 104      * - definition of load on heterogeneous hosts ?
 105      * - how to detect convergence ?
 106      * - how to manage link failures ?
 107      */
 108
 109     VERB0("Done.");
 110     if (opt::bookkeeping)
 111         INFO4("Final load after %d iteration%s: %g ; expected: %g",
 112               iter, ESSE(iter), load, expected_load);
 113     else
 114         INFO3("Final load after %d iteration%s: %g", iter, ESSE(iter), load);
 115     return 0;
 116 }
 117
 118 double process::sum_of_to_send() const
 119 {
 120     using namespace std::tr1;
 121     using namespace std::tr1::placeholders;
 122
 123     return std::accumulate(neigh.begin(), neigh.end(), 0.0,
 124                            bind(std::plus<double>(),
 125                                 _1, bind(&neighbor::get_to_send, _2)));
 126 }
 127
 128 double process::load_balance(double /*my_load*/)
 129 {
 130     return 0.0;
 131 }
 132
 133 void process::compute()
 134 {
 135     // fixme: shall we do something special when duration is 0 ?
 136     double duration = opt::comp_cost(load);
 137     if (duration > 0)  {
 138         m_task_t task = MSG_task_create("computation", duration, 0.0, NULL);
 139         DEBUG2("compute %g flop%s.", duration, ESSE(duration));
 140         MSG_task_execute(task);
 141         MSG_task_destroy(task);
 142     } else {
 143         xbt_sleep(42);
 144         //        xbt_thread_yield();
 145     }
 146 }
 147
 148 void process::send1_no_bookkeeping(neighbor& nb)
 149 {
 150     comm.send(nb.get_ctrl_mbox(), new message(message::INFO, load));
 151     double load_to_send = nb.get_to_send();
 152     if (load_to_send > 0.0) {
 153         comm.send(nb.get_data_mbox(), new message(message::LOAD, load_to_send));
 154         nb.set_to_send(0.0);
 155     }
 156 }
 157
 158 void process::send1_bookkeeping(neighbor& nb)
 159 {
 160     comm.send(nb.get_ctrl_mbox(), new message(message::INFO, expected_load));
 161     double load_to_send;
 162     double new_debt;
 163     double debt_to_send = nb.get_to_send();
 164     if (debt_to_send > 0.0) {
 165         comm.send(nb.get_ctrl_mbox(),
 166                   new message(message::CREDIT, debt_to_send));
 167         nb.set_to_send(0.0);
 168         new_debt = nb.get_debt() + debt_to_send;
 169     } else {
 170         new_debt = nb.get_debt();
 171     }
 172     if (load <= new_debt) {
 173         load_to_send = load;
 174         nb.set_debt(new_debt - load_to_send);
 175         load = 0.0;
 176     } else {
 177         load_to_send = new_debt;
 178         nb.set_debt(0.0);
 179         load -= load_to_send;
 180     }
 181     if (load_to_send > 0.0)
 182         comm.send(nb.get_data_mbox(), new message(message::LOAD, load_to_send));
 183 }
 184
 185 void process::send()
 186 {
 187     using namespace std::tr1;
 188     using namespace std::tr1::placeholders;
 189
 190     // fixme: shall we send data at all iterations?
 191     if (opt::bookkeeping)
 192         std::for_each(neigh.begin(), neigh.end(),
 193                       bind(&process::send1_bookkeeping, this, _1));
 194     else
 195         std::for_each(neigh.begin(), neigh.end(),
 196                       bind(&process::send1_no_bookkeeping, this, _1));
 197 }
 198
 199 // Returns false if a CLOSE message was received.
 200 bool process::receive(bool wait_for_close)
 201 {
 202     bool result = true;
 203     message* msg;
 204     m_host_t from;
 205     while ((ctrl_close_pending ||
 206             data_close_pending) && comm.recv(msg, from, wait_for_close)) {
 207         DEBUG2("received %s from %s",
 208                msg->to_string().c_str(), MSG_host_get_name(from));
 209         switch (msg->get_type()) {
 210         case message::INFO: {
 211             neighbor* n = rev_neigh[from];
 212             n->set_load(msg->get_amount());
 213             break;
 214         }
 215         case message::CREDIT:
 216             expected_load += msg->get_amount();
 217             break;
 218         case message::LOAD:
 219             load += msg->get_amount();
 220             break;
 221         case message::CTRL_CLOSE:
 222             if (--ctrl_close_pending == 1)
 223                 comm.next_close_on_ctrl_is_last();
 224             DEBUG1("ctrl_close_pending = %d", ctrl_close_pending);
 225             result = false;
 226             break;
 227         case message::DATA_CLOSE:
 228             if (--data_close_pending == 1)
 229                 comm.next_close_on_data_is_last();
 230             DEBUG1("data_close_pending = %d", data_close_pending);
 231             result = false;
 232             break;
 233         }
 234         delete msg;
 235     }
 236     return result;
 237 }
 238
 239 void process::finalize1(neighbor& nb)
 240 {
 241     comm.send(nb.get_ctrl_mbox(), new message(message::CTRL_CLOSE, 0.0));
 242     comm.send(nb.get_data_mbox(), new message(message::DATA_CLOSE, 0.0));
 243 }
 244
 245 void process::finalize()
 246 {
 247     using namespace std::tr1;
 248     using namespace std::tr1::placeholders;
 249
 250     DEBUG2("send CLOSE to %d neighbor%s.",
 251            (int )neigh.size(), ESSE(neigh.size()));
 252     std::for_each(neigh.begin(), neigh.end(),
 253                   bind(&process::finalize1, this, _1));
 254
 255     DEBUG2("wait for CLOSE from %d neighbor%s.",
 256            (int )neigh.size(), ESSE(neigh.size()));
 257     receive(true);
 258
 259     comm.flush(true);
 260 }
 261
 262 void process::print_loads(e_xbt_log_priority_t logp)
 263 {
 264     if (!LOG_ISENABLED(logp))
 265         return;
 266
 267     std::ostringstream oss;
 268     if (neigh.empty()) {
 269         oss << "no neighbor!";
 270     } else {
 271         std::transform(neigh.begin(), neigh.end() - 1,
 272                        std::ostream_iterator<double>(oss, ", "),
 273                        std::tr1::mem_fn(&neighbor::get_load));
 274         oss << neigh.back().get_load();
 275     }
 276     LOG1(logp, "Neighbor loads: %s", oss.str().c_str());
 277 }
 278
 279 void process::insert_neighbor_in_map(neighbor& nb)
 280 {
 281     neighbor* nbp = &nb;
 282     m_host_t host = MSG_get_host_by_name(nb.get_name());
 283     rev_neigh.insert(std::make_pair(host, nbp));
 284 }
 285
 286 // Local variables:
 287 // mode: c++
 288 // End: