From 27d258ae26ce11a07dba3e01597531d5d882082f Mon Sep 17 00:00:00 2001 From: Thomas Heller Date: Thu, 22 Nov 2018 06:50:19 +0100 Subject: [PATCH 1/2] mods for review --- allscale/data_item_manager/acquire.hpp | 2 ++ allscale/data_item_manager/add_allowance.hpp | 2 +- allscale/data_item_reference.hpp | 10 +++++++--- src/dashboard.cpp | 21 ++++++++++---------- src/data_item_reference.cpp | 11 ++++++++++ src/schedule_policy.cpp | 20 +++++++++++++------ 6 files changed, 45 insertions(+), 21 deletions(-) diff --git a/allscale/data_item_manager/acquire.hpp b/allscale/data_item_manager/acquire.hpp index 9b886a1..629eb9f 100644 --- a/allscale/data_item_manager/acquire.hpp +++ b/allscale/data_item_manager/acquire.hpp @@ -89,6 +89,7 @@ namespace allscale { namespace data_item_manager { if(!missing.empty()) { + entry.resize_fragment(req, missing, true); HPX_ASSERT(req.allowance.empty()); hpx::future update_cache = @@ -99,6 +100,7 @@ namespace allscale { namespace data_item_manager { return hpx::dataflow(hpx::launch::sync, [req, addr](hpx::future infof, hpx::future cache_update) ->hpx::future { + cache_update.get(); auto info = infof.get(); if (info.regions.empty()) return hpx::make_ready_future(); diff --git a/allscale/data_item_manager/add_allowance.hpp b/allscale/data_item_manager/add_allowance.hpp index 2e4e2ce..38c4dcb 100644 --- a/allscale/data_item_manager/add_allowance.hpp +++ b/allscale/data_item_manager/add_allowance.hpp @@ -33,7 +33,7 @@ namespace allscale { namespace data_item_manager { { entry.add_full(req.allowance); if (isLeaf) - entry.resize_fragment(req, req.region, true); + entry.resize_fragment(req, req.allowance, true); } else { diff --git a/allscale/data_item_reference.hpp b/allscale/data_item_reference.hpp index f2e931a..9488445 100644 --- a/allscale/data_item_reference.hpp +++ b/allscale/data_item_reference.hpp @@ -12,6 +12,10 @@ /////////////////////////////////////////////////////////////////////////// namespace allscale { + namespace detail + { + std::uint32_t create_data_item_id(); + } struct data_item_id { std::uint32_t locality_; @@ -40,11 +44,12 @@ namespace allscale { ar & id_; } + static data_item_id create() { - static hpx::util::atomic_count id(0); - return {hpx::get_locality_id(), static_cast(++id - 1)}; + return {hpx::get_locality_id(), detail::create_data_item_id()}; } + }; template @@ -130,7 +135,6 @@ namespace allscale { // a transient cache for the referenced fragment mutable std::atomic cache; - }; } diff --git a/src/dashboard.cpp b/src/dashboard.cpp index 476669e..ee326f1 100644 --- a/src/dashboard.cpp +++ b/src/dashboard.cpp @@ -452,15 +452,20 @@ namespace allscale { namespace dashboard auto start = std::chrono::steady_clock::now(); - auto f = client.get_system_state(); - f.then( - [&client](hpx::future> nodes) + auto f = client.get_system_state().then( + [&client, start](hpx::future> nodes) { system_state state; auto next_update = - []() - {}; + [start]() + { + using namespace std::chrono_literals; + auto duration = std::chrono::steady_clock::now() - start; + if (duration < 1s) + std::this_thread::sleep_for(1s - duration); + update(); + }; try { @@ -507,12 +512,6 @@ namespace allscale { namespace dashboard next_update(); } }); - - using namespace std::chrono_literals; - auto duration = std::chrono::steady_clock::now() - start; - if (duration < 1s) - std::this_thread::sleep_for(1s - duration); - hpx::apply(update); } void get_commands() diff --git a/src/data_item_reference.cpp b/src/data_item_reference.cpp index df47abc..05b8531 100644 --- a/src/data_item_reference.cpp +++ b/src/data_item_reference.cpp @@ -1,2 +1,13 @@ #include +namespace allscale { + namespace detail { + std::uint32_t create_data_item_id() + { + static hpx::util::atomic_count id(0); + std::uint32_t this_id = static_cast(++id - 1); + std::cout << "creating DIR " << this_id << "\n"; + return this_id; + } + } +} diff --git a/src/schedule_policy.cpp b/src/schedule_policy.cpp index 3835905..35d42c0 100644 --- a/src/schedule_policy.cpp +++ b/src/schedule_policy.cpp @@ -405,8 +405,8 @@ namespace allscale { // float margin = 3.291f * std::sqrt(var / task_costs.size()); // ... 99.9% // float margin = 2.807f * std::sqrt(var / task_costs.size()); // ... 99.5% // float margin = 2.576f * std::sqrt(var / task_costs.size()); // ... 99% - float margin = 1.96f * std::sqrt(var / task_costs.size()); // ... 95% -// float margin = 1.282f * std::sqrt(var / task_costs.size()); // ... 80% +// float margin = 1.96f * std::sqrt(var / task_costs.size()); // ... 95% + float margin = 1.282f * std::sqrt(var / task_costs.size()); // ... 80% float lower = avg - margin; float upper = avg + margin; @@ -438,6 +438,7 @@ namespace allscale { if (node_avg[node + 1] < node_avg[node] - margin) { target_node = node + 1; + // advance to node that hasn't been disabled... node_contribution[node]--; node_contribution[target_node]++; update_mapping = true; @@ -449,6 +450,7 @@ namespace allscale { if (node_avg[node - 1] < node_avg[node] - margin) { target_node = node - 1; + // advance to node that hasn't been disabled... node_contribution[node]--; node_contribution[target_node]++; update_mapping = true; @@ -461,6 +463,7 @@ namespace allscale { if (node_avg[node - 1] < node_avg[node] - margin) { target_node = node - 1; + // advance to node that hasn't been disabled... node_contribution[node]--; node_contribution[target_node]++; update_mapping = true; @@ -471,6 +474,7 @@ namespace allscale { if (node_avg[node + 1] < node_avg[node] - margin) { target_node = node + 1; + // advance to node that hasn't been disabled... node_contribution[node]--; node_contribution[target_node]++; update_mapping = true; @@ -487,6 +491,7 @@ namespace allscale { if (node_avg[node + 1] > node_avg[node] + margin) { target_node = node + 1; + // advance to node that hasn't been disabled... node_contribution[node]++; node_contribution[target_node]--; update_mapping = true; @@ -499,6 +504,7 @@ namespace allscale { if (node_avg[node - 1] > node_avg[node] + margin) { target_node = node - 1; + // advance to node that hasn't been disabled... node_contribution[node]++; node_contribution[target_node]--; update_mapping = true; @@ -512,6 +518,7 @@ namespace allscale { if (node_avg[node + 1] > node_avg[node] + margin) { target_node = node + 1; + // advance to node that hasn't been disabled... node_contribution[node]++; node_contribution[target_node]--; update_mapping = true; @@ -523,16 +530,17 @@ namespace allscale { if (node_avg[node - 1] > node_avg[node] + margin) { target_node = node - 1; + // advance to node that hasn't been disabled... node_contribution[node]++; node_contribution[target_node]--; update_mapping = true; } } }; - if (var < 0.01f) - { - } - else +// if (var < 0.01f) +// { +// } +// else { // Go over all node averages for (std::size_t i = 0; i != node_avg.size(); ++i) From 6e308e5a1df10a59f5d039945eb2cdf87ab9262c Mon Sep 17 00:00:00 2001 From: Thomas Heller Date: Fri, 23 Nov 2018 01:57:13 +0100 Subject: [PATCH 2/2] demo prep --- allscale/components/monitor.hpp | 5 +- allscale/dashboard.hpp | 2 +- src/components/monitor_component.cpp | 54 ++-- src/dashboard.cpp | 14 +- src/data_item_reference.cpp | 2 +- src/optimizer.cpp | 16 +- src/schedule_policy.cpp | 448 ++++++++++++++------------- src/scheduler.cpp | 5 + 8 files changed, 297 insertions(+), 249 deletions(-) diff --git a/allscale/components/monitor.hpp b/allscale/components/monitor.hpp index 3ffc3c8..963e018 100644 --- a/allscale/components/monitor.hpp +++ b/allscale/components/monitor.hpp @@ -71,9 +71,10 @@ namespace allscale { namespace components { task_times task_times_; task_times last_task_times_; std::chrono::high_resolution_clock::time_point last_task_times_sample_; + std::chrono::high_resolution_clock::time_point start_time_; task_times::time_t process_time_; - measure_buffer process_time_buffer_; - static constexpr std::size_t idle_rate_history_count = 1; + measure_buffer process_time_buffer_; + static constexpr std::size_t idle_rate_history_count = 8; std::array idle_rates_; std::size_t idle_rate_idx_; diff --git a/allscale/dashboard.hpp b/allscale/dashboard.hpp index 73670a2..11977e7 100644 --- a/allscale/dashboard.hpp +++ b/allscale/dashboard.hpp @@ -55,7 +55,7 @@ namespace allscale { namespace dashboard float weighted_task_throughput = 0.0f; // percentage of time being idle - float idle_rate = 0.0f; + float idle_rate = 100.0f; // -- network I/O -- diff --git a/src/components/monitor_component.cpp b/src/components/monitor_component.cpp index 3b6e4b5..862247b 100644 --- a/src/components/monitor_component.cpp +++ b/src/components/monitor_component.cpp @@ -55,6 +55,7 @@ namespace allscale { namespace components { monitor::monitor(std::uint64_t rank) : last_task_times_sample_(std::chrono::high_resolution_clock::now()) + , start_time_(std::chrono::high_resolution_clock::now()) , idle_rate_idx_(0) , rank_(rank) , num_localities_(0) @@ -144,25 +145,40 @@ namespace allscale { namespace components { double monitor::get_idle_rate() { - std::lock_guard l(task_times_mtx_); - auto now = std::chrono::high_resolution_clock::now(); - - auto process_time = process_time_; - - auto d1 = std::chrono::duration_cast(process_time - process_time_buffer_.oldest_data()); - auto d2 = std::chrono::duration_cast(now - process_time_buffer_.oldest_time()); - - - double cur_idle_rate = 1. - static_cast(d1.count())/(static_cast(d2.count()) * num_cpus_); - - idle_rates_[idle_rate_idx_] = cur_idle_rate; - idle_rate_idx_ = (idle_rate_idx_ + 1) % idle_rate_history_count; - - // aggregate process time... - process_time_buffer_.push(process_time, now); - - // calucalate the mean idle rate over all as the result - return std::accumulate(idle_rates_.begin(), idle_rates_.end(), 0.0) / idle_rate_history_count; + task_times::time_t process_time; + task_times::time_t duration; + { + std::lock_guard l(task_times_mtx_); + auto now = std::chrono::high_resolution_clock::now(); + process_time = process_time_; + duration = now - start_time_; + // Reset each 10 seconds... + using namespace std::chrono_literals; + if (duration > 10s) + { + process_time_ = task_times::time_t(); + start_time_ = std::chrono::high_resolution_clock::now(); + } + } + return static_cast(process_time.count()) / (duration.count() * num_cpus_); +// auto now = std::chrono::high_resolution_clock::now(); +// +// auto process_time = process_time_; +// +// auto d1 = std::chrono::duration_cast(process_time - process_time_buffer_.oldest_data()); +// auto d2 = std::chrono::duration_cast(now - process_time_buffer_.oldest_time()); +// +// +// double cur_idle_rate = 1. - static_cast(d1.count())/(static_cast(d2.count()) * num_cpus_); +// +// idle_rates_[idle_rate_idx_] = cur_idle_rate; +// idle_rate_idx_ = (idle_rate_idx_ + 1) % idle_rate_history_count; +// +// // aggregate process time... +// process_time_buffer_.push(process_time, now); +// +// // calucalate the mean idle rate over all as the result +// return std::accumulate(idle_rates_.begin(), idle_rates_.end(), 0.0) / idle_rate_history_count; } #ifdef REALTIME_VIZ diff --git a/src/dashboard.cpp b/src/dashboard.cpp index ee326f1..1b3d8c6 100644 --- a/src/dashboard.cpp +++ b/src/dashboard.cpp @@ -33,14 +33,16 @@ namespace allscale { namespace dashboard state.online = true; state.active = scheduler::active(); + + state.ownership = data_item_manager::get_ownership_json(); + if (!state.active) return state; + allscale::components::monitor *monitor_c = &allscale::monitor::get(); // FIXME: add proper metrics here... state.num_cores = monitor_c->get_num_cpus(); state.cpu_load = monitor_c->get_cpu_load(); - state.ownership = data_item_manager::get_ownership_json(); - state.total_memory = monitor_c->get_node_total_memory(); state.memory_load = monitor_c->get_consumed_memory(); state.task_throughput = monitor_c->get_throughput(); @@ -57,8 +59,8 @@ namespace allscale { namespace dashboard state.productive_cycles_per_second = float(state.cur_frequency) * (1.f - state.idle_rate); // freq to Hz - state.speed = 1.f - state.idle_rate; - state.efficiency = state.speed * (float(state.cur_frequency * active_cores) / float(state.max_frequency * state.num_cores)); + state.efficiency = 1.f - state.idle_rate; + state.speed = state.efficiency * (float(state.cur_frequency * active_cores) / float(state.max_frequency * state.num_cores)); #ifdef POWER_ESTIMATE state.cur_power = monitor_c->get_current_power(); @@ -479,6 +481,7 @@ namespace allscale { namespace dashboard float max_power = 0.f; float cur_power = 0.f; + std::size_t active_nodes = 0; for (auto const& cur: state.nodes) { @@ -489,6 +492,7 @@ namespace allscale { namespace dashboard // speeds[cur.rank] = cur.speed; total_efficiency += cur.efficiency; cur_power += cur.cur_power; + active_nodes++; } max_power += cur.max_power; } @@ -496,7 +500,7 @@ namespace allscale { namespace dashboard state.speed = total_speed / client.localities_.size(); // state.speed = std::pow(total_speed, 1.f/client.localities_.size()); - state.efficiency = total_efficiency / client.localities_.size(); + state.efficiency = total_efficiency / active_nodes; state.power = (max_power > 0) ? cur_power/max_power : 0; auto exponents = scheduler::get_optimizer_exponents(); diff --git a/src/data_item_reference.cpp b/src/data_item_reference.cpp index 05b8531..c7466ce 100644 --- a/src/data_item_reference.cpp +++ b/src/data_item_reference.cpp @@ -4,7 +4,7 @@ namespace allscale { namespace detail { std::uint32_t create_data_item_id() { - static hpx::util::atomic_count id(0); + static hpx::util::atomic_count id(1); std::uint32_t this_id = static_cast(++id - 1); std::cout << "creating DIR " << this_id << "\n"; return this_id; diff --git a/src/optimizer.cpp b/src/optimizer.cpp index 19731e8..6a8f0c4 100644 --- a/src/optimizer.cpp +++ b/src/optimizer.cpp @@ -170,7 +170,9 @@ float estimate_power(float frequency) global_optimizer::global_optimizer() : u_balance_every(10), u_steps_till_rebalance(u_balance_every), - active_nodes_(allscale::get_num_localities(), true), tuner_(new simple_coordinate_descent(tuner_configuration{active_nodes_, allscale::monitor::get().get_current_freq(0)})), + active_nodes_(allscale::get_num_localities(), true), + active_frequency_(allscale::monitor::get().get_current_freq(0)), + tuner_(new simple_coordinate_descent(tuner_configuration{active_nodes_, allscale::monitor::get().get_current_freq(0)})), objective_(get_default_objective()), active_(true), localities_(hpx::find_all_localities()), f_resource_max(-1.0f), f_resource_leeway(-1.0f) @@ -216,7 +218,8 @@ void global_optimizer::tune(std::vector const &state) allscale::components::monitor *monitor_c = &allscale::monitor::get(); std::uint64_t max_frequency = monitor_c->get_max_freq(0); - std::size_t num_active_nodes = std::count(active_nodes_.begin(), active_nodes_.end(), true); + std::size_t num_active_nodes = 0; + for (bool x : active_nodes_) if (x) num_active_nodes++; // compute overall speed @@ -230,8 +233,8 @@ void global_optimizer::tune(std::vector const &state) { if (i < num_active_nodes) { - total_speed += state[i].load_; - total_efficiency += state[i].load_ * (float(state[i].active_frequency_ * state[i].cores_per_node_) / float(max_frequency * state[i].cores_per_node_));; + total_efficiency += state[i].load_; + total_speed += state[i].load_ * (float(state[i].active_frequency_ * state[i].cores_per_node_) / float(max_frequency * state[i].cores_per_node_));; used_power += state[i].energy_; } #ifdef POWER_ESTIMATE @@ -242,7 +245,7 @@ void global_optimizer::tune(std::vector const &state) tuner_state tune_state; tune_state.speed = total_speed / active_nodes_.size(); - tune_state.efficiency = total_efficiency / active_nodes_.size(); + tune_state.efficiency = total_efficiency / num_active_nodes; tune_state.power = used_power / max_power; tune_state.score = objective_.score(tune_state.speed, tune_state.efficiency, tune_state.power); @@ -281,7 +284,8 @@ void global_optimizer::tune(std::vector const &state) // compute the load variance float avg_load = 0.0f; - std::size_t num_active_nodes = std::count(active_nodes_.begin(), active_nodes_.end(), true); + std::size_t num_active_nodes = 0; + for (bool x : active_nodes_) if (x) num_active_nodes++; std::for_each(state.begin(), state.end(), [×, &avg_load](optimizer_state const& s) { diff --git a/src/schedule_policy.cpp b/src/schedule_policy.cpp index 35d42c0..3fb972b 100644 --- a/src/schedule_policy.cpp +++ b/src/schedule_policy.cpp @@ -230,7 +230,7 @@ namespace allscale { namespace { - std::unique_ptr rebalance_tasks(tree_scheduling_policy const& old, const std::vector& task_costs, std::vector const& mask) + std::unique_ptr rebalance_tasks(tree_scheduling_policy const& old, const std::vector& task_costs, std::vector const& mask, bool output) { // check input... std::size_t mapping_size = 1 << old.get_granularity(); @@ -250,28 +250,36 @@ namespace allscale { float total_costs = std::accumulate(task_costs.begin(), task_costs.end(), 0.0f); float share = total_costs / available_nodes; +// if (output) +// std::cout << total_costs << " " << share << "\n"; + float cur_costs = 0; float next_goal = share; std::size_t cur_node = 0; while (!mask[cur_node]) cur_node++; - std::vector new_mapping(mapping_size); + std::vector new_mapping(old.task_distribution_mapping()); for (std::size_t i = 0; i < mapping_size; ++i) { // compute next costs auto next_costs = cur_costs + task_costs[i]; +// if (output) +// std::cout << i << " " << cur_node << " " << next_costs << " " << next_goal << "\n"; // if we cross a boundary - if (cur_node < num_nodes - 1) + if (cur_node <= num_nodes) { - if (next_costs > next_goal) -// if (std::abs(cur_costs - next_goal) < std::abs(next_goal - next_costs)) +// if (next_costs > next_goal) + if (std::abs(cur_costs - next_goal) < std::abs(next_goal - next_costs)) { +// if (i > 0 && new_mapping[i-1] == cur_node) + { // std::cout << "advance to node " << cur_node+1 << '\n'; - cur_node++; - while (!mask[cur_node]) cur_node++; - next_goal += share; + cur_node++; + while (!mask[cur_node]) cur_node++; + next_goal += share; + } } } @@ -355,7 +363,7 @@ namespace allscale { task_costs[i] = costs[mapping[i]]; } - return rebalance_tasks(old, task_costs, mask); + return rebalance_tasks(old, task_costs, mask, false); } namespace { @@ -378,214 +386,224 @@ namespace allscale { { tree_scheduling_policy const& old = static_cast(old_base); - // extract task cost vector... - std::size_t mapping_size = 1 << old.get_granularity(); - std::vector task_costs(mapping_size); - - // sample measured times - sample_task_costs(times, task_id::task_path::root(), old.get_granularity(), task_costs); - - std::vector mapping = old.task_distribution_mapping(); - - std::vector node_avg(mask.size(), 0.f); - std::vector node_contribution(mask.size(), 0); - - auto total = std::accumulate(task_costs.begin(), task_costs.end(), 0.0f); - auto avg = total / task_costs.size(); - float sum_dist = 0.f; - for(std::size_t i=0; i node_avg[node] + margin) - { - target_node = node + 1; - // advance to node that hasn't been disabled... - node_contribution[node]++; - node_contribution[target_node]--; - update_mapping = true; - } - return; - } - if (node == node_avg.size() -1) - { - if (node_contribution[node - 1] == 1) return; - if (node_avg[node - 1] > node_avg[node] + margin) - { - target_node = node - 1; - // advance to node that hasn't been disabled... - node_contribution[node]++; - node_contribution[target_node]--; - update_mapping = true; - } - return; - } - - if (node_avg[node - 1] < node_avg[node + 1]) - { - if (node_contribution[node + 1] == 1) return; - if (node_avg[node + 1] > node_avg[node] + margin) - { - target_node = node + 1; - // advance to node that hasn't been disabled... - node_contribution[node]++; - node_contribution[target_node]--; - update_mapping = true; - } - } - else - { - if (node_contribution[node - 1] == 1) return; - if (node_avg[node - 1] > node_avg[node] + margin) - { - target_node = node - 1; - // advance to node that hasn't been disabled... - node_contribution[node]++; - node_contribution[target_node]--; - update_mapping = true; - } - } - }; -// if (var < 0.01f) + // extract task cost vector from task times + std::size_t mappingSize = 1 << old.granularity_; + std::vector taskCosts(mappingSize); + + // sample measured times + sample_task_costs(times, task_id::task_path::root(), old.granularity_, taskCosts); + + // compute new schedule + return rebalance_tasks(old, taskCosts, mask, output); + +// // extract task cost vector... +// std::size_t mapping_size = 1 << old.get_granularity(); +// std::vector task_costs(mapping_size); +// +// // sample measured times +// sample_task_costs(times, task_id::task_path::root(), old.get_granularity(), task_costs); +// +// std::vector mapping = old.task_distribution_mapping(); +// +// std::vector node_avg(mask.size(), 0.f); +// std::vector node_contribution(mask.size(), 0); +// +// auto total = std::accumulate(task_costs.begin(), task_costs.end(), 0.0f); +// auto avg = total / task_costs.size(); +// float sum_dist = 0.f; +// for(std::size_t i=0; i upper) - { - // if (output) std::cout << "new max?\n"; - eligable_max(i); - } - // If this node has lower load than the confidence interval, - // put the work to our neighbors - if (node_avg[i] < lower) - { - // if (output) std::cout << "new min?\n"; - eligable_min(i); - } - } - } -// if (output) std::cout << "\n"; - - if (update_mapping) - { - // Generating new mapping by using the node_contribution. - auto jt = mapping.begin(); - for (std::size_t i = 0; i != node_avg.size(); ++i) - { - for (std::size_t j = 0; j != node_contribution[i]; ++j) - { - *jt = i; - ++jt; - } - } - } - -// if (output) std::cout << mapping << '\n'; - - // create new scheduling policy - auto log2 = old.root().getLayer(); - - return std::unique_ptr(new tree_scheduling_policy( - old.root(), old.get_granularity(), - toDecisionTree((1< node_avg[node] + margin) +// { +// target_node = node + 1; +// // advance to node that hasn't been disabled... +// node_contribution[node]++; +// node_contribution[target_node]--; +// update_mapping = true; +// } +// return; +// } +// if (node == node_avg.size() -1) +// { +// if (node_contribution[node - 1] == 1) return; +// if (node_avg[node - 1] > node_avg[node] + margin) +// { +// target_node = node - 1; +// // advance to node that hasn't been disabled... +// node_contribution[node]++; +// node_contribution[target_node]--; +// update_mapping = true; +// } +// return; +// } +// +// if (node_avg[node - 1] < node_avg[node + 1]) +// { +// if (node_contribution[node + 1] == 1) return; +// if (node_avg[node + 1] > node_avg[node] + margin) +// { +// target_node = node + 1; +// // advance to node that hasn't been disabled... +// node_contribution[node]++; +// node_contribution[target_node]--; +// update_mapping = true; +// } +// } +// else +// { +// if (node_contribution[node - 1] == 1) return; +// if (node_avg[node - 1] > node_avg[node] + margin) +// { +// target_node = node - 1; +// // advance to node that hasn't been disabled... +// node_contribution[node]++; +// node_contribution[target_node]--; +// update_mapping = true; +// } +// } +// }; +// // if (var < 0.01f) +// // { +// // } +// // else +// { +// // Go over all node averages +// for (std::size_t i = 0; i != node_avg.size(); ++i) +// { +// // if(output) std::cout << node_avg[i] << " " << lower << " " << upper << " " << avg << " " << margin << '\n'; +// // If this node has higher load than the confidence interval, +// // put the work to our neighbors +// if (node_avg[i] > upper) +// { +// // if (output) std::cout << "new max?\n"; +// eligable_max(i); +// } +// // If this node has lower load than the confidence interval, +// // put the work to our neighbors +// if (node_avg[i] < lower) +// { +// // if (output) std::cout << "new min?\n"; +// eligable_min(i); +// } +// } +// } +// // if (output) std::cout << "\n"; +// +// if (update_mapping) +// { +// // Generating new mapping by using the node_contribution. +// auto jt = mapping.begin(); +// for (std::size_t i = 0; i != node_avg.size(); ++i) +// { +// for (std::size_t j = 0; j != node_contribution[i]; ++j) +// { +// *jt = i; +// ++jt; +// } +// } +// } +// +// // if (output) std::cout << mapping << '\n'; +// +// // create new scheduling policy +// auto log2 = old.root().getLayer(); +// +// return std::unique_ptr(new tree_scheduling_policy( +// old.root(), old.get_granularity(), +// toDecisionTree((1< diff --git a/src/scheduler.cpp b/src/scheduler.cpp index 95367c6..2ac45ae 100644 --- a/src/scheduler.cpp +++ b/src/scheduler.cpp @@ -744,6 +744,11 @@ namespace allscale } ); + std::lock_guard l(active_mtx_); + { + active_ = mask[hpx::get_locality_id()]; + } + monitor::get().set_cur_freq(freq); }