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

Private GIT Repository
Mutex and condition must be initialized after start of main...
[loba.git] / process.h
1 #ifndef PROCESS_H
2 #define PROCESS_H
3
4 #define USE_UNORDERED_MAP 1
5 //#undef USE_UNORDERED_MAP
6
7 #include <algorithm>
8 #include <tr1/functional>
9 #ifdef USE_UNORDERED_MAP
10 #  include <tr1/unordered_map>
11 #  define MAP_TEMPLATE std::tr1::unordered_map
12 #else
13 #  include <map>
14 #  define MAP_TEMPLATE std::map
15 #endif
16 #include <vector>
17 #include <msg/msg.h>
18 #include <xbt/log.h>
19 #include "communicator.h"
20 #include "misc.h"
21 #include "msg_thread.h"
22 #include "neighbor.h"
23 #include "options.h"
24
25 class process {
26 public:
27     static double get_total_load_init()    { return total_load_init;    }
28     static double get_total_load_running() { return total_load_running; }
29     static double get_total_load_exit()    { return total_load_exit;    }
30
31     process(int argc, char* argv[]);
32     virtual ~process();
33
34     double get_comp() const                { return comp; }
35     double get_real_load() const           { return real_load; }
36
37     int run();
38
39 protected:
40     typedef std::vector<neighbor> neigh_type;
41     typedef std::vector<neighbor*> pneigh_type;
42
43     pneigh_type pneigh;         // list of pointers to neighbors that
44                                 // we are free to reorder
45
46     // Get and set current load, which may be real load, or expected
47     // load if opt::bookkeeping is true.
48     double get_load() const;
49     void set_load(double load);
50
51     // The load balancing algorithm comes here...
52     virtual void load_balance();
53
54     // Register some amount of load to send to given neighbor.
55     void send(neighbor& nb, double amount);
56     void send(neighbor* nb, double amount) { send(*nb, amount); }
57
58     // Sort pneigh by applying comp to their loads
59     template <typename Compare>
60     void pneigh_sort_by_load(const Compare& comp);
61
62     // Calls neighbor::print(verbose, logp, cat) for each member of neigh.
63     void print_loads(bool verbose = false,
64                      e_xbt_log_priority_t logp = xbt_log_priority_info,
65                      xbt_log_category_t cat = _XBT_LOGV(default)) const;
66
67     // Calls neighbor::print(verbose, logp, cat) for each member of pneigh.
68     void print_loads_p(bool verbose = false,
69                        e_xbt_log_priority_t logp = xbt_log_priority_info,
70                        xbt_log_category_t cat = _XBT_LOGV(default)) const;
71
72 private:
73     static double total_load_init; // sum of process loads at init
74     static double total_load_running; // sum of loads while running
75     static double total_load_exit; // sum of process loads at exit
76
77     typedef MAP_TEMPLATE<m_host_t, neighbor*> rev_neigh_type;
78     neigh_type neigh;           // list of neighbors (do not alter
79                                 // after construction!)
80     rev_neigh_type rev_neigh;   // map m_host_t -> neighbor
81
82     communicator comm;          // communicator for this process
83     int ctrl_close_pending;     // number of "close" messages to wait
84                                 // on ctrl channel
85     int data_close_pending;     // number of "close" messages to wait
86                                 // on data channel
87     bool close_received;        // true if we received a "close" message
88     bool finalizing;            // true when finalize() is running
89
90     unsigned lb_iter;           // counter of load-balancing iterations
91     unsigned comp_iter;         // counter of computation iterations
92
93     double comp;                // total computing done so far (flops)
94
95     double prev_load_broadcast; // used to ensure that we do not send
96                                 // a same information messages
97     double real_load;           // current load
98     double expected_load;       // expected load in bookkeeping mode
99
100     // Load-balancing loop
101     void load_balance_loop();
102
103     // Simulate computation loop
104     msg_thread* compute_thread;
105     void compute_loop();
106
107     bool still_running();
108
109     // Send procedures
110     void ctrl_send(neighbor& nb);
111     void data_send(neighbor& nb);
112     void ctrl_close(neighbor& nb);
113     void data_close(neighbor& nb);
114
115     // Receive procedure
116     // Parameter "timeout" may be 0 for non-blocking operation, -1 for
117     // infinite waiting, or any positive timeout.
118     void ctrl_receive(double timeout);
119     void data_receive(double timeout);
120     void handle_message(message* msg, m_host_t from);
121 };
122
123 inline
124 double process::get_load() const
125 {
126     if (opt::bookkeeping)
127         return expected_load;
128     else
129         return real_load;
130 }
131
132 inline
133 void process::set_load(double load)
134 {
135     if (opt::bookkeeping)
136         expected_load = load;
137     else
138         real_load = load;
139 }
140
141 template <typename Compare>
142 void process::pneigh_sort_by_load(const Compare& comp)
143 {
144     using std::tr1::bind;
145     using std::tr1::placeholders::_1;
146     using std::tr1::placeholders::_2;
147     std::sort(pneigh.begin(), pneigh.end(),
148               bind(comp,
149                    bind(&neighbor::get_load, _1),
150                    bind(&neighbor::get_load, _2)));
151 }
152
153 #endif // !PROCESS_H
154
155 // Local variables:
156 // mode: c++
157 // End: