ComponentThreads: now basics are working.

Next step is to get the unified event loops working generically
and then we can begin region-splitting up the data in harikoff.

We'll assign all the global resource managers to Marionette and
then assign the Mind components to the respective component threads.
This commit is contained in:
2025-01-11 04:34:49 -04:00
parent 876526364b
commit bffc32519b
6 changed files with 115 additions and 69 deletions
+11 -1
View File
@@ -63,7 +63,17 @@
"typeinfo": "cpp",
"variant": "cpp",
"cstring": "cpp",
"cinttypes": "cpp"
"cinttypes": "cpp",
"any": "cpp",
"codecvt": "cpp",
"complex": "cpp",
"coroutine": "cpp",
"csignal": "cpp",
"list": "cpp",
"source_location": "cpp",
"future": "cpp",
"shared_mutex": "cpp",
"typeindex": "cpp"
},
"editor.rulers": [80, 120]
}
+1 -1
View File
@@ -2,5 +2,5 @@ SUBDIRS = deviceManager senseApis
AM_CPPFLAGS+= -I"$(top_srcdir)/hcore/include"
noinst_LIBRARIES = libhcore.a
libhcore_a_SOURCES = mind.cpp opts.cpp componentThreads.cpp
libhcore_a_SOURCES = mind.cpp opts.cpp componentThread.cpp
+70
View File
@@ -0,0 +1,70 @@
#include <iostream>
#include <componentThread.h>
namespace hk {
namespace director {
ComponentThread director;
}
namespace simulator {
ComponentThread canvas;
}
namespace subconscious {
ComponentThread subconscious;
}
std::unordered_map<std::thread::id, ComponentThread&>
ComponentThread::componentThreads =
{
{director::director.thread.get_id(), director::director},
{simulator::canvas.thread.get_id(), simulator::canvas},
{subconscious::subconscious.thread.get_id(), subconscious::subconscious}
};
void ComponentThread::signalThread(std::thread::id id)
{
auto it = componentThreads.find(id);
if (it == componentThreads.end())
{
throw std::runtime_error(std::string(__func__)
+ ": Thread ID not found in componentThreads map");
}
ComponentThread& componentThread = it->second;
{
std::lock_guard<std::mutex> lock(componentThread.startupSync.mutex);
componentThread.startupSync.ready = true;
}
componentThread.startupSync.cv.notify_one();
}
void ComponentThread::main(ComponentThread& self)
{
std::cout << __func__ << ": Entered." << std::endl;
// We sleep on spawn until the main thread tells us to continue.
{
std::unique_lock<std::mutex> lock(self.startupSync.mutex);
self.startupSync.cv.wait(lock, [&self]() {
return self.startupSync.ready;
});
}
std::cout << __func__ << ": Starting event loop." << std::endl;
self.getIoService().run();
std::cout << __func__ << ": Exiting." << std::endl;
}
void ComponentThread::validateThreadIds(void)
{
for (const auto& [id, componentThread] : componentThreads)
{
// std::thread::id() is usable as an invalid ID.
if (id == std::thread::id())
{
throw std::runtime_error(
std::string(__func__) + ": Invalid Thread ID.");
}
}
}
} // namespace hk
-23
View File
@@ -1,23 +0,0 @@
#include <componentThread.h>
namespace hk {
namespace director {
ComponentThread director;
}
namespace simulator {
ComponentThread canvas;
}
namespace subconscious {
ComponentThread subconscious;
}
std::unordered_map<std::thread::id, ComponentThread&>
ComponentThread::componentThreads =
{
{director::director.thread.get_id(), director::director},
{simulator::canvas.thread.get_id(), simulator::canvas},
{subconscious::subconscious.thread.get_id(), subconscious::subconscious}
};
} // namespace hk
+10 -2
View File
@@ -13,7 +13,8 @@ class ComponentThread
{
public:
ComponentThread()
: work(io_service), startupSync()
: work(io_service), startupSync(),
thread(ComponentThread::main, std::ref(*this))
{}
boost::asio::io_service& getIoService(void) { return io_service; }
@@ -31,10 +32,13 @@ public:
return it->second.getIoService();
}
static void main(ComponentThread &self);
static void signalThread(std::thread::id id);
static void validateThreadIds(void);
public:
boost::asio::io_service io_service;
boost::asio::io_service::work work;
std::thread thread;
struct StartupSync {
std::mutex mutex;
std::condition_variable cv;
@@ -43,6 +47,10 @@ public:
StartupSync() : ready(false) {}
} startupSync;
/* Always ensure that this is last so that the thread is spawned after
* everything else.
*/
std::thread thread;
static std::unordered_map<std::thread::id, ComponentThread&> componentThreads;
};
+23 -42
View File
@@ -14,8 +14,6 @@
namespace hk {
static int initializeHarikoff(int argc, char **argv, char **envp);
void startThreads();
void signalThreads();
} // namespace hk
@@ -24,20 +22,29 @@ int main(int argc, char **argv, char **envp)
try {
std::cout << __func__ << ": Entering main()" << std::endl;
// Print out the keys for each index in the map
for (const auto& [id, componentThread]
: hk::ComponentThread::componentThreads)
{
std::cout << __func__ << ": Thread ID: " << id << std::endl;
}
// Validate thread IDs
hk::ComponentThread::validateThreadIds();
int ret = hk::initializeHarikoff(argc, argv, envp);
if (ret != 0) {
return ret;
}
// Signal all threads
for (const auto& [id, componentThread]
: hk::ComponentThread::componentThreads) {
hk::ComponentThread::signalThread(id);
}
// Infinite loop calling yield
while (true) {
std::this_thread::yield();
}
}
catch (const std::exception& e) {
std::cerr << __func__ << ": Exception occurred: " << e.what() << std::endl;
catch (const std::exception& e)
{
std::cerr << __func__ << ": Exception occurred: " << e.what()
<< std::endl;
return EXIT_FAILURE;
}
catch (...) {
@@ -64,8 +71,10 @@ static int initializeHarikoff(int argc, char **argv, char **envp)
try {
options.parseArguments(argc, argv, envp);
}
catch (const std::invalid_argument& e) {
std::cerr << __func__ << ": Exception occurred: " << e.what() << '\n' << options.getUsage() << '\n';
catch (const std::invalid_argument& e)
{
std::cerr << __func__ << ": Exception occurred: " << e.what() << '\n'
<< options.getUsage() << '\n';
return EXIT_FAILURE;
}
@@ -80,38 +89,10 @@ static int initializeHarikoff(int argc, char **argv, char **envp)
std::cout << DeviceManager::stringifyDeviceSpecs() << std::endl;
sense_api::SenseApiManager::getInstance().loadAllSenseApiLibsFromOptions();
/* Start the threads */
std::cout << __func__ << ": Exiting" << std::endl;
return 0;
}
void startThreads()
{
for (auto& [id, componentThread] : ComponentThread::componentThreads) {
componentThread.thread = std::thread([&componentThread]() {
// We sleep on spawn until the marionette tells us to continue.
{
std::unique_lock<std::mutex> lock(
componentThread.startupSync.mutex);
componentThread.startupSync.cv.wait(lock, [&componentThread]() {
return componentThread.startupSync.ready;
});
}
componentThread.getIoService().run();
});
}
}
void signalThreads()
{
for (auto& [id, componentThread] : ComponentThread::componentThreads) {
{
std::lock_guard<std::mutex> lock(componentThread.startupSync.mutex);
componentThread.startupSync.ready = true;
}
componentThread.startupSync.cv.notify_one();
}
}
} // namespace hk