From 01d9ab7df808c476f606da549d11ed9d8d991ede Mon Sep 17 00:00:00 2001 From: Gloop Team Date: Wed, 12 Aug 2026 21:17:33 -0700 Subject: [PATCH] Internal cleanup PiperOrigin-RevId: 963831712 --- gloop/thread/thread_manager.cc | 144 ++++++++++++++-------------- gloop/thread/thread_manager.h | 23 +++-- gloop/thread/thread_manager_test.cc | 43 +++++---- 3 files changed, 104 insertions(+), 106 deletions(-) diff --git a/gloop/thread/thread_manager.cc b/gloop/thread/thread_manager.cc index b16b31b8..bf118c65 100644 --- a/gloop/thread/thread_manager.cc +++ b/gloop/thread/thread_manager.cc @@ -198,7 +198,7 @@ struct TMWork { // An entry in a queue of work }; struct TMThread { - Thread* t; // The Thread; read-only after creation. + std::unique_ptr t; // The Thread; read-only after creation. bool die; // Whether thread should die; under TMPool::pool_mu. bool on_idle_list; // Whether in TMPool::idle_threads; // under TMPool::pool_mu @@ -296,7 +296,7 @@ struct TMPool { // overseer_saw_idle; // pool_mu > tm_mu. TMWorkQueue pool_queue; // ManagedQueue of work; under pool_mu. - absl::flat_hash_set + absl::flat_hash_set> thread_set; // all threads in pool; under pool_mu. std::vector idle_threads; // List of idle_threads; under pool_mu. int64_t last_idle_check_ms; // abs time of last check for idle threads. @@ -337,23 +337,25 @@ struct TMPool { struct ThreadManagerRep final : public ThreadManager::RepBase { ~ThreadManagerRep() override; - ManagedQueue* NewQueue(absl::string_view name, - const ManagedQueueOptions& queue_options) override; + std::unique_ptr NewQueue( + absl::string_view name, + const ManagedQueueOptions& queue_options) override; int n_pools; // number of pools in use in pool[]; power of 2; // read-only after init. int index; // Index of this Rep in tm_vec; under tm_mu. - thread::Options thread_options; // Read-only after init - ThreadManagerPolicy* policy; // thread creation policy; read-only after - // init. - WatchdogCallback watchdog_callback; // Read-only after init - std::atomic rand; // random number generator; not locked. - absl::Mutex rep_mu; // protects refcount - absl::CondVar refcount_cv; // when refcount drops to 1 - int refcount; // number of child Queues; under rep_mu - uint32_t next_q_id; // queue_id for next q_rep; under rep_mu - thread::CpuSubContainer* subcontainer; // Thread scheduling container if not - // nullptr. + thread::Options thread_options; // Read-only after init + std::unique_ptr policy; // thread creation policy; + // read-only after init. + WatchdogCallback watchdog_callback; // Read-only after init + std::atomic rand; // random number generator; + // not locked. + absl::Mutex rep_mu; // protects refcount + absl::CondVar refcount_cv; // when refcount drops to 1 + int refcount; // number of child Queues; under rep_mu + uint32_t next_q_id; // queue_id for next q_rep; under rep_mu + std::unique_ptr + subcontainer; // Thread scheduling container if not nullptr. TMPool pool[kTMMaxPools]; // Individually locked with TMPool::pool_mu. }; @@ -401,12 +403,13 @@ static void TMOverseer(); // Overseer thread. // TM instances; under tm_mu. static std::vector& TMVec() { static absl::NoDestructor> tm_vec; + static absl::NoDestructor> overseer_thread; [[maybe_unused]] static bool need_init = [] { thread::Options overseer_options; overseer_options.set_stack_size(128 * 1024); - Thread* t = new ClosureThread(overseer_options, "thread_manager_overseer", - &TMOverseer); - t->Start(); + *overseer_thread = std::make_unique( + overseer_options, "thread_manager_overseer", &TMOverseer); + (*overseer_thread)->Start(); return false; }(); return *tm_vec; @@ -417,7 +420,8 @@ static int (*tm_num_cpus)() = &::base::AvailableCPUs; // Protects qu_set; qu_mu < queue_mu ABSL_CONST_INIT static absl::Mutex qu_mu(absl::kConstInit); -static absl::flat_hash_set* qu_set ABSL_GUARDED_BY(qu_mu); +static absl::NoDestructor> qu_set + ABSL_GUARDED_BY(qu_mu); // Forward declarations. static void TMQueueRepDelete(ThreadManagerRep* rep, ManagedQueueRep* q_rep); @@ -603,10 +607,10 @@ static void TMWorker(ThreadManagerRep* rep, TMPool* pool, TMThread* self) { // TMThread along with its TMPool; used by TMTakeExitingThreads, etc. struct TMThreadWithPool { - TMThreadWithPool(TMThread* thread, TMPool* pool) - : thread(thread), pool(pool) {} + TMThreadWithPool(std::unique_ptr thread, TMPool* pool) + : thread(std::move(thread)), pool(pool) {} - TMThread* thread; + std::unique_ptr thread; TMPool* pool; }; @@ -617,9 +621,11 @@ static void TMTakeExitingThreads( TMPool* pool, std::vector* exiting_threads) { while (!pool->exiting_threads.empty()) { TMThread* thread = pool->exiting_threads.back(); - exiting_threads->push_back(TMThreadWithPool(thread, pool)); - pool->thread_set.erase(thread); pool->exiting_threads.pop_back(); + auto it = pool->thread_set.find(thread); + DCHECK(it != pool->thread_set.end()); + auto node = pool->thread_set.extract(it); + exiting_threads->push_back(TMThreadWithPool(std::move(node.value()), pool)); } } @@ -635,7 +641,7 @@ static void TMDestroyExitingThreads( std::vector* exiting_threads, WatchDog* watchdog) { int num_in_pool = 0; for (size_t i = 0; i != exiting_threads->size(); i++) { - TMThread* thread = (*exiting_threads)[i].thread; + TMThread* thread = (*exiting_threads)[i].thread.get(); TMPool* pool = (*exiting_threads)[i].pool; thread->t->Join(); if (watchdog) { @@ -643,8 +649,6 @@ static void TMDestroyExitingThreads( // our watchdog. watchdog->Alive(); } - delete thread->t; - delete thread; num_in_pool++; // Update pool state when we're done with this pool. if (i + 1 == exiting_threads->size() || @@ -663,11 +667,13 @@ static void TMDestroyExitingThreads( // L >= pool->pool_mu static void TMCreateWorker(ThreadManagerRep* rep, TMPool* pool, ThreadStarter* new_threads) { - TMThread* t = new TMThread; - t->t = new ClosureThread(rep->thread_options, pool->name_prefix, - absl::bind_front(TMWorker, rep, pool, t)); + auto t = std::make_unique(); + TMThread* t_ptr = t.get(); + t->t = std::make_unique( + rep->thread_options, pool->name_prefix, + absl::bind_front(TMWorker, rep, pool, t_ptr)); if (rep->subcontainer != nullptr) { - t->t->SetInitialCpuSubContainer(rep->subcontainer); + t->t->SetInitialCpuSubContainer(rep->subcontainer.get()); } t->die = false; t->on_idle_list = false; @@ -677,11 +683,11 @@ static void TMCreateWorker(ThreadManagerRep* rep, TMPool* pool, t->work.q_id = 0; t->work.counted = false; t->t->SetJoinable(true); - new_threads->Add(t->t); + new_threads->Add(t->t.get()); pool->created++; pool->create_pending++; - pool->thread_set.insert(t); - VLOG(3) << "TMCreateWorker exit. t=" << t; + pool->thread_set.insert(std::move(t)); + VLOG(3) << "TMCreateWorker exit. t=" << t_ptr; } // Kill a single indexed thread @@ -998,9 +1004,9 @@ static void TMEnsureRegisteredWithOverseer(ThreadManagerRep* rep) { // Called by the constructor // L < tm_mu -static ThreadManagerRep* TMRepNew(absl::string_view name_prefix, - const ManagerOptions& options) { - ThreadManagerRep* rep = new ThreadManagerRep; +static std::unique_ptr TMRepNew(absl::string_view name_prefix, + ManagerOptions options) { + auto rep = std::make_unique(); rep->n_pools = 1; if (options.n_pools >= 1) { // round up n_pools to power of two no greater than kTMMaxPools @@ -1010,22 +1016,21 @@ static ThreadManagerRep* TMRepNew(absl::string_view name_prefix, } } rep->index = -1; // -1 means not known to overseer; no queue created yet - rep->rand.store(reinterpret_cast(rep), + rep->rand.store(reinterpret_cast(rep.get()), std::memory_order_relaxed); // init arbitrarily rep->thread_options = options.thread_options; rep->watchdog_callback = options.get_watchdog_callback(); - ThreadManagerPolicy* policy = options.policy; + std::unique_ptr policy = std::move(options.policy); if (absl::GetFlag(FLAGS_threadmanager_ignore_policy)) { - delete policy; policy = nullptr; } if (policy != nullptr) { - rep->policy = policy; + rep->policy = std::move(policy); } else { tm_mu.lock(); int (*num_cpus)() = tm_num_cpus; tm_mu.unlock(); - rep->policy = DefaultThreadManagerPolicy(num_cpus); + rep->policy.reset(DefaultThreadManagerPolicy(num_cpus)); } rep->refcount = 0; rep->next_q_id = 0; @@ -1036,8 +1041,8 @@ static ThreadManagerRep* TMRepNew(absl::string_view name_prefix, << "--use_thread_subcontainers flag is not set"; } else { - rep->subcontainer = thread::CpuSubContainer::Create( - options.thread_options, std::string(name_prefix)); + rep->subcontainer.reset(thread::CpuSubContainer::Create( + options.thread_options, std::string(name_prefix))); } } for (int pool_i = 0; pool_i != rep->n_pools; pool_i++) { @@ -1119,9 +1124,6 @@ static void TMRepDelete(ThreadManagerRep* rep) { } pool->pool_mu.unlock(); } - - delete rep->policy; - delete rep->subcontainer; } // Return a pointer to a randomly-chosen pool within the thread manager. @@ -1239,6 +1241,7 @@ static bool TMQueueAdd(ThreadManagerRep* rep, absl::AnyInvocable cb, // L < rep->rep_mu, tm_mu, pool_mu static void TMQueueRepDelete(ThreadManagerRep* rep, ManagedQueueRep* q_rep) { VLOG(3) << "TMQueueRepDelete entry."; + std::unique_ptr q_rep_deleter(q_rep); rep->rep_mu.lock(); rep->refcount--; CHECK_GE(rep->refcount, 0); @@ -1246,7 +1249,6 @@ static void TMQueueRepDelete(ThreadManagerRep* rep, ManagedQueueRep* q_rep) { rep->refcount_cv.SignalAll(); } rep->rep_mu.unlock(); - delete q_rep; VLOG(3) << "TMQueueRepDelete exit."; } @@ -1267,9 +1269,9 @@ static void TMQueueRepUnref(ManagedQueueRep* q_rep) { // ----------------------------------------------------------------------- // public interface -ThreadManager::ThreadManager(absl::string_view name_prefix, - const ManagerOptions& options) - : rep_(TMRepNew(name_prefix, options)) {} +ThreadManager::ThreadManager(absl::string_view thread_name_prefix, + ManagerOptions options) + : rep_(TMRepNew(thread_name_prefix, std::move(options))) {} // L < tm_mu, this->rep_->pool[*].pool_mu, this->rep->rep_mu ThreadManagerRep::~ThreadManagerRep() { @@ -1284,12 +1286,11 @@ ThreadManagerRep::~ThreadManagerRep() { } // L < this->rep_->rep_mu, tm_mu -ManagedQueue* ThreadManagerRep::NewQueue( +std::unique_ptr ThreadManagerRep::NewQueue( absl::string_view name, const ManagedQueueOptions& queue_options) { TMEnsureRegisteredWithOverseer(this); // need an overseer to hand out a queue rep_mu.lock(); - ManagedQueueRep* q_rep; - q_rep = new ManagedQueueRep; + auto q_rep = std::make_unique(); q_rep->queue_name = std::string(name); q_rep->queue_options = queue_options; q_rep->parent_rep = this; @@ -1301,17 +1302,15 @@ ManagedQueue* ThreadManagerRep::NewQueue( refcount++; rep_mu.unlock(); // queue_options is read-only after this point - return new ManagedQueueImpl(q_rep); + return std::make_unique(q_rep.release()); } std::vector ThreadManager::QueueStats() ABSL_LOCKS_EXCLUDED(qu_mu) { std::vector stats; qu_mu.lock(); - if (qu_set != nullptr) { - for (ManagedQueue* q : *qu_set) { - stats.push_back(q->Stats()); - } + for (ManagedQueue* q : *qu_set) { + stats.push_back(q->Stats()); } qu_mu.unlock(); return stats; @@ -1376,33 +1375,35 @@ class ExecutorManagedQueue final : public ManagedQueue { // The default ThreadManager and ManagedQueue are initilized under // tm_default_once static absl::once_flag tm_default_once; -static ThreadManager* tm_default_thread_manager; // the default ThreadManager -static ManagedQueue* tm_default_queue; // the default ManagedQueue +static absl::NoDestructor> + tm_default_thread_manager; // the default ThreadManager +static absl::NoDestructor> + tm_default_queue; // the default ManagedQueue -// Create the default thread manager and queue. Called using GoogleOnceInit(). +// Create the default thread manager and queue. Called using absl::call_once. static void TMMakeDefault() { ManagerOptions manager_options; manager_options.n_pools = absl::GetFlag(FLAGS_threadmanager_default_manager_pools); - tm_default_thread_manager = - new ThreadManager("default_ThreadManager", manager_options); + *tm_default_thread_manager = std::make_unique( + "default_ThreadManager", std::move(manager_options)); if (absl::GetFlag(FLAGS_threadmanager_default_queue_executor)) { - tm_default_queue = - new ExecutorManagedQueue(*thread::Executor::DefaultExecutor()); + *tm_default_queue = std::make_unique( + *thread::Executor::DefaultExecutor()); } else { - tm_default_queue = tm_default_thread_manager->NewQueue( - "default_queue", ManagedQueueOptions()); + *tm_default_queue = (*tm_default_thread_manager) + ->NewQueue("default_queue", ManagedQueueOptions()); } } ThreadManager* DefaultManager() { absl::call_once(tm_default_once, &TMMakeDefault); - return tm_default_thread_manager; + return tm_default_thread_manager->get(); } ManagedQueue* DefaultQueue() { absl::call_once(tm_default_once, &TMMakeDefault); - return tm_default_queue; + return tm_default_queue->get(); } // Set the default version of NumCPUs() @@ -1468,9 +1469,6 @@ static void TMCountAllWorkFromQueue(ManagedQueueRep* q_rep, ManagedQueueImpl::ManagedQueueImpl(ManagedQueueRep* q_rep) : q_rep_(q_rep) { if (this != &this->q_rep_->queue_external) { qu_mu.lock(); - if (qu_set == nullptr) { - qu_set = new absl::flat_hash_set; - } qu_set->insert(this); qu_mu.unlock(); } diff --git a/gloop/thread/thread_manager.h b/gloop/thread/thread_manager.h index b41e60a1..791baf05 100644 --- a/gloop/thread/thread_manager.h +++ b/gloop/thread/thread_manager.h @@ -94,6 +94,7 @@ #include "absl/strings/string_view.h" #include "absl/time/time.h" #include "gloop/thread/executor.h" +#include "gloop/thread/thread_manager_policy.h" #include "gloop/thread/thread_options.h" #include "gloop/thread/watchdog.h" #include "gtest/gtest_prod.h" @@ -133,11 +134,11 @@ struct ManagerOptions { int n_pools; // Expert clients may set "policy" to control thread-creation policy; see // thread_manager_policy.h. Most users should use the default: 0. - // The ThreadManager destructor will "delete policy". + // The ThreadManager takes ownership of this policy. ABSL_DEPRECATED( "ThreadManagerPolicy is almost never set in google3. " "Remaining callers are being removed to remove this option.") - ThreadManagerPolicy* policy; + std::unique_ptr policy; WatchdogCallback get_watchdog_callback() const { return watchdog_callback; } @@ -171,8 +172,7 @@ struct ManagedQueueStats { class ThreadManager { public: - ThreadManager(absl::string_view thread_name_prefix, - const ManagerOptions& options); + ThreadManager(absl::string_view thread_name_prefix, ManagerOptions options); // This type is neither copyable nor movable. ThreadManager(const ThreadManager&) = delete; @@ -182,14 +182,13 @@ class ThreadManager { // Queues have been deleted and the work // associated with them has completed. - // Return a pointer to a named work queue serviced by this ThreadManager with + // Return a named work queue serviced by this ThreadManager with // limits given by queue_options. Repeated calls to NewQueue() with the same - // name will provide distinct queues. Queues should be discarded with - // "delete" when no longer needed; delete will return immediately, but the - // underlying data structures will be discarded when all pending work is - // complete. - ManagedQueue* NewQueue(absl::string_view name, - const ManagedQueueOptions& queue_options) { + // name will provide distinct queues. The returned unique_ptr handles the + // lifetime; when it is destroyed, the underlying data structures will be + // discarded when all pending work is complete. + std::unique_ptr NewQueue( + absl::string_view name, const ManagedQueueOptions& queue_options) { return rep_->NewQueue(name, queue_options); } @@ -207,7 +206,7 @@ class ThreadManager { struct RepBase { virtual ~RepBase() = default; - virtual ManagedQueue* NewQueue( + virtual std::unique_ptr NewQueue( absl::string_view name, const ManagedQueueOptions& queue_options) = 0; }; diff --git a/gloop/thread/thread_manager_test.cc b/gloop/thread/thread_manager_test.cc index 59e0e622..be0e91dc 100644 --- a/gloop/thread/thread_manager_test.cc +++ b/gloop/thread/thread_manager_test.cc @@ -205,8 +205,9 @@ TEST(ThreadManagerTest, WaitUntilComplete) { TEST(ThreadManagerTest, SchedulingDuringManagedQueueDestruction) { thread::ManagerOptions mgr_options; - mgr_options.policy = thread::EagerThreadManagerPolicy(/*max_threads=*/1); - auto tm = std::make_unique("a", mgr_options); + mgr_options.policy.reset(thread::EagerThreadManagerPolicy(/*max_threads=*/1)); + auto tm = + std::make_unique("a", std::move(mgr_options)); std::unique_ptr q( tm->NewQueue("b", thread::ManagedQueueOptions())); @@ -315,8 +316,7 @@ TEST(ThreadManagerTest, SleepingClosures) { queue_info.cb.reset(::util::functional::ToPermanentCallback( absl::bind_front(&ParameterizedTestClosure, &key, 1000, 0, &queue_info))); queue_info.name = "sleeping"; - queue_info.queue.reset( - tm->NewQueue(queue_info.name, queue_info.queue_options)); + queue_info.queue = tm->NewQueue(queue_info.name, queue_info.queue_options); const absl::Time start_time = absl::Now(); for (int i = 0; i != queue_info.expected_calls; i++) { queue_info.queue->Schedule( @@ -351,8 +351,7 @@ TEST(ThreadManagerTest, CPUBoundClosures) { queue_info.cb.reset(::util::functional::ToPermanentCallback(absl::bind_front( &ParameterizedTestClosure, &key, 0, kSpin, &queue_info))); queue_info.name = "cpu_bound"; - queue_info.queue.reset( - tm->NewQueue(queue_info.name, queue_info.queue_options)); + queue_info.queue = tm->NewQueue(queue_info.name, queue_info.queue_options); for (int i = 0; i != queue_info.expected_calls; i++) { queue_info.queue->Schedule( util::functional::FromCallback(queue_info.cb.get())); @@ -383,8 +382,7 @@ TEST(ThreadManagerTest, ContendingClosures) { queue_info.cb.reset(::util::functional::ToPermanentCallback( absl::bind_front(&ParameterizedTestClosure, &key, 0, 0, &queue_info))); queue_info.name = "contending"; - queue_info.queue.reset( - tm->NewQueue(queue_info.name, queue_info.queue_options)); + queue_info.queue = tm->NewQueue(queue_info.name, queue_info.queue_options); for (int i = 0; i != queue_info.expected_calls; i++) { queue_info.queue->Schedule( util::functional::FromCallback(queue_info.cb.get())); @@ -425,8 +423,8 @@ TEST(ThreadManagerTest, Queues) { ::util::functional::ToPermanentCallback(absl::bind_front( &ParameterizedTestClosure, &key, kSleepMS, 0, &queue_info[i]))); queue_info[i].name = absl::StrFormat("queue %d", i); - queue_info[i].queue.reset( - tm->NewQueue(queue_info[i].name, queue_info[i].queue_options)); + queue_info[i].queue = + tm->NewQueue(queue_info[i].name, queue_info[i].queue_options); CHECK_EQ(queue_info[i].queue->num_pending_closures(), 0); } @@ -718,8 +716,9 @@ TEST(ThreadManagerWatchdogTest, UsesCustomWatchDogCallback) { thread::ManagedQueueOptions queue_options; queue_options.time_limit_s = 1; - std::optional tm( - std::in_place, "custom_watchdog_callback_test", manager_options); + std::optional tm(std::in_place, + "custom_watchdog_callback_test", + std::move(manager_options)); std::unique_ptr q( tm->NewQueue("custom_watchdog_callback_test_queue", queue_options)); @@ -832,9 +831,9 @@ static void BM_ThreadManagerRun(benchmark::State& state) { // Make a single-thread manager thread::ManagerOptions options; options.n_pools = 1; - options.policy = new SingleThreadPolicy; - std::optional manager(std::in_place, - "benchmark_manager", options); + options.policy = std::make_unique(); + std::optional manager( + std::in_place, "benchmark_manager", std::move(options)); // Make a queue with the specified thread limit thread::ManagedQueueOptions qoptions; @@ -852,9 +851,9 @@ static void BM_ThreadManagerSchedule(benchmark::State& state) { // Make a single-thread manager thread::ManagerOptions options; options.n_pools = 1; - options.policy = new SingleThreadPolicy; - std::optional manager(std::in_place, - "benchmark_manager", options); + options.policy = std::make_unique(); + std::optional manager( + std::in_place, "benchmark_manager", std::move(options)); // Make a queue with the specified thread limit thread::ManagedQueueOptions qoptions; @@ -881,7 +880,8 @@ BENCHMARK(BM_ThreadManagerSchedule)->Arg(1)->Arg(INT_MAX); static void BM_ThreadManagerDefaultPolicyRun(benchmark::State& state) { // Make a thread manager with the default policy thread::ManagerOptions options; - thread::ThreadManager manager("benchmark_manager_with_defaults", options); + thread::ThreadManager manager("benchmark_manager_with_defaults", + std::move(options)); // Make a queue with the specified thread limit thread::ManagedQueueOptions qoptions; @@ -903,8 +903,9 @@ static void BM_ThreadManagerDefaultPolicyQueuedInAdvance( // Make a thread manager with the default policy thread::ManagerOptions options; options.n_pools = 1; - options.policy = new SingleThreadPolicy; - thread::ThreadManager manager("benchmark_manager_with_defaults", options); + options.policy = std::make_unique(); + thread::ThreadManager manager("benchmark_manager_with_defaults", + std::move(options)); // Make a queue with the specified thread limit thread::ManagedQueueOptions qoptions; std::unique_ptr q(