Skip to content

Commit

Permalink
Merge pull request #5064 from msimberg/rp-cmdline-separate
Browse files Browse the repository at this point in the history 
Refactor runtime configuration, command line handling, and resource partitioner
  • Loading branch information
@msimberg
msimberg authored Jan 5, 2021
2 parents e2522aa + de22c4f commit 8cb1829
Show file tree
Hide file tree
Showing 18 changed files with 334 additions and 540 deletions.
5 changes: 1 addition & 4 deletions hpx/runtime/agas/addressing_service.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -124,10 +124,7 @@ struct HPX_EXPORT addressing_service
resolved_localities_type;
resolved_localities_type resolved_localities_;

addressing_service(
util::runtime_configuration const& ini_
, runtime_mode runtime_type_
);
explicit addressing_service(util::runtime_configuration const& ini_);

#if defined(HPX_HAVE_NETWORKING)
~addressing_service()
Expand Down
8 changes: 7 additions & 1 deletion libs/core/affinity/include/hpx/affinity/affinity_data.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ namespace hpx { namespace threads { namespace policies { namespace detail {
affinity_data();
~affinity_data();

std::size_t init(std::size_t num_threads = 1, std::size_t max_cores = 1,
void init(std::size_t num_threads = 1, std::size_t max_cores = 1,
std::size_t pu_offset = 0, std::size_t pu_step = 1,
std::size_t used_cores = 0, std::string affinity_domain = "pu",
std::string const& affinity_description = "balanced",
Expand Down Expand Up @@ -79,6 +79,11 @@ namespace hpx { namespace threads { namespace policies { namespace detail {
void add_punit(std::size_t virt_core, std::size_t thread_num);
void init_cached_pu_nums(std::size_t hardware_concurrency);

std::size_t get_num_pus_needed() const
{
return num_pus_needed_;
}

protected:
std::size_t get_pu_num(
std::size_t num_thread, std::size_t hardware_concurrency) const;
Expand All @@ -96,6 +101,7 @@ namespace hpx { namespace threads { namespace policies { namespace detail {
no_affinity_; ///< mask of processing units which have no affinity
bool
use_process_mask_; ///< use the process CPU mask to limit available PUs
std::size_t num_pus_needed_;
static std::atomic<int>
instance_number_counter_; ///< counter for instance numbers
};
Expand Down
10 changes: 6 additions & 4 deletions libs/core/affinity/src/affinity_data.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ namespace hpx { namespace threads { namespace policies { namespace detail {
, pu_nums_()
, no_affinity_()
, use_process_mask_(false)
, num_pus_needed_(0)
{
threads::resize(no_affinity_, hardware_concurrency());
}
Expand All @@ -50,9 +51,9 @@ namespace hpx { namespace threads { namespace policies { namespace detail {
--instance_number_counter_;
}

std::size_t affinity_data::init(std::size_t num_threads,
std::size_t max_cores, std::size_t pu_offset, std::size_t pu_step,
std::size_t used_cores, std::string affinity_domain, // -V813
void affinity_data::init(std::size_t num_threads, std::size_t max_cores,
std::size_t pu_offset, std::size_t pu_step, std::size_t used_cores,
std::string affinity_domain, // -V813
std::string const& affinity_description, bool use_process_mask)
{
num_threads_ = num_threads;
Expand Down Expand Up @@ -143,7 +144,8 @@ namespace hpx { namespace threads { namespace policies { namespace detail {
cores.erase(it, cores.end());

std::size_t num_unique_cores = cores.size();
return (std::max)(num_unique_cores, max_cores);

num_pus_needed_ = (std::max)(num_unique_cores, max_cores);
}

mask_cref_type affinity_data::get_pu_mask(
Expand Down
34 changes: 13 additions & 21 deletions libs/full/command_line_handling/include/hpx/command_line_handling/command_line_handling.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,28 +21,16 @@
///////////////////////////////////////////////////////////////////////////////
namespace hpx { namespace util {

struct command_line_handling;

///////////////////////////////////////////////////////////////////////////
// Helper functions for retrieving command line options (with error
// checking)
std::size_t get_num_high_priority_queues(
util::command_line_handling const& cfg, std::size_t num_threads);

std::string get_affinity_domain(util::command_line_handling const& cfg);

std::size_t get_affinity_description(
util::command_line_handling const& cfg, std::string& affinity_desc);

std::size_t get_pu_offset(util::command_line_handling const& cfg);

std::size_t get_pu_step(util::command_line_handling const& cfg);

///////////////////////////////////////////////////////////////////////////
struct command_line_handling
{
command_line_handling()
: rtcfg_(nullptr, runtime_mode::default_)
command_line_handling(runtime_configuration rtcfg,
std::vector<std::string> ini_config,
function_nonser<int(hpx::program_options::variables_map& vm)>
hpx_main_f)
: rtcfg_(rtcfg)
, ini_config_(ini_config)
, hpx_main_f_(hpx_main_f)
, node_(std::size_t(-1))
, num_threads_(1)
, num_cores_(1)
Expand All @@ -54,7 +42,6 @@ namespace hpx { namespace util {
, cmd_line_parsed_(false)
, info_printed_(false)
, version_printed_(false)
, parse_result_(0)
{
}

Expand Down Expand Up @@ -84,9 +71,14 @@ namespace hpx { namespace util {
bool cmd_line_parsed_;
bool info_printed_;
bool version_printed_;
int parse_result_;

protected:
// Helper functions for checking command line options
void check_affinity_domain() const;
void check_affinity_description() const;
void check_pu_offset() const;
void check_pu_step() const;

bool handle_arguments(util::manage_config& cfgmap,
hpx::program_options::variables_map& vm,
std::vector<std::string>& ini_config, std::size_t& node,
Expand Down
141 changes: 72 additions & 69 deletions libs/full/command_line_handling/src/command_line_handling.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -562,104 +562,62 @@ namespace hpx { namespace util {
} // namespace detail

///////////////////////////////////////////////////////////////////////
std::size_t get_num_high_priority_queues(
util::command_line_handling const& cfg, std::size_t num_threads)
void command_line_handling::check_affinity_domain() const
{
std::size_t num_high_priority_queues = num_threads;
if (cfg.vm_.count("hpx:high-priority-threads"))
if (affinity_domain_ != "pu")
{
num_high_priority_queues =
cfg.vm_["hpx:high-priority-threads"].as<std::size_t>();
if (num_high_priority_queues > num_threads)
{
throw hpx::detail::command_line_error(
"Invalid command line option: "
"number of high priority threads ("
"--hpx:high-priority-threads), should not be larger "
"than number of threads (--hpx:threads)");
}
}
return num_high_priority_queues;
}

///////////////////////////////////////////////////////////////////////
std::string get_affinity_domain(util::command_line_handling const& cfg)
{
std::string affinity_domain("pu");
if (cfg.affinity_domain_ != "pu")
{
affinity_domain = cfg.affinity_domain_;
if (0 != std::string("pu").find(affinity_domain) &&
0 != std::string("core").find(affinity_domain) &&
0 != std::string("numa").find(affinity_domain) &&
0 != std::string("machine").find(affinity_domain))
if (0 != std::string("pu").find(affinity_domain_) &&
0 != std::string("core").find(affinity_domain_) &&
0 != std::string("numa").find(affinity_domain_) &&
0 != std::string("machine").find(affinity_domain_))
{
throw hpx::detail::command_line_error(
"Invalid command line option "
"--hpx:affinity, value must be one of: pu, core, numa, "
"or machine.");
}
}
return affinity_domain;
}

std::size_t get_affinity_description(
util::command_line_handling const& cfg, std::string& affinity_desc)
void command_line_handling::check_affinity_description() const
{
if (cfg.affinity_bind_.empty())
return cfg.numa_sensitive_;
if (affinity_bind_.empty())
{
return;
}

if (!(cfg.pu_offset_ == std::size_t(-1) ||
cfg.pu_offset_ == std::size_t(0)) ||
cfg.pu_step_ != 1 || cfg.affinity_domain_ != "pu")
if (!(pu_offset_ == std::size_t(-1) || pu_offset_ == std::size_t(0)) ||
pu_step_ != 1 || affinity_domain_ != "pu")
{
throw hpx::detail::command_line_error(
"Command line option --hpx:bind "
"should not be used with --hpx:pu-step, --hpx:pu-offset, "
"or --hpx:affinity.");
}

affinity_desc = cfg.affinity_bind_;
return cfg.numa_sensitive_;
}

std::size_t get_pu_offset(util::command_line_handling const& cfg)
void command_line_handling::check_pu_offset() const
{
std::size_t pu_offset = std::size_t(-1);

if (cfg.pu_offset_ != std::size_t(-1))
if (pu_offset_ != std::size_t(-1) &&
pu_offset_ >= hpx::threads::hardware_concurrency())
{
pu_offset = cfg.pu_offset_;
if (pu_offset >= hpx::threads::hardware_concurrency())
{
throw hpx::detail::command_line_error(
"Invalid command line option "
"--hpx:pu-offset, value must be smaller than number of "
"available processing units.");
}
throw hpx::detail::command_line_error(
"Invalid command line option "
"--hpx:pu-offset, value must be smaller than number of "
"available processing units.");
}

return pu_offset;
}

std::size_t get_pu_step(util::command_line_handling const& cfg)
void command_line_handling::check_pu_step() const
{
std::size_t pu_step = 1;

if (cfg.pu_step_ != 1)
if (pu_step_ == 0 || pu_step_ >= hpx::threads::hardware_concurrency())
{
pu_step = cfg.pu_step_;
if (pu_step == 0 || pu_step >= hpx::threads::hardware_concurrency())
{
throw hpx::detail::command_line_error(
"Invalid command line option "
"--hpx:pu-step, value must be non-zero and smaller "
"than "
"number of available processing units.");
}
throw hpx::detail::command_line_error(
"Invalid command line option "
"--hpx:pu-step, value must be non-zero and smaller "
"than "
"number of available processing units.");
}

return pu_step;
}

///////////////////////////////////////////////////////////////////////////
Expand Down Expand Up @@ -794,6 +752,9 @@ namespace hpx { namespace util {
}
#endif

ini_config.emplace_back(
"hpx.use_process_mask!=" + std::to_string(use_process_mask_));

#if defined(HPX_HAVE_MODULE_MPI_BASE)
bool have_mpi = util::mpi_environment::check_mpi_environment(rtcfg_);
#else
Expand Down Expand Up @@ -1021,6 +982,8 @@ namespace hpx { namespace util {
affinity_domain_ = detail::handle_affinity(cfgmap, vm, "pu");
ini_config.emplace_back("hpx.affinity=" + affinity_domain_);

check_affinity_domain();

affinity_bind_ = detail::handle_affinity_bind(cfgmap, vm, "");
if (!affinity_bind_.empty())
{
Expand Down Expand Up @@ -1048,6 +1011,8 @@ namespace hpx { namespace util {
#endif
ini_config.emplace_back("hpx.pu_step=" + std::to_string(pu_step_));

check_pu_step();

pu_offset_ = detail::handle_pu_offset(cfgmap, vm, std::size_t(-1));

// NOLINTNEXTLINE(bugprone-branch-clone)
Expand All @@ -1070,6 +1035,8 @@ namespace hpx { namespace util {
ini_config.emplace_back("hpx.pu_offset=0");
}

check_pu_offset();

numa_sensitive_ = detail::handle_numa_sensitive(
cfgmap, vm, affinity_bind_.empty() ? 0 : 1);
ini_config.emplace_back(
Expand All @@ -1088,6 +1055,8 @@ namespace hpx { namespace util {
ini_config.emplace_back("hpx.bind!=" + affinity_bind_);
}

check_affinity_description();

// handle number of cores and threads
num_threads_ = detail::handle_num_threads(cfgmap, rtcfg_, vm, env,
using_nodelist, initial, use_process_mask_);
Expand All @@ -1099,6 +1068,32 @@ namespace hpx { namespace util {
"hpx.os_threads=" + std::to_string(num_threads_));
ini_config.emplace_back("hpx.cores=" + std::to_string(num_cores_));

if (vm_.count("hpx:high-priority-threads"))
{
std::size_t num_high_priority_queues =
vm_["hpx:high-priority-threads"].as<std::size_t>();
if (num_high_priority_queues != std::size_t(-1) &&
num_high_priority_queues > num_threads_)
{
throw hpx::detail::command_line_error(
"Invalid command line option: "
"number of high priority threads ("
"--hpx:high-priority-threads), should not be larger "
"than number of threads (--hpx:threads)");
}

if (!(queuing_ == "local-priority" || queuing_ == "abp-priority"))
{
throw hpx::detail::command_line_error(
"Invalid command line option --hpx:high-priority-threads, "
"valid for --hpx:queuing=local-priority and "
"--hpx:queuing=abp-priority only");
}

ini_config.emplace_back("hpx.thread_queue.high_priority_queues!=" +
std::to_string(num_high_priority_queues));
}

// map host names to ip addresses, if requested
hpx_host = mapnames.map(hpx_host, hpx_port);
agas_host = mapnames.map(agas_host, agas_port);
Expand Down Expand Up @@ -1541,7 +1536,9 @@ namespace hpx { namespace util {
// handle all --hpx:foo options, determine node
std::vector<std::string> ini_config; // discard
if (!handle_arguments(cfgmap, prevm, ini_config, node_, true))
{
return -2;
}

// re-initialize runtime configuration object
if (prevm.count("hpx:config"))
Expand Down Expand Up @@ -1619,7 +1616,9 @@ namespace hpx { namespace util {

// handle all --hpx:foo and --hpx:*:foo options
if (!handle_arguments(cfgmap, vm_, ini_config_, node_))
{
return -2;
}

// store unregistered command line and arguments
store_command_line(argc, argv);
Expand All @@ -1630,7 +1629,9 @@ namespace hpx { namespace util {

// help can be printed only after the runtime mode has been set
if (handle_help_options(help))
{
return 1; // exit application gracefully
}

// print version/copyright information
if (vm_.count("hpx:version"))
Expand All @@ -1640,6 +1641,7 @@ namespace hpx { namespace util {
detail::print_version(std::cout);
version_printed_ = true;
}

return 1;
}

Expand All @@ -1651,6 +1653,7 @@ namespace hpx { namespace util {
detail::print_info(std::cout, *this);
info_printed_ = true;
}

return 1;
}

Expand Down
Loading

0 comments on commit 8cb1829

Please sign in to comment.