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

Private GIT Repository
4ee3647d6895c4eef6586f92f29ffc498b45b14e
[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 <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     neigh.assign(argv + 2, argv + argc);
19     expected_load = load;
20     e_xbt_log_priority_t logp = xbt_log_priority_verbose;
21     if (!LOG_ISENABLED(logp))
22         return;
23     LOG1(logp, "My initial load is: %g", load);
24     std::ostringstream oss;
25     oss << neigh.size() << " neighbor";
26     if (!neigh.empty()) {
27         oss << (neigh.size() > 1 ? "s: " : ": ");
28         std::transform(neigh.begin(), neigh.end() - 1,
29                        std::ostream_iterator<const char*>(oss, ", "),
30                        std::mem_fun_ref(&neighbor::get_name));
31         oss << neigh.back().get_name();
32     }
33     LOG1(logp, "Got %s.", oss.str().c_str());
34     print_loads(logp);
35 }
36
37 int process::run()
38 {
39     INFO0("Coucou !");
40
41     int n = 100;
42     while (n--) {
43         if (load > 0)
44             compute();
45         else
46             xbt_sleep(0.5);     // fixme
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     bool received;
73     do {
74         double amount;
75         m_host_t from;
76         received = false;
77         if (comm.recv_info(amount, from)) {
78             // fixme: update neighbor
79             received = true;
80         }
81         if (comm.recv_credit(amount, from)) {
82             expected_load += amount;
83             received = true;
84         }
85         if (comm.recv_load(amount, from)) {
86             load += amount;
87             received = true;
88         }
89     } while (received);
90 }
91
92 void process::compute()
93 {
94     double duration = opt::comp_cost(load);
95     m_task_t task = MSG_task_create("computation", duration, 0.0, NULL);
96     DEBUG2("Compute %g flop%s.", duration, duration > 1 ? "s" : "");
97     MSG_task_execute(task);
98     MSG_task_destroy(task);
99 }
100
101 void process::print_loads(e_xbt_log_priority_t logp)
102 {
103     if (!LOG_ISENABLED(logp))
104         return;
105     std::ostringstream oss;
106     if (neigh.empty()) {
107         oss << "no neighbor!";
108     } else {
109         std::transform(neigh.begin(), neigh.end() - 1,
110                        std::ostream_iterator<double>(oss, ", "),
111                        std::mem_fun_ref(&neighbor::get_load));
112         oss << neigh.back().get_load();
113     }
114     LOG1(logp, "Neighbor loads: %s", oss.str().c_str());
115 }
116
117 // Local variables:
118 // mode: c++
119 // End: