#include "threadpool.hpp" #ifndef _MSC_VER #include #endif #include #include #include #include #include "logger.hpp" #if(__FreeBSD__) || (__OpenBSD__) || (__NetBSD__) #include #endif #ifdef __linux__ #include #endif #ifdef _MSC_VER #include extern "C" void SetThreadName(DWORD dwThreadID, LPCSTR szThreadName); #endif namespace llarp { namespace thread { void Pool::Spawn(size_t workers, const char *name) { stop = false; while(workers--) { threads.emplace_back([this, name] { if(name) { #if(__APPLE__ && __MACH__) pthread_setname_np(name); #elif(__FreeBSD__) || (__OpenBSD__) || (__NetBSD__) pthread_set_name_np(pthread_self(), name); #elif !defined(_MSC_VER) || !defined(_WIN32) pthread_setname_np(pthread_self(), name); #else SetThreadName(GetCurrentThreadId(), name); #endif } for(;;) { llarp_thread_job *job; { lock_t lock(this->queue_mutex); this->condition.WaitUntil( lock, [this] { return this->stop || !this->jobs.empty(); }); if(this->stop) { // discard pending jobs while(this->jobs.size()) { delete this->jobs.top().job; this->jobs.pop(); } return; } job = this->jobs.top().job; this->jobs.pop(); } // do work job->work(job->user); delete job; } }); } } void Pool::Stop() { { lock_t lock(queue_mutex); stop = true; } condition.NotifyAll(); } void Pool::Join() { for(auto &t : threads) t.join(); threads.clear(); done.NotifyAll(); } void Pool::QueueJob(const llarp_thread_job &job) { { lock_t lock(queue_mutex); // don't allow enqueueing after stopping the pool if(stop) return; jobs.emplace(ids++, new llarp_thread_job(job.user, job.work)); } condition.NotifyOne(); } static int runIsolated(void *arg) { IsolatedPool *self = static_cast< IsolatedPool * >(arg); if(!self->Isolated()) { llarp::LogError("failed to set up isolated environment"); return 1; } auto func = std::bind(&Pool::Spawn, self, self->m_IsolatedWorkers, self->m_IsolatedName); func(); return 0; } void IsolatedPool::Spawn(int workers, const char *name) { if(m_isolated) return; #ifdef __linux__ IsolatedPool *self = this; self->m_IsolatedName = name; self->m_IsolatedWorkers = workers; m_isolated = new std::thread([self] { pid_t isolated; isolated = clone(runIsolated, self->m_childstack + sizeof(self->m_childstack), self->m_flags | SIGCHLD, self); if(isolated == -1) { llarp::LogError("failed to run isolated threadpool, ", strerror(errno)); return; } llarp::LogInfo("Spawned isolated process pool"); if(waitpid(isolated, nullptr, 0) == -1) { llarp::LogError("failed to wait for pid ", isolated, ", ", strerror(errno)); } }); #else llarp::LogError("isolated processes not supported on your platform"); Pool::Spawn(workers, name); #endif } void IsolatedPool::Join() { Pool::Join(); if(m_isolated) { m_isolated->join(); delete m_isolated; m_isolated = nullptr; } } #ifdef __linux__ NetIsolatedPool::NetIsolatedPool(std::function< bool(void *) > setupNet, void *user) : IsolatedPool(CLONE_NEWNET) { m_NetSetup = setupNet; m_user = user; } #else NetIsolatedPool::NetIsolatedPool(std::function< bool(void *) > setupNet, void *user) : IsolatedPool(0) { m_NetSetup = setupNet; m_user = user; } #endif } // namespace thread } // namespace llarp struct llarp_threadpool { llarp::thread::Pool *impl; llarp::util::Mutex m_access; std::queue< llarp_thread_job * > jobs; llarp_threadpool(int workers, const char *name, bool isolate, setup_net_func setup = nullptr, void *user = nullptr) { if(isolate) impl = new llarp::thread::NetIsolatedPool(setup, user); else impl = new llarp::thread::Pool(); impl->Spawn(workers, name); } llarp_threadpool() : impl(nullptr) { } }; struct llarp_threadpool * llarp_init_threadpool(int workers, const char *name) { if(workers > 0) return new llarp_threadpool(workers, name, false); else return nullptr; } struct llarp_threadpool * llarp_init_same_process_threadpool() { return new llarp_threadpool(); } struct llarp_threadpool * llarp_init_isolated_net_threadpool(const char *name, setup_net_func setup, void *context) { return new llarp_threadpool(1, name, true, setup, context); } void llarp_threadpool_join(struct llarp_threadpool *pool) { llarp::LogDebug("threadpool join"); if(pool->impl) pool->impl->Join(); } void llarp_threadpool_start(struct llarp_threadpool *pool) { /** no op */ } void llarp_threadpool_stop(struct llarp_threadpool *pool) { llarp::LogDebug("threadpool stop"); if(pool->impl) pool->impl->Stop(); } void llarp_threadpool_wait(struct llarp_threadpool *pool) { llarp::util::Mutex mtx; llarp::LogDebug("threadpool wait"); if(pool->impl) { llarp::util::Lock lock(mtx); pool->impl->done.Wait(lock); } } void llarp_threadpool_queue_job(struct llarp_threadpool *pool, struct llarp_thread_job job) { if(pool->impl) pool->impl->QueueJob(job); else if(job.user && job.work) { auto j = new llarp_thread_job; j->work = job.work; j->user = job.user; { llarp::util::Lock lock(pool->m_access); pool->jobs.push(j); } } } void llarp_threadpool_tick(struct llarp_threadpool *pool) { while(pool->jobs.size()) { llarp_thread_job *job; { llarp::util::Lock lock(pool->m_access); job = pool->jobs.front(); pool->jobs.pop(); } job->work(job->user); delete job; } } void llarp_free_threadpool(struct llarp_threadpool **pool) { if(*pool) { delete *pool; } *pool = nullptr; }