// Copyright 2021 Google LLC // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // https://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. #include "internal/platform/implementation/windows/thread_pool.h" #include #include #include #include "absl/memory/memory.h" #include "absl/synchronization/mutex.h" #include "internal/platform/logging.h" #include "internal/platform/runnable.h" namespace nearby { namespace windows { VOID CALLBACK WorkCallback(PTP_CALLBACK_INSTANCE instance, PVOID parameter, PTP_WORK work) { // Instance is not used in thread pool. UNREFERENCED_PARAMETER(instance); ThreadPool* thread_pool = reinterpret_cast(parameter); thread_pool->RunNextTask(); CloseThreadpoolWork(work); } std::unique_ptr ThreadPool::Create(int max_pool_size) { NEARBY_LOGS(VERBOSE) << __func__ << ": Create thread pool with maximum size(" << max_pool_size << ")."; PTP_POOL thread_pool = nullptr; TP_CALLBACK_ENVIRON thread_pool_environ; InitializeThreadpoolEnvironment(&thread_pool_environ); if (max_pool_size <= 0) { NEARBY_LOGS(ERROR) << __func__ << ": Maximum pool size must be positive integer value."; return nullptr; } thread_pool = CreateThreadpool(NULL); if (thread_pool == nullptr) { NEARBY_LOGS(ERROR) << __func__ << ": failed to create thread pool. LastError: " << GetLastError(); return nullptr; } // Sets thread pool maximum value. In order to release all threads, // will keep at least one thread. SetThreadpoolThreadMaximum(thread_pool, max_pool_size); if (!SetThreadpoolThreadMinimum(thread_pool, 1)) { NEARBY_LOGS(ERROR) << __func__ << ": failed to set minimum thread pool size. LastError: " << GetLastError(); CloseThreadpool(thread_pool); return nullptr; } // // Associate the callback environment with our thread pool. // SetThreadpoolCallbackPool(&thread_pool_environ, thread_pool); return absl::WrapUnique( new ThreadPool(thread_pool, thread_pool_environ, max_pool_size)); } ThreadPool::ThreadPool(PTP_POOL thread_pool, TP_CALLBACK_ENVIRON thread_pool_environ, int max_pool_size) : thread_pool_(thread_pool), thread_pool_environ_(thread_pool_environ), max_pool_size_(max_pool_size) { NEARBY_LOGS(VERBOSE) << __func__ << ": Thread pool(" << this << ") is created."; } ThreadPool::~ThreadPool() { NEARBY_LOGS(VERBOSE) << __func__ << ": Thread pool(" << this << ") is released."; ShutDown(); } bool ThreadPool::Run(Runnable task) { if (thread_pool_ == nullptr) { return false; } absl::MutexLock lock(&mutex_); PTP_WORK work; tasks_.push(std::move(task)); NEARBY_LOGS(VERBOSE) << __func__ << ": Scheduled to run task(" << &tasks_.back() << ")."; work = CreateThreadpoolWork(WorkCallback, this, &thread_pool_environ_); if (work == nullptr) { NEARBY_LOGS(ERROR) << __func__ << ": failed to create thread pool work. LastError: " << GetLastError(); return false; } // // Submit the work to the pool. Because this was a pre-allocated // work item (using CreateThreadpoolWork), it is guaranteed to execute. // SubmitThreadpoolWork(work); return true; } void ThreadPool::ShutDown() { NEARBY_LOGS(VERBOSE) << __func__ << ": Shutdown thread pool(" << this << ")."; if (thread_pool_ == nullptr) { NEARBY_LOGS(WARNING) << __func__ << ": Shutdown on closed thread pool(" << this << ")."; return; } CloseThreadpool(thread_pool_); thread_pool_ = nullptr; } void ThreadPool::RunNextTask() { if (thread_pool_ == nullptr) { return; } Runnable task = nullptr; { absl::MutexLock lock(&mutex_); if (!tasks_.empty()) { NEARBY_LOGS(VERBOSE) << __func__ << ": Run task(" << &tasks_.front() << ")."; task = std::move(tasks_.front()); tasks_.pop(); } } if (task == nullptr) { NEARBY_LOGS(WARNING) << __func__ << ": Tried to run task in an empty thread pool."; return; } task(); } } // namespace windows } // namespace nearby