#include #include #include #include #include #include #include #include #include #include #include #include namespace sscl { namespace mrntt { /* Global variable to store the marionette thread ID * Default value is 0, but should be set by application code via * setMarionetteThreadId(). */ ThreadId marionetteThreadId = 0; /* Global marionette thread instance - defined here but initialized by * application code. */ std::shared_ptr thread; void setMarionetteThreadId(ThreadId id) { marionetteThreadId = id; } } // namespace mrntt thread_local std::shared_ptr thisComponentThread; // Implementation of static method std::shared_ptr ComponentThread::getMrntt() { return sscl::mrntt::thread; } void MarionetteThread::initializeTls(void) { thisComponentThread = shared_from_this(); } void PuppetThread::initializeTls(void) { thisComponentThread = shared_from_this(); } bool ComponentThread::tlsInitialized(void) { return thisComponentThread != nullptr; } const std::shared_ptr ComponentThread::getSelf(void) { if (!thisComponentThread) { throw std::runtime_error(std::string(__func__) + ": TLS not initialized"); } return thisComponentThread; } class PuppetThread::ThreadLifetimeMgmtOp : public PostedAsynchronousContinuation { public: ThreadLifetimeMgmtOp( const std::shared_ptr &caller, const std::shared_ptr &target, Callback callback) : PostedAsynchronousContinuation( caller, callback), target(target) {} public: const std::shared_ptr target; public: void joltThreadReq1_posted( [[maybe_unused]] std::shared_ptr context ) { std::cout << __func__ << ": Thread '" << target->name << "': handling " "JOLT request." << "\n"; target->io_service.stop(); callOriginalCb(); } void startThreadReq1_posted( [[maybe_unused]] std::shared_ptr context ) { std::cout << __func__ << ": Thread '" << target->name << "': handling " "startThread." << "\n"; // Execute private setup sequence here // This is where each thread would implement its specific initialization callOriginalCb(); } void exitThreadReq1_mainQueue_posted( [[maybe_unused]] std::shared_ptr context ) { std::cout << __func__ << ": Thread '" << target->name << "': handling " "exitThread (main queue)." << "\n"; target->cleanup(); target->io_service.stop(); callOriginalCb(); } void exitThreadReq1_pauseQueue_posted( [[maybe_unused]] std::shared_ptr context ) { std::cout << __func__ << ": Thread '" << target->name << "': handling " "exitThread (pause queue)."<< "\n"; target->cleanup(); target->pause_io_service.stop(); target->io_service.stop(); callOriginalCb(); } void pauseThreadReq1_posted( [[maybe_unused]] std::shared_ptr context ) { std::cout << __func__ << ": Thread '" << target->name << "': handling " "pauseThread." << "\n"; /* We have to invoke the callback here before moving on because * our next operation is going to block the thread, so it won't * have a chance to invoke the callback until it's unblocked. */ callOriginalCb(); target->pause_io_service.reset(); target->pause_io_service.run(); } void resumeThreadReq1_posted( [[maybe_unused]] std::shared_ptr context ) { std::cout << __func__ << ": Thread '" << target->name << "': handling " "resumeThread." << "\n"; target->pause_io_service.stop(); callOriginalCb(); } }; void ComponentThread::cleanup(void) { this->keepLooping = false; } void PuppetThread::joltThreadReq( const std::shared_ptr& selfPtr, Callback callback) { /** EXPLANATION: * We can't use shared_from_this() here because JOLTing occurs prior to * TLS being set up. * * We also can't use getSelf() as yet for the same reason: getSelf() * requires TLS to be set up. * * To obtain a sh_ptr to the caller, we just supply the mrntt thread since * JOLT is always invoked by the mrntt thread. The JOLT sequence that the * CRT main() function invokes on the mrntt thread is special since it * supplies cmdline args and envp. * * To obtain a sh_ptr to the target thread, we use the selfPtr parameter * passed in by the caller. */ if (id == sscl::mrntt::marionetteThreadId) { throw std::runtime_error(std::string(__func__) + ": invoked on mrntt thread"); } std::shared_ptr mrntt = sscl::mrntt::thread; auto request = std::make_shared( mrntt, selfPtr, callback); this->getIoService().post( STC(std::bind( &ThreadLifetimeMgmtOp::joltThreadReq1_posted, request.get(), request))); } // Thread management method implementations void PuppetThread::startThreadReq(Callback callback) { std::shared_ptr caller = getSelf(); auto request = std::make_shared( caller, std::static_pointer_cast(shared_from_this()), callback); this->getIoService().post( STC(std::bind( &ThreadLifetimeMgmtOp::startThreadReq1_posted, request.get(), request))); } void PuppetThread::exitThreadReq(Callback callback) { std::shared_ptr caller = getSelf(); auto request = std::make_shared( caller, std::static_pointer_cast(shared_from_this()), callback); this->getIoService().post( STC(std::bind( &ThreadLifetimeMgmtOp::exitThreadReq1_mainQueue_posted, request.get(), request))); pause_io_service.post( STC(std::bind( &ThreadLifetimeMgmtOp::exitThreadReq1_pauseQueue_posted, request.get(), request))); } void PuppetThread::pauseThreadReq(Callback callback) { if (id == sscl::mrntt::marionetteThreadId) { throw std::runtime_error(std::string(__func__) + ": invoked on mrntt thread"); } std::shared_ptr caller = getSelf(); auto request = std::make_shared( caller, std::static_pointer_cast(shared_from_this()), callback); this->getIoService().post( STC(std::bind( &ThreadLifetimeMgmtOp::pauseThreadReq1_posted, request.get(), request))); } void PuppetThread::resumeThreadReq(Callback callback) { if (id == sscl::mrntt::marionetteThreadId) { throw std::runtime_error(std::string(__func__) + ": invoked on mrntt thread"); } // Post to the pause_io_service to unblock the paused thread std::shared_ptr caller = getSelf(); auto request = std::make_shared( caller, std::static_pointer_cast(shared_from_this()), callback); pause_io_service.post( STC(std::bind( &ThreadLifetimeMgmtOp::resumeThreadReq1_posted, request.get(), request))); } // CPU management method implementations int ComponentThread::getAvailableCpuCount() { int cpuCount = sysconf(_SC_NPROCESSORS_ONLN); if (cpuCount <= 0) { throw std::runtime_error(std::string(__func__) + ": Failed to determine CPU count"); } // Check if std::thread::hardware_concurrency() matches sysconf result unsigned int hwConcurrency = std::thread::hardware_concurrency(); if (hwConcurrency != static_cast(cpuCount)) { std::cerr << "Warning: CPU count mismatch - " "std::thread::hardware_concurrency() = " << hwConcurrency << ", sysconf(_SC_NPROCESSORS_ONLN) = " << cpuCount << "\n"; } return cpuCount; } void PuppetThread::pinToCpu(int cpuId) { if (cpuId < 0) { throw std::runtime_error(std::string(__func__) + ": Invalid CPU ID: " + std::to_string(cpuId)); } cpu_set_t cpuset; CPU_ZERO(&cpuset); CPU_SET(cpuId, &cpuset); int result = pthread_setaffinity_np( thread.native_handle(), sizeof(cpu_set_t), &cpuset); if (result != 0) { throw std::runtime_error(std::string(__func__) + ": Failed to pin thread to CPU " + std::to_string(cpuId) + ": " + std::strerror(result)); } pinnedCpuId = cpuId; } } // namespace sscl