#include #include #include #include #include namespace sscl::probe { namespace { constexpr sscl::ThreadId PROBE_PUPPETEER_THREAD_ID = 2; class ProbeDummyPuppeteerComponent : public sscl::pptr::PuppeteerComponent { public: explicit ProbeDummyPuppeteerComponent( const std::shared_ptr& componentThreadIn) : sscl::pptr::PuppeteerComponent(componentThreadIn) {} void handleLoopExceptionHook() override { std::cerr << "ProbeComponentThreadHarness: puppeteer loop exception\n"; } }; /** EXPLANATION: * PuppeteerThread starts its std::thread inside the constructor, but * enable_shared_from_this::weak_this is only armed after make_shared returns. * Without a barrier, initializeTls()'s shared_from_this() races and throws * std::bad_weak_ptr. DedicatedIoThread uses the same handshake. */ struct ProbeThreadStartupState { std::mutex mutex; std::condition_variable condition; bool allowInitialization = false; }; void waitForProbeThreadStartupPermission( const std::shared_ptr& startupState) { std::unique_lock lock(startupState->mutex); startupState->condition.wait( lock, [&startupState]() { return startupState->allowInitialization; }); } void releaseProbeThreadStartupBarrier( const std::shared_ptr& startupState) { { std::lock_guard guard(startupState->mutex); startupState->allowInitialization = true; } startupState->condition.notify_all(); } void probePuppeteerMain( const sscl::PuppeteerThread::EntryFnArguments& args, const std::function&)>& work, std::promise& donePromise, const std::shared_ptr& startupState) { waitForProbeThreadStartupPermission(startupState); sscl::PuppeteerThread& thr = args.usableBeforeJolt; thr.initializeTls(); sscl::ComponentThread::setPuppeteerThreadId(PROBE_PUPPETEER_THREAD_ID); std::shared_ptr thrPtr = std::static_pointer_cast(thr.shared_from_this()); sscl::ComponentThread::setPuppeteerThread(thrPtr); try { work(thrPtr); donePromise.set_value(nullptr); } catch (...) { donePromise.set_value(std::current_exception()); } thr.getIoContext().stop(); } } // namespace ProbeComponentThreadHarness::ProbeComponentThreadHarness( const char *threadName) : threadName(threadName), dummyComponent(std::make_shared( std::shared_ptr())) {} ProbeComponentThreadHarness::~ProbeComponentThreadHarness() = default; std::shared_ptr ProbeComponentThreadHarness::componentThread() const { return lastComponentThread; } void ProbeComponentThreadHarness::runSync( const std::function&)>& work) { std::promise donePromise; std::future doneFuture = donePromise.get_future(); auto startupState = std::make_shared(); std::shared_ptr runThread = std::make_shared( PROBE_PUPPETEER_THREAD_ID, threadName, [&work, &donePromise, startupState]( const sscl::PuppeteerThread::EntryFnArguments& args) { probePuppeteerMain(args, work, donePromise, startupState); }, *dummyComponent, nullptr); dummyComponent->thread = runThread; lastComponentThread = runThread; releaseProbeThreadStartupBarrier(startupState); runThread->thread.join(); std::exception_ptr probeException = doneFuture.get(); if (probeException) { std::rethrow_exception(probeException); } } void runNonViralNurseryOnComponentThread( const std::shared_ptr& componentThread, std::function invokerFactory, std::chrono::milliseconds timeout) { (void)timeout; sscl::co::NonViralTaskNursery nursery; nursery.openAdmission(); nursery.launch( [&invokerFactory](sscl::co::NonViralTaskNursery::Slot::Lease& lease) { return invokerFactory(lease); }); nursery.closeAdmission(); nursery.syncAwaitAllSettlements(componentThread->getIoContext()); } } // namespace sscl::probe