diff --git a/adb/Android.bp b/adb/Android.bp index 583248b62..f6aede818 100644 --- a/adb/Android.bp +++ b/adb/Android.bp @@ -125,6 +125,7 @@ libadb_srcs = [ "adb_unique_fd.cpp", "adb_utils.cpp", "fdevent/fdevent.cpp", + "fdevent/fdevent_poll.cpp", "services.cpp", "sockets.cpp", "socket_spec.cpp", diff --git a/adb/fdevent/fdevent.cpp b/adb/fdevent/fdevent.cpp index eda946363..82bb0a224 100644 --- a/adb/fdevent/fdevent.cpp +++ b/adb/fdevent/fdevent.cpp @@ -1,64 +1,31 @@ -/* http://frotznet.googlecode.com/svn/trunk/utils/fdevent.c -** -** Copyright 2006, Brian Swetland -** -** 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 -** -** http://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. -*/ +/* + * Copyright 2006, Brian Swetland + * + * 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 + * + * http://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. + */ #define TRACE_TAG FDEVENT #include "sysdeps.h" -#include "fdevent.h" -#include #include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -#include -#include -#include #include -#include -#include -#include "adb_io.h" -#include "adb_trace.h" -#include "adb_unique_fd.h" -#include "adb_utils.h" -#include "sysdeps/chrono.h" +#include "fdevent.h" +#include "fdevent_poll.h" -#define FDE_EVENTMASK 0x00ff -#define FDE_STATEMASK 0xff00 - -#define FDE_ACTIVE 0x0100 -#define FDE_PENDING 0x0200 -#define FDE_CREATED 0x0400 - -static std::string dump_fde(const fdevent* fde) { +std::string dump_fde(const fdevent* fde) { std::string state; if (fde->state & FDE_ACTIVE) { state += "A"; @@ -82,486 +49,12 @@ static std::string dump_fde(const fdevent* fde) { state.c_str()); } -struct PollNode { - fdevent* fde; - adb_pollfd pollfd; - - explicit PollNode(fdevent* fde) : fde(fde) { - memset(&pollfd, 0, sizeof(pollfd)); - pollfd.fd = fde->fd.get(); - -#if defined(__linux__) - // Always enable POLLRDHUP, so the host server can take action when some clients disconnect. - // Then we can avoid leaving many sockets in CLOSE_WAIT state. See http://b/23314034. - pollfd.events = POLLRDHUP; -#endif - } -}; - -struct fdevent_context_poll : public fdevent_context { - virtual ~fdevent_context_poll() = default; - - virtual fdevent* Create(unique_fd fd, std::variant func, void* arg) final; - virtual unique_fd Destroy(fdevent* fde) final; - - virtual void Set(fdevent* fde, unsigned events) final; - virtual void Add(fdevent* fde, unsigned events) final; - virtual void Del(fdevent* fde, unsigned events) final; - virtual void SetTimeout(fdevent* fde, std::optional timeout) final; - - virtual void Loop() final; - - virtual void CheckMainThread() final; - - virtual void Run(std::function fn) final; - - virtual void TerminateLoop() final; - virtual size_t InstalledCount() final; - virtual void Reset() final; - - // All operations to fdevent should happen only in the main thread. - // That's why we don't need a lock for fdevent. - std::unordered_map poll_node_map_; - std::list pending_list_; - bool main_thread_valid_ = false; - uint64_t main_thread_id_ = 0; - uint64_t fdevent_id_ = 0; - - bool run_needs_flush_ = false; - unique_fd run_queue_notify_fd_; - std::mutex run_queue_mutex_; - std::deque> run_queue_ GUARDED_BY(run_queue_mutex_); - - std::atomic terminate_loop_ = false; -}; - static fdevent_context* g_ambient_fdevent_context = new fdevent_context_poll(); static fdevent_context* fdevent_get_ambient() { return g_ambient_fdevent_context; } -void fdevent_context_poll::CheckMainThread() { - if (main_thread_valid_) { - CHECK_EQ(main_thread_id_, android::base::GetThreadId()); - } -} - -fdevent* fdevent_context_poll::Create(unique_fd fd, std::variant func, - void* arg) { - CheckMainThread(); - CHECK_GE(fd.get(), 0); - - fdevent* fde = new fdevent(); - fde->id = fdevent_id_++; - fde->state = FDE_ACTIVE; - fde->fd = std::move(fd); - fde->func = func; - fde->arg = arg; - if (!set_file_block_mode(fde->fd, false)) { - // Here is not proper to handle the error. If it fails here, some error is - // likely to be detected by poll(), then we can let the callback function - // to handle it. - LOG(ERROR) << "failed to set non-blocking mode for fd " << fde->fd.get(); - } - auto pair = poll_node_map_.emplace(fde->fd.get(), PollNode(fde)); - CHECK(pair.second) << "install existing fd " << fde->fd.get(); - - fde->state |= FDE_CREATED; - return fde; -} - -unique_fd fdevent_context_poll::Destroy(fdevent* fde) { - CheckMainThread(); - if (!fde) { - return {}; - } - - if (!(fde->state & FDE_CREATED)) { - LOG(FATAL) << "destroying fde not created by fdevent_create(): " << dump_fde(fde); - } - - unique_fd result = std::move(fde->fd); - if (fde->state & FDE_ACTIVE) { - poll_node_map_.erase(result.get()); - - if (fde->state & FDE_PENDING) { - pending_list_.remove(fde); - } - fde->state = 0; - fde->events = 0; - } - - delete fde; - return result; -} - -void fdevent_context_poll::Set(fdevent* fde, unsigned events) { - CheckMainThread(); - events &= FDE_EVENTMASK; - if ((fde->state & FDE_EVENTMASK) == events) { - return; - } - CHECK(fde->state & FDE_ACTIVE); - - auto it = poll_node_map_.find(fde->fd.get()); - CHECK(it != poll_node_map_.end()); - PollNode& node = it->second; - if (events & FDE_READ) { - node.pollfd.events |= POLLIN; - } else { - node.pollfd.events &= ~POLLIN; - } - - if (events & FDE_WRITE) { - node.pollfd.events |= POLLOUT; - } else { - node.pollfd.events &= ~POLLOUT; - } - fde->state = (fde->state & FDE_STATEMASK) | events; - - D("fdevent_set: %s, events = %u", dump_fde(fde).c_str(), events); - - if (fde->state & FDE_PENDING) { - // If we are pending, make sure we don't signal an event that is no longer wanted. - fde->events &= events; - if (fde->events == 0) { - pending_list_.remove(fde); - fde->state &= ~FDE_PENDING; - } - } -} - -void fdevent_context_poll::Add(fdevent* fde, unsigned events) { - Set(fde, (fde->state & FDE_EVENTMASK) | events); -} - -void fdevent_context_poll::Del(fdevent* fde, unsigned events) { - CHECK(!(events & FDE_TIMEOUT)); - Set(fde, (fde->state & FDE_EVENTMASK) & ~events); -} - -void fdevent_context_poll::SetTimeout(fdevent* fde, - std::optional timeout) { - CheckMainThread(); - fde->timeout = timeout; - fde->last_active = std::chrono::steady_clock::now(); -} - -static std::string dump_pollfds(const std::vector& pollfds) { - std::string result; - for (const auto& pollfd : pollfds) { - std::string op; - if (pollfd.events & POLLIN) { - op += "R"; - } - if (pollfd.events & POLLOUT) { - op += "W"; - } - android::base::StringAppendF(&result, " %d(%s)", pollfd.fd, op.c_str()); - } - return result; -} - -static std::optional calculate_timeout(fdevent_context_poll* ctx) { - std::optional result = std::nullopt; - auto now = std::chrono::steady_clock::now(); - ctx->CheckMainThread(); - - for (const auto& [fd, pollnode] : ctx->poll_node_map_) { - UNUSED(fd); - auto timeout_opt = pollnode.fde->timeout; - if (timeout_opt) { - auto deadline = pollnode.fde->last_active + *timeout_opt; - auto time_left = std::chrono::duration_cast(deadline - now); - if (time_left < std::chrono::milliseconds::zero()) { - time_left = std::chrono::milliseconds::zero(); - } - - if (!result) { - result = time_left; - } else { - result = std::min(*result, time_left); - } - } - } - - return result; -} - -static void fdevent_process(fdevent_context_poll* ctx) { - std::vector pollfds; - for (const auto& pair : ctx->poll_node_map_) { - pollfds.push_back(pair.second.pollfd); - } - CHECK_GT(pollfds.size(), 0u); - D("poll(), pollfds = %s", dump_pollfds(pollfds).c_str()); - - auto timeout = calculate_timeout(ctx); - int timeout_ms; - if (!timeout) { - timeout_ms = -1; - } else { - timeout_ms = timeout->count(); - } - - int ret = adb_poll(&pollfds[0], pollfds.size(), timeout_ms); - if (ret == -1) { - PLOG(ERROR) << "poll(), ret = " << ret; - return; - } - - auto post_poll = std::chrono::steady_clock::now(); - - for (const auto& pollfd : pollfds) { - if (pollfd.revents != 0) { - D("for fd %d, revents = %x", pollfd.fd, pollfd.revents); - } - unsigned events = 0; - if (pollfd.revents & POLLIN) { - events |= FDE_READ; - } - if (pollfd.revents & POLLOUT) { - events |= FDE_WRITE; - } - if (pollfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { - // We fake a read, as the rest of the code assumes that errors will - // be detected at that point. - events |= FDE_READ | FDE_ERROR; - } -#if defined(__linux__) - if (pollfd.revents & POLLRDHUP) { - events |= FDE_READ | FDE_ERROR; - } -#endif - auto it = ctx->poll_node_map_.find(pollfd.fd); - CHECK(it != ctx->poll_node_map_.end()); - fdevent* fde = it->second.fde; - - if (events == 0) { - // Check for timeout. - if (fde->timeout) { - auto deadline = fde->last_active + *fde->timeout; - if (deadline < post_poll) { - events |= FDE_TIMEOUT; - } - } - } - - if (events != 0) { - CHECK_EQ(fde->fd.get(), pollfd.fd); - fde->events |= events; - fde->last_active = post_poll; - D("%s got events %x", dump_fde(fde).c_str(), events); - fde->state |= FDE_PENDING; - ctx->pending_list_.push_back(fde); - } - } -} - -template -struct always_false : std::false_type {}; - -static void fdevent_call_fdfunc(fdevent* fde) { - unsigned events = fde->events; - fde->events = 0; - CHECK(fde->state & FDE_PENDING); - fde->state &= (~FDE_PENDING); - D("fdevent_call_fdfunc %s", dump_fde(fde).c_str()); - std::visit( - [&](auto&& f) { - using F = std::decay_t; - if constexpr (std::is_same_v) { - f(fde->fd.get(), events, fde->arg); - } else if constexpr (std::is_same_v) { - f(fde, events, fde->arg); - } else { - static_assert(always_false::value, "non-exhaustive visitor"); - } - }, - fde->func); -} - -static void fdevent_run_flush(fdevent_context_poll* ctx) EXCLUDES(ctx->run_queue_mutex_) { - // We need to be careful around reentrancy here, since a function we call can queue up another - // function. - while (true) { - std::function fn; - { - std::lock_guard lock(ctx->run_queue_mutex_); - if (ctx->run_queue_.empty()) { - break; - } - fn = ctx->run_queue_.front(); - ctx->run_queue_.pop_front(); - } - fn(); - } -} - -static void fdevent_run_func(int fd, unsigned ev, void* data) { - CHECK_GE(fd, 0); - CHECK(ev & FDE_READ); - - bool* run_needs_flush = static_cast(data); - char buf[1024]; - - // Empty the fd. - if (adb_read(fd, buf, sizeof(buf)) == -1) { - PLOG(FATAL) << "failed to empty run queue notify fd"; - } - - // Mark that we need to flush, and then run it at the end of fdevent_loop. - *run_needs_flush = true; -} - -static void fdevent_run_setup(fdevent_context_poll* ctx) { - { - std::lock_guard lock(ctx->run_queue_mutex_); - CHECK(ctx->run_queue_notify_fd_.get() == -1); - int s[2]; - if (adb_socketpair(s) != 0) { - PLOG(FATAL) << "failed to create run queue notify socketpair"; - } - - if (!set_file_block_mode(s[0], false) || !set_file_block_mode(s[1], false)) { - PLOG(FATAL) << "failed to make run queue notify socket nonblocking"; - } - - ctx->run_queue_notify_fd_.reset(s[0]); - fdevent* fde = ctx->Create(unique_fd(s[1]), fdevent_run_func, &ctx->run_needs_flush_); - CHECK(fde != nullptr); - ctx->Add(fde, FDE_READ); - } - - fdevent_run_flush(ctx); -} - -void fdevent_context_poll::Run(std::function fn) { - std::lock_guard lock(run_queue_mutex_); - run_queue_.push_back(std::move(fn)); - - // run_queue_notify_fd could still be -1 if we're called before fdevent has finished setting up. - // In that case, rely on the setup code to flush the queue without a notification being needed. - if (run_queue_notify_fd_ != -1) { - int rc = adb_write(run_queue_notify_fd_.get(), "", 1); - - // It's possible that we get EAGAIN here, if lots of notifications came in while handling. - if (rc == 0) { - PLOG(FATAL) << "run queue notify fd was closed?"; - } else if (rc == -1 && errno != EAGAIN) { - PLOG(FATAL) << "failed to write to run queue notify fd"; - } - } -} - -static void fdevent_check_spin(fdevent_context_poll* ctx, uint64_t cycle) { - // Check to see if we're spinning because we forgot about an fdevent - // by keeping track of how long fdevents have been continuously pending. - struct SpinCheck { - fdevent* fde; - android::base::boot_clock::time_point timestamp; - uint64_t cycle; - }; - - // TODO: Move this into the base fdevent_context. - static auto& g_continuously_pending = *new std::unordered_map(); - static auto last_cycle = android::base::boot_clock::now(); - - auto now = android::base::boot_clock::now(); - if (now - last_cycle > 10ms) { - // We're not spinning. - g_continuously_pending.clear(); - last_cycle = now; - return; - } - last_cycle = now; - - for (auto* fde : ctx->pending_list_) { - auto it = g_continuously_pending.find(fde->id); - if (it == g_continuously_pending.end()) { - g_continuously_pending[fde->id] = - SpinCheck{.fde = fde, .timestamp = now, .cycle = cycle}; - } else { - it->second.cycle = cycle; - } - } - - for (auto it = g_continuously_pending.begin(); it != g_continuously_pending.end();) { - if (it->second.cycle != cycle) { - it = g_continuously_pending.erase(it); - } else { - // Use an absurdly long window, since all we really care about is - // getting a bugreport eventually. - if (now - it->second.timestamp > 300s) { - LOG(FATAL_WITHOUT_ABORT) - << "detected spin in fdevent: " << dump_fde(it->second.fde); -#if defined(__linux__) - int fd = it->second.fde->fd.get(); - std::string fd_path = android::base::StringPrintf("/proc/self/fd/%d", fd); - std::string path; - if (!android::base::Readlink(fd_path, &path)) { - PLOG(FATAL_WITHOUT_ABORT) << "readlink of fd " << fd << " failed"; - } - LOG(FATAL_WITHOUT_ABORT) << "fd " << fd << " = " << path; -#endif - abort(); - } - ++it; - } - } -} - -void fdevent_context_poll::Loop() { - this->main_thread_id_ = android::base::GetThreadId(); - this->main_thread_valid_ = true; - fdevent_run_setup(this); - - uint64_t cycle = 0; - while (true) { - if (terminate_loop_) { - return; - } - - D("--- --- waiting for events"); - - fdevent_process(this); - - fdevent_check_spin(this, cycle++); - - while (!pending_list_.empty()) { - fdevent* fde = pending_list_.front(); - pending_list_.pop_front(); - fdevent_call_fdfunc(fde); - } - - if (run_needs_flush_) { - fdevent_run_flush(this); - run_needs_flush_ = false; - } - } -} - -void fdevent_context_poll::TerminateLoop() { - terminate_loop_ = true; -} - -size_t fdevent_context_poll::InstalledCount() { - return poll_node_map_.size(); -} - -void fdevent_context_poll::Reset() { - poll_node_map_.clear(); - pending_list_.clear(); - - std::lock_guard lock(run_queue_mutex_); - run_queue_notify_fd_.reset(); - run_queue_.clear(); - - main_thread_valid_ = false; - terminate_loop_ = false; -} - fdevent* fdevent_create(int fd, fd_func func, void* arg) { unique_fd ufd(fd); return fdevent_get_ambient()->Create(std::move(ufd), func, arg); diff --git a/adb/fdevent/fdevent.h b/adb/fdevent/fdevent.h index a953cc17d..7bc50f4b2 100644 --- a/adb/fdevent/fdevent.h +++ b/adb/fdevent/fdevent.h @@ -33,10 +33,19 @@ #define FDE_ERROR 0x0004 #define FDE_TIMEOUT 0x0008 +// Internal states. +#define FDE_EVENTMASK 0x00ff +#define FDE_STATEMASK 0xff00 + +#define FDE_ACTIVE 0x0100 +#define FDE_PENDING 0x0200 +#define FDE_CREATED 0x0400 + typedef void (*fd_func)(int fd, unsigned events, void *userdata); typedef void (*fd_func2)(struct fdevent* fde, unsigned events, void* userdata); struct fdevent; +std::string dump_fde(const fdevent* fde); struct fdevent_context { virtual ~fdevent_context() = default; diff --git a/adb/fdevent/fdevent_poll.cpp b/adb/fdevent/fdevent_poll.cpp new file mode 100644 index 000000000..63ef3f1fa --- /dev/null +++ b/adb/fdevent/fdevent_poll.cpp @@ -0,0 +1,472 @@ +/* + * Copyright (C) 2019 The Android Open Source Project + * + * 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 + * + * http://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. + */ + +#define TRACE_TAG FDEVENT + +#include "sysdeps.h" +#include "fdevent_poll.h" + +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +#include "adb_io.h" +#include "adb_trace.h" +#include "adb_unique_fd.h" +#include "adb_utils.h" +#include "fdevent.h" +#include "sysdeps/chrono.h" + +void fdevent_context_poll::CheckMainThread() { + if (main_thread_valid_) { + CHECK_EQ(main_thread_id_, android::base::GetThreadId()); + } +} + +fdevent* fdevent_context_poll::Create(unique_fd fd, std::variant func, + void* arg) { + CheckMainThread(); + CHECK_GE(fd.get(), 0); + + fdevent* fde = new fdevent(); + fde->id = fdevent_id_++; + fde->state = FDE_ACTIVE; + fde->fd = std::move(fd); + fde->func = func; + fde->arg = arg; + if (!set_file_block_mode(fde->fd, false)) { + // Here is not proper to handle the error. If it fails here, some error is + // likely to be detected by poll(), then we can let the callback function + // to handle it. + LOG(ERROR) << "failed to set non-blocking mode for fd " << fde->fd.get(); + } + auto pair = poll_node_map_.emplace(fde->fd.get(), PollNode(fde)); + CHECK(pair.second) << "install existing fd " << fde->fd.get(); + + fde->state |= FDE_CREATED; + return fde; +} + +unique_fd fdevent_context_poll::Destroy(fdevent* fde) { + CheckMainThread(); + if (!fde) { + return {}; + } + + if (!(fde->state & FDE_CREATED)) { + LOG(FATAL) << "destroying fde not created by fdevent_create(): " << dump_fde(fde); + } + + unique_fd result = std::move(fde->fd); + if (fde->state & FDE_ACTIVE) { + poll_node_map_.erase(result.get()); + + if (fde->state & FDE_PENDING) { + pending_list_.remove(fde); + } + fde->state = 0; + fde->events = 0; + } + + delete fde; + return result; +} + +void fdevent_context_poll::Set(fdevent* fde, unsigned events) { + CheckMainThread(); + events &= FDE_EVENTMASK; + if ((fde->state & FDE_EVENTMASK) == events) { + return; + } + CHECK(fde->state & FDE_ACTIVE); + + auto it = poll_node_map_.find(fde->fd.get()); + CHECK(it != poll_node_map_.end()); + PollNode& node = it->second; + if (events & FDE_READ) { + node.pollfd.events |= POLLIN; + } else { + node.pollfd.events &= ~POLLIN; + } + + if (events & FDE_WRITE) { + node.pollfd.events |= POLLOUT; + } else { + node.pollfd.events &= ~POLLOUT; + } + fde->state = (fde->state & FDE_STATEMASK) | events; + + D("fdevent_set: %s, events = %u", dump_fde(fde).c_str(), events); + + if (fde->state & FDE_PENDING) { + // If we are pending, make sure we don't signal an event that is no longer wanted. + fde->events &= events; + if (fde->events == 0) { + pending_list_.remove(fde); + fde->state &= ~FDE_PENDING; + } + } +} + +void fdevent_context_poll::Add(fdevent* fde, unsigned events) { + Set(fde, (fde->state & FDE_EVENTMASK) | events); +} + +void fdevent_context_poll::Del(fdevent* fde, unsigned events) { + CHECK(!(events & FDE_TIMEOUT)); + Set(fde, (fde->state & FDE_EVENTMASK) & ~events); +} + +void fdevent_context_poll::SetTimeout(fdevent* fde, + std::optional timeout) { + CheckMainThread(); + fde->timeout = timeout; + fde->last_active = std::chrono::steady_clock::now(); +} + +static std::string dump_pollfds(const std::vector& pollfds) { + std::string result; + for (const auto& pollfd : pollfds) { + std::string op; + if (pollfd.events & POLLIN) { + op += "R"; + } + if (pollfd.events & POLLOUT) { + op += "W"; + } + android::base::StringAppendF(&result, " %d(%s)", pollfd.fd, op.c_str()); + } + return result; +} + +static std::optional calculate_timeout(fdevent_context_poll* ctx) { + std::optional result = std::nullopt; + auto now = std::chrono::steady_clock::now(); + ctx->CheckMainThread(); + + for (const auto& [fd, pollnode] : ctx->poll_node_map_) { + UNUSED(fd); + auto timeout_opt = pollnode.fde->timeout; + if (timeout_opt) { + auto deadline = pollnode.fde->last_active + *timeout_opt; + auto time_left = std::chrono::duration_cast(deadline - now); + if (time_left < std::chrono::milliseconds::zero()) { + time_left = std::chrono::milliseconds::zero(); + } + + if (!result) { + result = time_left; + } else { + result = std::min(*result, time_left); + } + } + } + + return result; +} + +static void fdevent_process(fdevent_context_poll* ctx) { + std::vector pollfds; + for (const auto& pair : ctx->poll_node_map_) { + pollfds.push_back(pair.second.pollfd); + } + CHECK_GT(pollfds.size(), 0u); + D("poll(), pollfds = %s", dump_pollfds(pollfds).c_str()); + + auto timeout = calculate_timeout(ctx); + int timeout_ms; + if (!timeout) { + timeout_ms = -1; + } else { + timeout_ms = timeout->count(); + } + + int ret = adb_poll(&pollfds[0], pollfds.size(), timeout_ms); + if (ret == -1) { + PLOG(ERROR) << "poll(), ret = " << ret; + return; + } + + auto post_poll = std::chrono::steady_clock::now(); + + for (const auto& pollfd : pollfds) { + if (pollfd.revents != 0) { + D("for fd %d, revents = %x", pollfd.fd, pollfd.revents); + } + unsigned events = 0; + if (pollfd.revents & POLLIN) { + events |= FDE_READ; + } + if (pollfd.revents & POLLOUT) { + events |= FDE_WRITE; + } + if (pollfd.revents & (POLLERR | POLLHUP | POLLNVAL)) { + // We fake a read, as the rest of the code assumes that errors will + // be detected at that point. + events |= FDE_READ | FDE_ERROR; + } +#if defined(__linux__) + if (pollfd.revents & POLLRDHUP) { + events |= FDE_READ | FDE_ERROR; + } +#endif + auto it = ctx->poll_node_map_.find(pollfd.fd); + CHECK(it != ctx->poll_node_map_.end()); + fdevent* fde = it->second.fde; + + if (events == 0) { + // Check for timeout. + if (fde->timeout) { + auto deadline = fde->last_active + *fde->timeout; + if (deadline < post_poll) { + events |= FDE_TIMEOUT; + } + } + } + + if (events != 0) { + CHECK_EQ(fde->fd.get(), pollfd.fd); + fde->events |= events; + fde->last_active = post_poll; + D("%s got events %x", dump_fde(fde).c_str(), events); + fde->state |= FDE_PENDING; + ctx->pending_list_.push_back(fde); + } + } +} + +template +struct always_false : std::false_type {}; + +static void fdevent_call_fdfunc(fdevent* fde) { + unsigned events = fde->events; + fde->events = 0; + CHECK(fde->state & FDE_PENDING); + fde->state &= (~FDE_PENDING); + D("fdevent_call_fdfunc %s", dump_fde(fde).c_str()); + std::visit( + [&](auto&& f) { + using F = std::decay_t; + if constexpr (std::is_same_v) { + f(fde->fd.get(), events, fde->arg); + } else if constexpr (std::is_same_v) { + f(fde, events, fde->arg); + } else { + static_assert(always_false::value, "non-exhaustive visitor"); + } + }, + fde->func); +} + +static void fdevent_run_flush(fdevent_context_poll* ctx) EXCLUDES(ctx->run_queue_mutex_) { + // We need to be careful around reentrancy here, since a function we call can queue up another + // function. + while (true) { + std::function fn; + { + std::lock_guard lock(ctx->run_queue_mutex_); + if (ctx->run_queue_.empty()) { + break; + } + fn = ctx->run_queue_.front(); + ctx->run_queue_.pop_front(); + } + fn(); + } +} + +static void fdevent_run_func(int fd, unsigned ev, void* data) { + CHECK_GE(fd, 0); + CHECK(ev & FDE_READ); + + bool* run_needs_flush = static_cast(data); + char buf[1024]; + + // Empty the fd. + if (adb_read(fd, buf, sizeof(buf)) == -1) { + PLOG(FATAL) << "failed to empty run queue notify fd"; + } + + // Mark that we need to flush, and then run it at the end of fdevent_loop. + *run_needs_flush = true; +} + +static void fdevent_run_setup(fdevent_context_poll* ctx) { + { + std::lock_guard lock(ctx->run_queue_mutex_); + CHECK(ctx->run_queue_notify_fd_.get() == -1); + int s[2]; + if (adb_socketpair(s) != 0) { + PLOG(FATAL) << "failed to create run queue notify socketpair"; + } + + if (!set_file_block_mode(s[0], false) || !set_file_block_mode(s[1], false)) { + PLOG(FATAL) << "failed to make run queue notify socket nonblocking"; + } + + ctx->run_queue_notify_fd_.reset(s[0]); + fdevent* fde = ctx->Create(unique_fd(s[1]), fdevent_run_func, &ctx->run_needs_flush_); + CHECK(fde != nullptr); + ctx->Add(fde, FDE_READ); + } + + fdevent_run_flush(ctx); +} + +void fdevent_context_poll::Run(std::function fn) { + std::lock_guard lock(run_queue_mutex_); + run_queue_.push_back(std::move(fn)); + + // run_queue_notify_fd could still be -1 if we're called before fdevent has finished setting up. + // In that case, rely on the setup code to flush the queue without a notification being needed. + if (run_queue_notify_fd_ != -1) { + int rc = adb_write(run_queue_notify_fd_.get(), "", 1); + + // It's possible that we get EAGAIN here, if lots of notifications came in while handling. + if (rc == 0) { + PLOG(FATAL) << "run queue notify fd was closed?"; + } else if (rc == -1 && errno != EAGAIN) { + PLOG(FATAL) << "failed to write to run queue notify fd"; + } + } +} + +static void fdevent_check_spin(fdevent_context_poll* ctx, uint64_t cycle) { + // Check to see if we're spinning because we forgot about an fdevent + // by keeping track of how long fdevents have been continuously pending. + struct SpinCheck { + fdevent* fde; + android::base::boot_clock::time_point timestamp; + uint64_t cycle; + }; + + // TODO: Move this into the base fdevent_context. + static auto& g_continuously_pending = *new std::unordered_map(); + static auto last_cycle = android::base::boot_clock::now(); + + auto now = android::base::boot_clock::now(); + if (now - last_cycle > 10ms) { + // We're not spinning. + g_continuously_pending.clear(); + last_cycle = now; + return; + } + last_cycle = now; + + for (auto* fde : ctx->pending_list_) { + auto it = g_continuously_pending.find(fde->id); + if (it == g_continuously_pending.end()) { + g_continuously_pending[fde->id] = + SpinCheck{.fde = fde, .timestamp = now, .cycle = cycle}; + } else { + it->second.cycle = cycle; + } + } + + for (auto it = g_continuously_pending.begin(); it != g_continuously_pending.end();) { + if (it->second.cycle != cycle) { + it = g_continuously_pending.erase(it); + } else { + // Use an absurdly long window, since all we really care about is + // getting a bugreport eventually. + if (now - it->second.timestamp > 300s) { + LOG(FATAL_WITHOUT_ABORT) + << "detected spin in fdevent: " << dump_fde(it->second.fde); +#if defined(__linux__) + int fd = it->second.fde->fd.get(); + std::string fd_path = android::base::StringPrintf("/proc/self/fd/%d", fd); + std::string path; + if (!android::base::Readlink(fd_path, &path)) { + PLOG(FATAL_WITHOUT_ABORT) << "readlink of fd " << fd << " failed"; + } + LOG(FATAL_WITHOUT_ABORT) << "fd " << fd << " = " << path; +#endif + abort(); + } + ++it; + } + } +} + +void fdevent_context_poll::Loop() { + this->main_thread_id_ = android::base::GetThreadId(); + this->main_thread_valid_ = true; + fdevent_run_setup(this); + + uint64_t cycle = 0; + while (true) { + if (terminate_loop_) { + return; + } + + D("--- --- waiting for events"); + + fdevent_process(this); + + fdevent_check_spin(this, cycle++); + + while (!pending_list_.empty()) { + fdevent* fde = pending_list_.front(); + pending_list_.pop_front(); + fdevent_call_fdfunc(fde); + } + + if (run_needs_flush_) { + fdevent_run_flush(this); + run_needs_flush_ = false; + } + } +} + +void fdevent_context_poll::TerminateLoop() { + terminate_loop_ = true; +} + +size_t fdevent_context_poll::InstalledCount() { + return poll_node_map_.size(); +} + +void fdevent_context_poll::Reset() { + poll_node_map_.clear(); + pending_list_.clear(); + + std::lock_guard lock(run_queue_mutex_); + run_queue_notify_fd_.reset(); + run_queue_.clear(); + + main_thread_valid_ = false; + terminate_loop_ = false; +} diff --git a/adb/fdevent/fdevent_poll.h b/adb/fdevent/fdevent_poll.h new file mode 100644 index 000000000..aa82ee52a --- /dev/null +++ b/adb/fdevent/fdevent_poll.h @@ -0,0 +1,81 @@ +#pragma once + +/* + * Copyright (C) 2019 The Android Open Source Project + * + * 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 + * + * http://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 "sysdeps.h" + +#include +#include +#include +#include + +#include + +#include "fdevent.h" + +struct PollNode { + fdevent* fde; + adb_pollfd pollfd; + + explicit PollNode(fdevent* fde) : fde(fde) { + memset(&pollfd, 0, sizeof(pollfd)); + pollfd.fd = fde->fd.get(); + +#if defined(__linux__) + // Always enable POLLRDHUP, so the host server can take action when some clients disconnect. + // Then we can avoid leaving many sockets in CLOSE_WAIT state. See http://b/23314034. + pollfd.events = POLLRDHUP; +#endif + } +}; + +struct fdevent_context_poll : public fdevent_context { + virtual ~fdevent_context_poll() = default; + + virtual fdevent* Create(unique_fd fd, std::variant func, void* arg) final; + virtual unique_fd Destroy(fdevent* fde) final; + + virtual void Set(fdevent* fde, unsigned events) final; + virtual void Add(fdevent* fde, unsigned events) final; + virtual void Del(fdevent* fde, unsigned events) final; + virtual void SetTimeout(fdevent* fde, std::optional timeout) final; + + virtual void Loop() final; + + virtual void CheckMainThread() final; + + virtual void Run(std::function fn) final; + + virtual void TerminateLoop() final; + virtual size_t InstalledCount() final; + virtual void Reset() final; + + // All operations to fdevent should happen only in the main thread. + // That's why we don't need a lock for fdevent. + std::unordered_map poll_node_map_; + std::list pending_list_; + bool main_thread_valid_ = false; + uint64_t main_thread_id_ = 0; + uint64_t fdevent_id_ = 0; + + bool run_needs_flush_ = false; + unique_fd run_queue_notify_fd_; + std::mutex run_queue_mutex_; + std::deque> run_queue_ GUARDED_BY(run_queue_mutex_); + + std::atomic terminate_loop_ = false; +};