X-Git-Url: https://bilbo.iut-bm.univ-fcomte.fr/and/gitweb/loba.git/blobdiff_plain/d4fbe175e440c3e9de9b5468b85bf56daab62bff..de9d56dddba96f345ba109a8cdb0a2986cd59292:/process.h?ds=inline diff --git a/process.h b/process.h index 24d4c39..3e912f0 100644 --- a/process.h +++ b/process.h @@ -4,7 +4,8 @@ #define USE_UNORDERED_MAP 1 //#undef USE_UNORDERED_MAP -#include +#include +#include #ifdef USE_UNORDERED_MAP # include # define MAP_TEMPLATE std::tr1::unordered_map @@ -12,11 +13,15 @@ # include # define MAP_TEMPLATE std::map #endif +#include #include #include #include "communicator.h" +#include "misc.h" +#include "msg_thread.h" #include "neighbor.h" #include "options.h" +#include "synchro.h" class process { public: @@ -44,10 +49,17 @@ protected: double get_load() const; void set_load(double load); + // The load balancing algorithm comes here... + virtual void load_balance(); + // Register some amount of load to send to given neighbor. void send(neighbor& nb, double amount); void send(neighbor* nb, double amount) { send(*nb, amount); } + // Sort pneigh by applying comp to their loads + template + void pneigh_sort_by_load(const Compare& comp); + // Calls neighbor::print(verbose, logp, cat) for each member of neigh. void print_loads(bool verbose = false, e_xbt_log_priority_t logp = xbt_log_priority_info, @@ -86,31 +98,30 @@ private: double real_load; // current load double expected_load; // expected load in bookkeeping mode - // The load balancing algorithm comes here... - virtual void load_balance(); + mutex_t mutex; // synchronization between threads + condition_t cond; + + // Load-balancing loop + msg_thread* lb_thread; + void load_balance_loop(); - // Virtually do some computation - void compute(); + // Simulate computation loop + void compute_loop(); - // Send procedures, with helpers for bookkeeping mode or not - void send1_no_bookkeeping(neighbor& nb); - void send1_bookkeeping(neighbor& nb); - void send_all(); + bool still_running(); - // Returns true if there remains neighbors to listen for - bool may_receive() const { - return ctrl_close_pending || data_close_pending; - } + // Send procedures + void ctrl_send(neighbor& nb); + void data_send(neighbor& nb); + void ctrl_close(neighbor& nb); + void data_close(neighbor& nb); // Receive procedure // Parameter "timeout" may be 0 for non-blocking operation, -1 for // infinite waiting, or any positive timeout. - void receive(double timeout); - - // Finalize sends a "close" message to each neighbor and wait for - // all of them to answer. - void finalize1(neighbor& nb); - void finalize(); + void ctrl_receive(double timeout); + void data_receive(double timeout); + void handle_message(message* msg, m_host_t from); }; inline @@ -131,6 +142,18 @@ void process::set_load(double load) real_load = load; } +template +void process::pneigh_sort_by_load(const Compare& comp) +{ + using std::tr1::bind; + using std::tr1::placeholders::_1; + using std::tr1::placeholders::_2; + std::sort(pneigh.begin(), pneigh.end(), + bind(comp, + bind(&neighbor::get_load, _1), + bind(&neighbor::get_load, _2))); +} + #endif // !PROCESS_H // Local variables: