X-Git-Url: https://bilbo.iut-bm.univ-fcomte.fr/and/gitweb/loba.git/blobdiff_plain/28a30c9d25c9555dd21935653d2386653b0f3fe8..9539c77f2b5e0eef8d06da7a7ddd1d5546ba0731:/process.cpp?ds=inline diff --git a/process.cpp b/process.cpp index 76524fd..b6a6d28 100644 --- a/process.cpp +++ b/process.cpp @@ -22,6 +22,7 @@ double process::total_load_exit = 0.0; int process::process_counter = 0; double process::total_load_average; +double process::load_diff_threshold; namespace { @@ -70,6 +71,8 @@ process::process(int argc, char* argv[]) process_counter++; total_load_average = total_load_running / process_counter; + load_diff_threshold = (opt::load_ratio_threshold + + opt::avg_load_ratio * total_load_average) / 100.0; ctrl_close_pending = data_close_pending = neigh.size(); close_received = false; @@ -200,37 +203,27 @@ void process::compute_loop() double next_iter_after_date = MSG_get_clock() + opt::min_comp_iter_duration; while (still_running()) { - // receive (do not block if there is something to compute) - data_receive(real_load > 0.0 ? 0.0 : opt::min_comp_iter_duration); - // send - comm.data_flush(false); - mutex.acquire(); - real_load += received_load; - received_load = 0.0; - std::for_each(neigh.begin(), neigh.end(), - std::bind(&process::data_send, this, _1)); - mutex.release(); + do { + // receive + // if there is something to compute, do not block + // else, block the duration of an *lb* iteration + data_receive(real_load > 0.0 ? 0.0 : opt::min_lb_iter_duration); - ++all_comp_iter; - if (real_load == 0.0) - continue; - - double load_ratio = - 100.0 * std::fabs(real_load / total_load_average - 1.0); - if (convergence >= 0.0) { - if (load_ratio > opt::avg_load_ratio) { - XBT_VERB("current load has diverged: %g (%.4g%%)", - real_load, load_ratio); - convergence = -1.0; - } - } else { - if (load_ratio <= opt::avg_load_ratio) { - XBT_VERB("current load has converged: %g (%.4g%%)", - real_load, load_ratio); - convergence = MSG_get_clock(); - } - } + // send + comm.data_flush(false); + mutex.acquire(); + real_load += received_load; + received_load = 0.0; + std::for_each(neigh.begin(), neigh.end(), + std::bind(&process::data_send, this, _1)); + mutex.release(); + + ++all_comp_iter; + + } while (real_load == 0.0); + + convergence_check(); // compute ++comp_iter; @@ -259,9 +252,30 @@ void process::compute_loop() real_load += received_load; received_load = 0.0; total_load_running -= real_load; + convergence_check(); comm.data_flush(true); } +void process::convergence_check() +{ + double load_diff = std::fabs(real_load - total_load_average); + bool converged = load_diff <= load_diff_threshold; + + if (convergence >= 0.0) { + if (!converged) { + XBT_VERB("current load has diverged: %g (%.4g%%)", + real_load, 100.0 * load_diff / total_load_average); + convergence = -1.0; + } + } else { + if (converged) { + XBT_VERB("current load has converged: %g (%.4g%%)", + real_load, 100.0 * load_diff / total_load_average); + convergence = MSG_get_clock(); + } + } +} + bool process::still_running() { static bool last_status = true;