Why can I only detect only one locality ? #6516
-
|
My code is as follows: hpx::this_thread::sleep_for(5s);
INFO("%s", __FUNCTION__);
std::vector<hpx::id_type> localities = hpx::find_all_localities();
if (localities.size() < 2)
{
ERROR("this program requires at least two localities");
return;
}When compiling HPX, I used HPX_WITH_PARCELPORT_TCP=ON. I ran the same program on two hosts in the same local network, but hpx::find_all_localities() cannot detect the locality on the other host. I checked the HPX documentation and added some command line parameters, such as: HPX options (additionally allowed in an options file) --hpx:console --hpx:connect --hpx:run-agas-server --hpx:run-hpx-main --hpx:hpx arg --hpx:agas arg However, I still can't detect localities on other hosts. What should I do? |
Beta Was this translation helpful? Give feedback.
Replies: 1 comment 11 replies
-
|
Please see here for an explanation of what's going on and how to resolve the problem: https://stackoverflow.com/questions/35367816/hpx-minimal-two-node-example-set-up |
Beta Was this translation helpful? Give feedback.
-
|
Beta Was this translation helpful? Give feedback.
-
here is my log: ./HpxDemo.elf --hpx:worker --hpx:agas 127.0.0.1:4001 --hpx:hpx 127.0.0.1:4002 --hpx:threads 2 Please note the timestamp when the myTask execution completes in node0.log and the timestamp when the myTask execution begins in node1.log. node1.log: What is the internal implementation principle of hpx::get_locality_name(loc).get();? If its asynchronous execution is normal, is it possible that I missed some compilation options when compiling the HPX library? |
Beta Was this translation helpful? Give feedback.
-
|
I think this phenomenon might be caused by an error in my code above. I slightly modified the code from examples/quickstart/hello_world_distributed.cpp and pasted it into the hpx_main function, overwriting my original code. When running in the same environment, the asynchronous function call auto locName = hpx::get_locality_name(node).get(); no longer blocks. ///////////////////////////////////////////////////////////////////////////////
// Copyright (c) 2007-2014 Hartmut Kaiser
// Copyright (c) 2011 Bryce Adelstein-Lelbach
//
// SPDX-License-Identifier: BSL-1.0
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
///////////////////////////////////////////////////////////////////////////////
// Including 'hpx/hpx_main.hpp' instead of the usual 'hpx/hpx_init.hpp' enables
// to use the plain C-main below as the direct main HPX entry point.
#include <hpx/config.hpp>
// #include <hpx/hpx_main.hpp>
#include <hpx/include/actions.hpp>
#include <hpx/include/components.hpp>
#include <hpx/include/lcos.hpp>
#include <hpx/include/parallel_executors.hpp>
#include <hpx/include/runtime.hpp>
#include <hpx/include/util.hpp>
#include <hpx/iostream.hpp>
#include "hpx/hpx_start.hpp"
#include "hpx/version.hpp"
#include "hpx/init.hpp"
#include <cstddef>
#include <cstdint>
#include <list>
#include <mutex>
#include <set>
#include <vector>
#include "hpx/thread.hpp"
#include <chrono>
using namespace std;
std::size_t hello_world_worker(std::size_t desired)
{
std::size_t current = hpx::get_worker_thread_num();
if (current == desired)
{
char const* msg = "hello world from OS-thread {1} on locality {2}\n";
// hpx::util::format_to(hpx::cout, msg, desired, hpx::get_locality_id())
// << std::flush;
printf("%s\n", __FUNCTION__);
auto info = hpx::util::format(msg, desired, hpx::get_locality_id());
printf("%s", info.c_str());
hpx::this_thread::sleep_for(10s);
printf("%s exit\n", __FUNCTION__);
return desired;
}
return std::size_t(-1);
}
void hello_world_foreman()
{
std::size_t const os_threads = hpx::get_os_thread_count();
std::set<std::size_t> attendance;
for (std::size_t os_thread = 0; os_thread < os_threads; ++os_thread)
attendance.insert(os_thread);
while (!attendance.empty())
{
std::vector<hpx::future<std::size_t>> futures;
futures.reserve(attendance.size());
for (std::size_t worker : attendance)
{
hpx::execution::parallel_executor exec(
hpx::threads::thread_priority::bound);
hpx::threads::thread_schedule_hint hint(
hpx::threads::thread_schedule_hint_mode::thread,
static_cast<std::int16_t>(worker));
futures.push_back(
hpx::async(hpx::execution::experimental::with_hint(exec, hint),
hello_world_worker, worker));
}
hpx::spinlock mtx;
hpx::wait_each(hpx::unwrapping([&](std::size_t t) {
if (std::size_t(-1) != t)
{
std::lock_guard<hpx::spinlock> lk(mtx);
attendance.erase(t);
}
}),
futures);
}
}
HPX_PLAIN_ACTION(hello_world_foreman, hello_world_foreman_action)
int hpx_main(int argc, char* argv[])
{
std::vector<hpx::id_type> localities = hpx::find_all_localities();
std::vector<hpx::future<void>> futures;
futures.reserve(localities.size());
for (hpx::id_type const& node : localities)
{
auto locName = hpx::get_locality_name(node).get();
printf("locality name: %s\n", locName.c_str());
typedef hello_world_foreman_action action_type;
futures.push_back(hpx::async<action_type>(node));
}
hpx::wait_all(futures);
return hpx::finalize();
}
int main(int argc, char **argv)
{
printf("hpx demostration running...\n");
auto ret = hpx::start(argc, argv);
if (false == ret)
{
printf("hpx runtime init failed\n");
return -3;
}
printf("enter any key to exit\n");
getchar();
printf("hpx demostration exiting...\n");
return hpx::stop();
} |
Beta Was this translation helpful? Give feedback.
-
|
Based on the previous tests, I seem to have discovered an interesting phenomenon. I modified the myTask function, while keeping the rest of the code unchanged. The rest code is the main.cpp and TestHpx.cpp that I pasted earlier (#6516 (reply in thread)). Now, my myTask function has three scenarios: // compute-intensive task.
static int myTask(const int id)
static int myTask(int id)
{
TRACE("%s", __FUNCTION__);
double result = 0.0, ts = 0.00;
let ticker = Ticker();
ticker.restart();
const double computeTime = 20'000.00;
#if 0
while (true)
{
// perform a large number of floating-point operations to ensure the CPU is fully loaded.
for (int i = 0; i < 1000000; ++i)
{
result += std::sin(i) * std::cos(i);
}
// check if it has been running for more than 'computeTime' milliseconds.
ts = ticker.elapsed(Ticker::TimePrecision::PrecisionMs);
if (ts > computeTime)
{
break;
}
}
#elif 1
std::this_thread::sleep_for(10s);
#else
hpx::this_thread::sleep_for(10s);
#endif
TRACE("task %d completed in %0.2f ms", id, ts);
return id;
}In the myTask function, the original while loop was simulating a compute-intensive task that would fully load the CPU. Later, I conducted multiple tests using std::this_thread::sleep_for or hpx::this_thread::sleep_for. Based on the test results, I can confirm:
These test results have made me even more confused. Could you please help me understand this? |
Beta Was this translation helpful? Give feedback.
-
|
@hkaiser Could you please take a moment to look at this code for me? |
Beta Was this translation helpful? Give feedback.
Please see here for an explanation of what's going on and how to resolve the problem: https://stackoverflow.com/questions/35367816/hpx-minimal-two-node-example-set-up