process.cpp

   1 #include <algorithm>
   2 #include <functional>
   3 #include <iterator>
   4 #include <stdexcept>
   5 #include <sstream>
   6 #include <xbt/log.h>
   7 #include "misc.h"
   8 #include "options.h"
   9 #include "process.h"
  10
  11 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(simu);
  12
  13 struct message {
  14     double measure;
  15     double transfer;
  16 };
  17
  18 process::process(int argc, char *argv[])
  19 {
  20     if (argc < 2 || !(std::istringstream(argv[1]) >> load))
  21         throw std::invalid_argument("bad or missing initial load");
  22     neigh.assign(argv + 2, argv + argc);
  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 << (neigh.size() > 1 ? "s: " : ": ");
  31         std::transform(neigh.begin(), neigh.end() - 1,
  32                        std::ostream_iterator<std::string>(oss, ", "),
  33                        std::mem_fun_ref(&neighbor::getName));
  34         oss << neigh.back().getName();
  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         compute();
  47         receive();
  48     }
  49     //    MSG_process_sleep(100.0);   // xxx
  50     /* xxx:
  51      * while (there is something to do) {
  52      *    compute some task;
  53      *    get received tasks;
  54      *    compute load balancing;
  55      *    send tasks to neighbors;
  56      * }
  57      */
  58
  59     /* Open Questions :
  60      * - definition of load on heterogeneous hosts ?
  61      * - how to detect convergence ?
  62      * - how to manage link failures ?
  63      */
  64
  65     // xxx: shall we retrieve pending tasks?
  66
  67     return 0;
  68 }
  69
  70 void process::receive()
  71 {
  72     m_task_t task;
  73     while ((task = comm.recv())) {
  74         message *msg = (message *)MSG_task_get_data(task);
  75         DEBUG3("Received load: %g, info: %g from %s",
  76                msg->transfer, msg->measure,
  77                MSG_host_get_name(MSG_task_get_source(task)));
  78         load += msg->transfer;
  79         // fixme: what is xxx ???
  80         // neigh[xxx].setLoad(msg->measure);
  81     }
  82 }
  83
  84 void process::compute()
  85 {
  86     double duration = opt::comp_cost(load);
  87     m_task_t task = MSG_task_create("computation", duration, 0.0, NULL);
  88     DEBUG2("Compute %g flop%s.", duration, duration > 1 ? "s" : "");
  89     MSG_task_execute(task);
  90     MSG_task_destroy(task);
  91 }
  92
  93 void process::print_loads(e_xbt_log_priority_t logp)
  94 {
  95     if (!LOG_ISENABLED(logp))
  96         return;
  97     std::ostringstream oss;
  98     if (neigh.empty()) {
  99         oss << "no neighbor!";
 100     } else {
 101         std::transform(neigh.begin(), neigh.end() - 1,
 102                        std::ostream_iterator<double>(oss, ", "),
 103                        std::mem_fun_ref(&neighbor::getLoad));
 104         oss << neigh.back().getLoad();
 105     }
 106     LOG1(logp, "Neighbor loads: %s", oss.str().c_str());
 107 }
 108
 109 // Local variables:
 110 // mode: c++
 111 // End: