]> AND Private Git Repository - loba.git/blob - process.cpp
Logo AND Algorithmique Numérique Distribuée

Private GIT Repository
Wip++...
[loba.git] / 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: