zircon/system/ulib/dispatcher-pool/dispatcher-thread-pool.cpp - fuchsia - Git at Google

 // Copyright 2017 The Fuchsia Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.

 #include <zircon/syscalls.h>
 #include <zircon/syscalls/port.h>
 #include <stdio.h>
 #include <string.h>

 #include <dispatcher-pool/dispatcher-execution-domain.h>
 #include <dispatcher-pool/dispatcher-thread-pool.h>

 #include <utility>

 #include "debug-logging.h"

 namespace dispatcher {

 fbl::Mutex ThreadPool::active_pools_lock_;
 fbl::WAVLTree<zx_koid_t, fbl::RefPtr<ThreadPool>> ThreadPool::active_pools_;
 bool ThreadPool::system_shutdown_ = false;

 namespace {

 zx_koid_t GetKoid(const zx::profile& profile) {
     // A null profile is valid and results in the default priority thread pool,
     // which is keyed by the 0-koid.
     if (!profile.is_valid())
         return ZX_KOID_INVALID;

     zx_info_handle_basic_t profile_info = {};
     if (profile.get_info(ZX_INFO_HANDLE_BASIC, &profile_info, sizeof(profile_info), nullptr,
                          nullptr) != ZX_OK) {
         return ZX_KOID_INVALID;
     }
     return profile_info.koid;
 }

 } // namespace

 // static
 zx_status_t ThreadPool::Get(fbl::RefPtr<ThreadPool>* pool_out, zx::profile profile) {
     if (pool_out == nullptr)
         return ZX_ERR_INVALID_ARGS;

     // From here on out, we need to be inside of the active pools lock.
     fbl::AutoLock lock(&active_pools_lock_);

     // We cannot return any pool references if we are in the process of
     // system-wide shutdown.
     if (system_shutdown_)
         return ZX_ERR_BAD_STATE;

     // Do we already have a pool running at the desired priority?  If so, just
     // return a reference to it.
     auto iter = active_pools_.find(GetKoid(profile));
     if (iter.IsValid()) {
         *pool_out = iter.CopyPointer();
         return ZX_OK;
     }

     // Looks like we don't have an appropriate pool just yet.  Try to create one
     // and add it to the active set of pools.
     fbl::AllocChecker ac;
     auto new_pool = fbl::AdoptRef(new (&ac) ThreadPool(std::move(profile)));
     if (!ac.check()) {
         printf("Failed to allocate new thread pool\n");
         return ZX_ERR_NO_MEMORY;
     }

     zx_status_t res = new_pool->Init();
     if (res != ZX_OK) {
         printf("Failed to initialize new thread pool (res %d)\n", res);
         return res;
     }

     *pool_out = new_pool;
     active_pools_.insert(std::move(new_pool));
     return ZX_OK;
 }

 // static
 void ThreadPool::ShutdownAll() {
     fbl::WAVLTree<zx_koid_t, fbl::RefPtr<ThreadPool>> shutdown_targets;

     {
         fbl::AutoLock lock(&active_pools_lock_);
         if (system_shutdown_) {
             ZX_DEBUG_ASSERT(active_pools_.is_empty());
             return;
         }

         system_shutdown_ = true;
         shutdown_targets = std::move(active_pools_);
     }

     for (auto& pool : shutdown_targets)
         pool.InternalShutdown();
 }

 void ThreadPool::Shutdown() {
     // If we have already been removed from the set of active thread pools, then
     // someone is already shutting us down and there is nothing to do.
     {
         fbl::AutoLock lock(&active_pools_lock_);
         if (!this->InContainer())
             return;

         active_pools_.erase(*this);
     }

     InternalShutdown();
 }

 zx_status_t ThreadPool::AddDomainToPool(fbl::RefPtr<ExecutionDomain> domain) {
     ZX_DEBUG_ASSERT(domain != nullptr);
     fbl::AutoLock pool_lock(&pool_lock_);

     if (pool_shutting_down_)
         return ZX_ERR_BAD_STATE;

     active_domains_.push_back(std::move(domain));
     ++active_domain_count_;

     while ((active_thread_count_ < active_domain_count_) &&
            (active_thread_count_ < zx_system_get_num_cpus())) {
         auto thread = Thread::Create(fbl::WrapRefPtr(this), active_thread_count_);
         if (thread == nullptr) {
             LOG("Failed to create new thread\n");
             break;
         }

         active_threads_.push_front(std::move(thread));
         if (active_threads_.front().Start() != ZX_OK) {
             LOG("Failed to start new thread\n");
             thread = active_threads_.pop_front();
             break;
         }

         active_thread_count_++;
     }

     return ZX_OK;
 }

 void ThreadPool::RemoveDomainFromPool(ExecutionDomain* domain) {
     ZX_DEBUG_ASSERT(domain != nullptr);
     fbl::AutoLock pool_lock(&pool_lock_);
     active_domains_.erase(*domain);
 }

 zx_status_t ThreadPool::WaitOnPort(const zx::handle& handle,
                                    uint64_t key,
                                    zx_signals_t signals,
                                    uint32_t options) {
     ZX_DEBUG_ASSERT(handle.is_valid());
     fbl::AutoLock pool_lock(&pool_lock_);

     if (!port_.is_valid()) {
         DEBUG_LOG("WaitOnPort failed, port handle is invalid\n");
         return ZX_ERR_BAD_STATE;
     }

     return handle.wait_async(port_, key, signals, options);
 }

 zx_status_t ThreadPool::CancelWaitOnPort(const zx::handle& handle, uint64_t key) {
     ZX_DEBUG_ASSERT(handle.is_valid());
     fbl::AutoLock pool_lock(&pool_lock_);

     if (!port_.is_valid()) {
         DEBUG_LOG("CancelWaitOnPort failed, port handle is invalid\n");
         return ZX_ERR_BAD_STATE;
     }

     return port_.cancel(handle, key);
 }

 zx_status_t ThreadPool::BindIrqToPort(const zx::handle& irq_handle, uint64_t key) {
     ZX_DEBUG_ASSERT(irq_handle.is_valid());
     fbl::AutoLock pool_lock(&pool_lock_);

     if (!port_.is_valid()) {
         DEBUG_LOG("BindIrqToPort failed, port handle is invalid\n");
         return ZX_ERR_BAD_STATE;
     }

     return zx_interrupt_bind(irq_handle.get(), port_.get(), key, 0u);
 }

 zx_koid_t ThreadPool::GetKey() const {
     return profile_koid_;
 }

 ThreadPool::ThreadPool(zx::profile profile) : profile_(std::move(profile)) {
     profile_koid_ = GetKoid(profile_);
 }

 void ThreadPool::PrintDebugPrefix() {
     printf("[ThreadPool %zu] ", GetKey());
 }

 zx_status_t ThreadPool::Init() {
     ZX_DEBUG_ASSERT(!port_.is_valid());

     zx_status_t res = zx::port::create(ZX_PORT_BIND_TO_INTERRUPT, &port_);
     if (res != ZX_OK) {
         LOG("Failed to create thread pool port (res %d)!\n", res);
         return res;
     }

     return ZX_OK;
 }

 void ThreadPool::InternalShutdown() {
     // Be careful when shutting down, a specific sequence needs to be followed.
     // See MG-1118 for details.
     decltype(active_domains_) domains_to_deactivate;
     {
         fbl::AutoLock lock(&pool_lock_);
         // If someone is already shutting us down, then we are done.
         if (pool_shutting_down_)
             return;

         // Prevent any new clients from joining the pool.
         pool_shutting_down_ = true;

         // Move the contents of the active domains list into a local variable so
         // that we don't need to hold onto the pool lock while we shut down the
         // domains.
         domains_to_deactivate = std::move(active_domains_);
     }

     // Deactivate any domains we may have still had assigned to us, then let go
     // of our references to them.  Deactivation of domains should synchronize
     // will all pending operations in the domain (meaning that all references
     // have been recovered and no new wait operations will be posted)
     for (auto& domain : domains_to_deactivate)
         domain.Deactivate();

     domains_to_deactivate.clear();

     // Manually queue a quit message for each thread in the thread pool.
     {
         zx_port_packet pkt;
         memset(&pkt, 0, sizeof(pkt));
         pkt.type = ZX_PKT_TYPE_USER;

         fbl::AutoLock lock(&pool_lock_);
         for (__UNUSED const auto& thread : active_threads_) {
             __UNUSED zx_status_t res;
             res = port_.queue(&pkt);
             ZX_DEBUG_ASSERT(res == ZX_OK);
         }
     }

     // Synchronize with the threads as they exit.
     while (true) {
         fbl::unique_ptr<Thread> thread;
         {
             fbl::AutoLock lock(&pool_lock_);
             if (active_threads_.is_empty())
                 break;

             thread = active_threads_.pop_front();
         }

         thread->Join();
     }
 }

 // static
 fbl::unique_ptr<ThreadPool::Thread> ThreadPool::Thread::Create(fbl::RefPtr<ThreadPool> pool,
                                                                 uint32_t id) {
     fbl::AllocChecker ac;
     fbl::unique_ptr<Thread> ret(new (&ac) Thread(std::move(pool), id));

     if (!ac.check())
         return nullptr;

     return ret;
 }

 ThreadPool::Thread::Thread(fbl::RefPtr<ThreadPool> pool, uint32_t id)
     : pool_(std::move(pool)),
       id_(id) {
 }

 zx_status_t ThreadPool::Thread::Start() {
     int c11_res = thrd_create(
             &thread_handle_,
             [](void* ctx) -> int { return static_cast<Thread*>(ctx)->Main(); },
             this);

     if (c11_res != thrd_success) {
         LOG("Failed to create new client thread (res %d)!\n", c11_res);
         // TODO(johngro) : translate C11 error
         return ZX_ERR_INTERNAL;
     }

     return ZX_OK;
 }

 void ThreadPool::Thread::Join() {
     // TODO(johngro) : Switch to native zircon threads so we can supply a
     // timeout to the join event.
     int musl_ret;
     thrd_join(thread_handle_, &musl_ret);
     ZX_DEBUG_ASSERT(pool_ == nullptr);
 }

 void ThreadPool::Thread::PrintDebugPrefix() const {
     printf("[Thread %03u-%zu] ", id_, pool_->GetKey());
 }

 int ThreadPool::Thread::Main() {
     zx_status_t res;

     ZX_DEBUG_ASSERT(pool_ != nullptr);
     DEBUG_LOG("Thread Starting\n");

     if (pool_->profile().is_valid()) {
         res = zx_object_set_profile(zx_thread_self(), pool_->profile().get(), 0);
         if (res != ZX_OK) {
             DEBUG_LOG("WARNING - Failed to set thread profile (res %d)\n", res);
         }
     }

     while (true) {
         zx_port_packet_t pkt;

         // TODO(johngro) : consider automatically shutting down if we have more
         // threads than clients.

         // Wait for there to be work to dispatch.  We should never encounter an
         // error, but if we do, shut down.
         res = pool_->port().wait(zx::time::infinite(), &pkt);
         ZX_DEBUG_ASSERT(res == ZX_OK);

         // Is it time to exit?
         if ((res != ZX_OK) || (pkt.type == ZX_PKT_TYPE_USER)) {
             break;
         }

         if ((pkt.type != ZX_PKT_TYPE_SIGNAL_ONE) &&
             (pkt.type != ZX_PKT_TYPE_INTERRUPT)) {
             LOG("Unexpected packet type (%u) in Thread pool!\n", pkt.type);
             continue;
         }

         // Reclaim our event source reference from the kernel.
         static_assert(sizeof(pkt.key) >= sizeof(EventSource*),
                       "Port packet keys are not large enough to hold a pointer!");
         auto event_source =
             fbl::internal::MakeRefPtrNoAdopt(reinterpret_cast<EventSource*>(pkt.key));

         // Schedule the dispatch of the pending events for this event source.
         // If ScheduleDispatch returns a valid ExecutionDomain reference, then
         // actually go ahead and perform the dispatch of pending work for this
         // domain.
         ZX_DEBUG_ASSERT(event_source != nullptr);
         fbl::RefPtr<ExecutionDomain> domain = event_source->ScheduleDispatch(pkt);

         if (domain != nullptr)
             domain->DispatchPendingWork();
     }

     DEBUG_LOG("Client work thread shutting down\n");
     pool_.reset();

     return 0;
 }

 }  // namespace dispatcher
	// Copyright 2017 The Fuchsia Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style license that can be
	// found in the LICENSE file.

	#include <zircon/syscalls.h>
	#include <zircon/syscalls/port.h>
	#include <stdio.h>
	#include <string.h>

	#include <dispatcher-pool/dispatcher-execution-domain.h>
	#include <dispatcher-pool/dispatcher-thread-pool.h>

	#include <utility>

	#include "debug-logging.h"

	namespace dispatcher {

	fbl::Mutex ThreadPool::active_pools_lock_;
	fbl::WAVLTree<zx_koid_t, fbl::RefPtr<ThreadPool>> ThreadPool::active_pools_;
	bool ThreadPool::system_shutdown_ = false;

	namespace {

	zx_koid_t GetKoid(const zx::profile& profile) {
	// A null profile is valid and results in the default priority thread pool,
	// which is keyed by the 0-koid.
	if (!profile.is_valid())
	return ZX_KOID_INVALID;

	zx_info_handle_basic_t profile_info = {};
	if (profile.get_info(ZX_INFO_HANDLE_BASIC, &profile_info, sizeof(profile_info), nullptr,
	nullptr) != ZX_OK) {
	return ZX_KOID_INVALID;
	}
	return profile_info.koid;
	}

	} // namespace

	// static
	zx_status_t ThreadPool::Get(fbl::RefPtr<ThreadPool>* pool_out, zx::profile profile) {
	if (pool_out == nullptr)
	return ZX_ERR_INVALID_ARGS;

	// From here on out, we need to be inside of the active pools lock.
	fbl::AutoLock lock(&active_pools_lock_);

	// We cannot return any pool references if we are in the process of
	// system-wide shutdown.
	if (system_shutdown_)
	return ZX_ERR_BAD_STATE;

	// Do we already have a pool running at the desired priority? If so, just
	// return a reference to it.
	auto iter = active_pools_.find(GetKoid(profile));
	if (iter.IsValid()) {
	*pool_out = iter.CopyPointer();
	return ZX_OK;
	}

	// Looks like we don't have an appropriate pool just yet. Try to create one
	// and add it to the active set of pools.
	fbl::AllocChecker ac;
	auto new_pool = fbl::AdoptRef(new (&ac) ThreadPool(std::move(profile)));
	if (!ac.check()) {
	printf("Failed to allocate new thread pool\n");
	return ZX_ERR_NO_MEMORY;
	}

	zx_status_t res = new_pool->Init();
	if (res != ZX_OK) {
	printf("Failed to initialize new thread pool (res %d)\n", res);
	return res;
	}

	*pool_out = new_pool;
	active_pools_.insert(std::move(new_pool));
	return ZX_OK;
	}

	// static
	void ThreadPool::ShutdownAll() {
	fbl::WAVLTree<zx_koid_t, fbl::RefPtr<ThreadPool>> shutdown_targets;

	{
	fbl::AutoLock lock(&active_pools_lock_);
	if (system_shutdown_) {
	ZX_DEBUG_ASSERT(active_pools_.is_empty());
	return;
	}

	system_shutdown_ = true;
	shutdown_targets = std::move(active_pools_);
	}

	for (auto& pool : shutdown_targets)
	pool.InternalShutdown();
	}

	void ThreadPool::Shutdown() {
	// If we have already been removed from the set of active thread pools, then
	// someone is already shutting us down and there is nothing to do.
	{
	fbl::AutoLock lock(&active_pools_lock_);
	if (!this->InContainer())
	return;

	active_pools_.erase(*this);
	}

	InternalShutdown();
	}

	zx_status_t ThreadPool::AddDomainToPool(fbl::RefPtr<ExecutionDomain> domain) {
	ZX_DEBUG_ASSERT(domain != nullptr);
	fbl::AutoLock pool_lock(&pool_lock_);

	if (pool_shutting_down_)
	return ZX_ERR_BAD_STATE;

	active_domains_.push_back(std::move(domain));
	++active_domain_count_;

	while ((active_thread_count_ < active_domain_count_) &&
	(active_thread_count_ < zx_system_get_num_cpus())) {
	auto thread = Thread::Create(fbl::WrapRefPtr(this), active_thread_count_);
	if (thread == nullptr) {
	LOG("Failed to create new thread\n");
	break;
	}

	active_threads_.push_front(std::move(thread));
	if (active_threads_.front().Start() != ZX_OK) {
	LOG("Failed to start new thread\n");
	thread = active_threads_.pop_front();
	break;
	}

	active_thread_count_++;
	}

	return ZX_OK;
	}

	void ThreadPool::RemoveDomainFromPool(ExecutionDomain* domain) {
	ZX_DEBUG_ASSERT(domain != nullptr);
	fbl::AutoLock pool_lock(&pool_lock_);
	active_domains_.erase(*domain);
	}

	zx_status_t ThreadPool::WaitOnPort(const zx::handle& handle,
	uint64_t key,
	zx_signals_t signals,
	uint32_t options) {
	ZX_DEBUG_ASSERT(handle.is_valid());
	fbl::AutoLock pool_lock(&pool_lock_);

	if (!port_.is_valid()) {
	DEBUG_LOG("WaitOnPort failed, port handle is invalid\n");
	return ZX_ERR_BAD_STATE;
	}

	return handle.wait_async(port_, key, signals, options);
	}

	zx_status_t ThreadPool::CancelWaitOnPort(const zx::handle& handle, uint64_t key) {
	ZX_DEBUG_ASSERT(handle.is_valid());
	fbl::AutoLock pool_lock(&pool_lock_);

	if (!port_.is_valid()) {
	DEBUG_LOG("CancelWaitOnPort failed, port handle is invalid\n");
	return ZX_ERR_BAD_STATE;
	}

	return port_.cancel(handle, key);
	}

	zx_status_t ThreadPool::BindIrqToPort(const zx::handle& irq_handle, uint64_t key) {
	ZX_DEBUG_ASSERT(irq_handle.is_valid());
	fbl::AutoLock pool_lock(&pool_lock_);

	if (!port_.is_valid()) {
	DEBUG_LOG("BindIrqToPort failed, port handle is invalid\n");
	return ZX_ERR_BAD_STATE;
	}

	return zx_interrupt_bind(irq_handle.get(), port_.get(), key, 0u);
	}

	zx_koid_t ThreadPool::GetKey() const {
	return profile_koid_;
	}

	ThreadPool::ThreadPool(zx::profile profile) : profile_(std::move(profile)) {
	profile_koid_ = GetKoid(profile_);
	}

	void ThreadPool::PrintDebugPrefix() {
	printf("[ThreadPool %zu] ", GetKey());
	}

	zx_status_t ThreadPool::Init() {
	ZX_DEBUG_ASSERT(!port_.is_valid());

	zx_status_t res = zx::port::create(ZX_PORT_BIND_TO_INTERRUPT, &port_);
	if (res != ZX_OK) {
	LOG("Failed to create thread pool port (res %d)!\n", res);
	return res;
	}

	return ZX_OK;
	}

	void ThreadPool::InternalShutdown() {
	// Be careful when shutting down, a specific sequence needs to be followed.
	// See MG-1118 for details.
	decltype(active_domains_) domains_to_deactivate;
	{
	fbl::AutoLock lock(&pool_lock_);
	// If someone is already shutting us down, then we are done.
	if (pool_shutting_down_)
	return;

	// Prevent any new clients from joining the pool.
	pool_shutting_down_ = true;

	// Move the contents of the active domains list into a local variable so
	// that we don't need to hold onto the pool lock while we shut down the
	// domains.
	domains_to_deactivate = std::move(active_domains_);
	}

	// Deactivate any domains we may have still had assigned to us, then let go
	// of our references to them. Deactivation of domains should synchronize
	// will all pending operations in the domain (meaning that all references
	// have been recovered and no new wait operations will be posted)
	for (auto& domain : domains_to_deactivate)
	domain.Deactivate();

	domains_to_deactivate.clear();

	// Manually queue a quit message for each thread in the thread pool.
	{
	zx_port_packet pkt;
	memset(&pkt, 0, sizeof(pkt));
	pkt.type = ZX_PKT_TYPE_USER;

	fbl::AutoLock lock(&pool_lock_);
	for (__UNUSED const auto& thread : active_threads_) {
	__UNUSED zx_status_t res;
	res = port_.queue(&pkt);
	ZX_DEBUG_ASSERT(res == ZX_OK);
	}
	}

	// Synchronize with the threads as they exit.
	while (true) {
	fbl::unique_ptr<Thread> thread;
	{
	fbl::AutoLock lock(&pool_lock_);
	if (active_threads_.is_empty())
	break;

	thread = active_threads_.pop_front();
	}

	thread->Join();
	}
	}

	// static
	fbl::unique_ptr<ThreadPool::Thread> ThreadPool::Thread::Create(fbl::RefPtr<ThreadPool> pool,
	uint32_t id) {
	fbl::AllocChecker ac;
	fbl::unique_ptr<Thread> ret(new (&ac) Thread(std::move(pool), id));

	if (!ac.check())
	return nullptr;

	return ret;
	}

	ThreadPool::Thread::Thread(fbl::RefPtr<ThreadPool> pool, uint32_t id)
	: pool_(std::move(pool)),
	id_(id) {
	}

	zx_status_t ThreadPool::Thread::Start() {
	int c11_res = thrd_create(
	&thread_handle_,
	[](void* ctx) -> int { return static_cast<Thread*>(ctx)->Main(); },
	this);

	if (c11_res != thrd_success) {
	LOG("Failed to create new client thread (res %d)!\n", c11_res);
	// TODO(johngro) : translate C11 error
	return ZX_ERR_INTERNAL;
	}

	return ZX_OK;
	}

	void ThreadPool::Thread::Join() {
	// TODO(johngro) : Switch to native zircon threads so we can supply a
	// timeout to the join event.
	int musl_ret;
	thrd_join(thread_handle_, &musl_ret);
	ZX_DEBUG_ASSERT(pool_ == nullptr);
	}

	void ThreadPool::Thread::PrintDebugPrefix() const {
	printf("[Thread %03u-%zu] ", id_, pool_->GetKey());
	}

	int ThreadPool::Thread::Main() {
	zx_status_t res;

	ZX_DEBUG_ASSERT(pool_ != nullptr);
	DEBUG_LOG("Thread Starting\n");

	if (pool_->profile().is_valid()) {
	res = zx_object_set_profile(zx_thread_self(), pool_->profile().get(), 0);
	if (res != ZX_OK) {
	DEBUG_LOG("WARNING - Failed to set thread profile (res %d)\n", res);
	}
	}

	while (true) {
	zx_port_packet_t pkt;

	// TODO(johngro) : consider automatically shutting down if we have more
	// threads than clients.

	// Wait for there to be work to dispatch. We should never encounter an
	// error, but if we do, shut down.
	res = pool_->port().wait(zx::time::infinite(), &pkt);
	ZX_DEBUG_ASSERT(res == ZX_OK);

	// Is it time to exit?
	if ((res != ZX_OK) \|\| (pkt.type == ZX_PKT_TYPE_USER)) {
	break;
	}

	if ((pkt.type != ZX_PKT_TYPE_SIGNAL_ONE) &&
	(pkt.type != ZX_PKT_TYPE_INTERRUPT)) {
	LOG("Unexpected packet type (%u) in Thread pool!\n", pkt.type);
	continue;
	}

	// Reclaim our event source reference from the kernel.
	static_assert(sizeof(pkt.key) >= sizeof(EventSource*),
	"Port packet keys are not large enough to hold a pointer!");
	auto event_source =
	fbl::internal::MakeRefPtrNoAdopt(reinterpret_cast<EventSource*>(pkt.key));

	// Schedule the dispatch of the pending events for this event source.
	// If ScheduleDispatch returns a valid ExecutionDomain reference, then
	// actually go ahead and perform the dispatch of pending work for this
	// domain.
	ZX_DEBUG_ASSERT(event_source != nullptr);
	fbl::RefPtr<ExecutionDomain> domain = event_source->ScheduleDispatch(pkt);

	if (domain != nullptr)
	domain->DispatchPendingWork();
	}

	DEBUG_LOG("Client work thread shutting down\n");
	pool_.reset();

	return 0;
	}

	} // namespace dispatcher