From 2d03ad42ed26cb2ea030c8f386b5122219f1ffa2 Mon Sep 17 00:00:00 2001 From: Josh Gao Date: Mon, 8 Jul 2019 14:36:02 -0700 Subject: [PATCH 01/11] adb: add superfluous include. Resolve a merge conflict between internal and aosp by making aosp look like internal. Test: none Change-Id: I7bea6ab61ad45ff20a3d07abf3f73903a7c05585 Merged-In: If73b81ca73ba4d64763cf49c1bbe42de81fa1cb6 --- adb/daemon/auth.cpp | 1 + 1 file changed, 1 insertion(+) diff --git a/adb/daemon/auth.cpp b/adb/daemon/auth.cpp index 1800f84b9..00268a71b 100644 --- a/adb/daemon/auth.cpp +++ b/adb/daemon/auth.cpp @@ -18,6 +18,7 @@ #include "adb.h" #include "adb_auth.h" +#include "adb_io.h" #include "fdevent.h" #include "sysdeps.h" #include "transport.h" From 57e09b16ca94618b17ccd5b74caec894df977f28 Mon Sep 17 00:00:00 2001 From: Josh Gao Date: Fri, 28 Jun 2019 13:50:37 -0700 Subject: [PATCH 02/11] adb: move fdevent to its own folder. Preparatory refactoring for platform-specific implementations. Test: mma Change-Id: I0f600122ac89241788c5f3300f362fd9ef02ddcd --- adb/Android.bp | 4 ++-- adb/adb.h | 2 +- adb/adb_listeners_test.cpp | 2 +- adb/client/transport_mdns.cpp | 2 +- adb/daemon/auth.cpp | 2 +- adb/daemon/framebuffer_service.cpp | 1 - adb/{ => fdevent}/fdevent.cpp | 0 adb/{ => fdevent}/fdevent.h | 0 adb/{ => fdevent}/fdevent_test.cpp | 0 adb/{ => fdevent}/fdevent_test.h | 0 adb/socket.h | 2 +- adb/socket_test.cpp | 4 ++-- adb/sysdeps.h | 2 -- adb/transport.cpp | 2 +- adb/transport_test.cpp | 2 +- 15 files changed, 11 insertions(+), 14 deletions(-) rename adb/{ => fdevent}/fdevent.cpp (100%) rename adb/{ => fdevent}/fdevent.h (100%) rename adb/{ => fdevent}/fdevent_test.cpp (100%) rename adb/{ => fdevent}/fdevent_test.h (100%) diff --git a/adb/Android.bp b/adb/Android.bp index b6aff3ee5..583248b62 100644 --- a/adb/Android.bp +++ b/adb/Android.bp @@ -124,7 +124,7 @@ libadb_srcs = [ "adb_trace.cpp", "adb_unique_fd.cpp", "adb_utils.cpp", - "fdevent.cpp", + "fdevent/fdevent.cpp", "services.cpp", "sockets.cpp", "socket_spec.cpp", @@ -144,7 +144,7 @@ libadb_test_srcs = [ "adb_io_test.cpp", "adb_listeners_test.cpp", "adb_utils_test.cpp", - "fdevent_test.cpp", + "fdevent/fdevent_test.cpp", "socket_spec_test.cpp", "socket_test.cpp", "sysdeps_test.cpp", diff --git a/adb/adb.h b/adb/adb.h index 3a6f059b8..352b2fe1e 100644 --- a/adb/adb.h +++ b/adb/adb.h @@ -26,7 +26,7 @@ #include #include "adb_trace.h" -#include "fdevent.h" +#include "fdevent/fdevent.h" #include "socket.h" #include "types.h" #include "usb.h" diff --git a/adb/adb_listeners_test.cpp b/adb/adb_listeners_test.cpp index b697769d2..a7e2deaf6 100644 --- a/adb/adb_listeners_test.cpp +++ b/adb/adb_listeners_test.cpp @@ -21,7 +21,7 @@ #include #include -#include "fdevent.h" +#include "fdevent/fdevent.h" #include "sysdeps.h" #include "transport.h" diff --git a/adb/client/transport_mdns.cpp b/adb/client/transport_mdns.cpp index 283fac554..1a34384d2 100644 --- a/adb/client/transport_mdns.cpp +++ b/adb/client/transport_mdns.cpp @@ -31,7 +31,7 @@ #include "adb_mdns.h" #include "adb_trace.h" -#include "fdevent.h" +#include "fdevent/fdevent.h" #include "sysdeps.h" static DNSServiceRef service_ref; diff --git a/adb/daemon/auth.cpp b/adb/daemon/auth.cpp index 00268a71b..2b8f4617f 100644 --- a/adb/daemon/auth.cpp +++ b/adb/daemon/auth.cpp @@ -19,7 +19,7 @@ #include "adb.h" #include "adb_auth.h" #include "adb_io.h" -#include "fdevent.h" +#include "fdevent/fdevent.h" #include "sysdeps.h" #include "transport.h" diff --git a/adb/daemon/framebuffer_service.cpp b/adb/daemon/framebuffer_service.cpp index 2a6418ade..676f8e9db 100644 --- a/adb/daemon/framebuffer_service.cpp +++ b/adb/daemon/framebuffer_service.cpp @@ -33,7 +33,6 @@ #include "adb.h" #include "adb_io.h" #include "adb_utils.h" -#include "fdevent.h" /* TODO: ** - sync with vsync to avoid tearing diff --git a/adb/fdevent.cpp b/adb/fdevent/fdevent.cpp similarity index 100% rename from adb/fdevent.cpp rename to adb/fdevent/fdevent.cpp diff --git a/adb/fdevent.h b/adb/fdevent/fdevent.h similarity index 100% rename from adb/fdevent.h rename to adb/fdevent/fdevent.h diff --git a/adb/fdevent_test.cpp b/adb/fdevent/fdevent_test.cpp similarity index 100% rename from adb/fdevent_test.cpp rename to adb/fdevent/fdevent_test.cpp diff --git a/adb/fdevent_test.h b/adb/fdevent/fdevent_test.h similarity index 100% rename from adb/fdevent_test.h rename to adb/fdevent/fdevent_test.h diff --git a/adb/socket.h b/adb/socket.h index b8c559a90..4276851d2 100644 --- a/adb/socket.h +++ b/adb/socket.h @@ -24,7 +24,7 @@ #include #include "adb_unique_fd.h" -#include "fdevent.h" +#include "fdevent/fdevent.h" #include "types.h" class atransport; diff --git a/adb/socket_test.cpp b/adb/socket_test.cpp index 5e28f7601..1601ff0ac 100644 --- a/adb/socket_test.cpp +++ b/adb/socket_test.cpp @@ -14,7 +14,7 @@ * limitations under the License. */ -#include "fdevent.h" +#include "fdevent/fdevent.h" #include @@ -29,7 +29,7 @@ #include "adb.h" #include "adb_io.h" -#include "fdevent_test.h" +#include "fdevent/fdevent_test.h" #include "socket.h" #include "sysdeps.h" #include "sysdeps/chrono.h" diff --git a/adb/sysdeps.h b/adb/sysdeps.h index 78abba54a..b0e7fa097 100644 --- a/adb/sysdeps.h +++ b/adb/sysdeps.h @@ -64,8 +64,6 @@ #include // unique_ptr #include -#include "fdevent.h" - #define OS_PATH_SEPARATORS "\\/" #define OS_PATH_SEPARATOR '\\' #define OS_PATH_SEPARATOR_STR "\\" diff --git a/adb/transport.cpp b/adb/transport.cpp index 841865afc..8bc925fab 100644 --- a/adb/transport.cpp +++ b/adb/transport.cpp @@ -49,7 +49,7 @@ #include "adb_io.h" #include "adb_trace.h" #include "adb_utils.h" -#include "fdevent.h" +#include "fdevent/fdevent.h" #include "sysdeps/chrono.h" using android::base::ScopedLockAssertion; diff --git a/adb/transport_test.cpp b/adb/transport_test.cpp index b66f8fa80..00beb3a2b 100644 --- a/adb/transport_test.cpp +++ b/adb/transport_test.cpp @@ -19,7 +19,7 @@ #include #include "adb.h" -#include "fdevent_test.h" +#include "fdevent/fdevent_test.h" struct TransportTest : public FdeventTest {}; From c2cf12117473f45ad5ebc562eac277225ef60427 Mon Sep 17 00:00:00 2001 From: Josh Gao Date: Fri, 28 Jun 2019 16:34:36 -0700 Subject: [PATCH 03/11] adb: de-globalize fdevent context. Test: adb_test Test: adbd_test Change-Id: If711257d831b9090b38412821466c1d358d53d2d --- adb/fdevent/fdevent.cpp | 354 ++++++++++++++++++++++++---------------- adb/fdevent/fdevent.h | 57 +++++-- 2 files changed, 251 insertions(+), 160 deletions(-) diff --git a/adb/fdevent/fdevent.cpp b/adb/fdevent/fdevent.cpp index 32f908636..eda946363 100644 --- a/adb/fdevent/fdevent.cpp +++ b/adb/fdevent/fdevent.cpp @@ -58,48 +58,6 @@ #define FDE_PENDING 0x0200 #define FDE_CREATED 0x0400 -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 - } -}; - -// All operations to fdevent should happen only in the main thread. -// That's why we don't need a lock for fdevent. -static auto& g_poll_node_map = *new std::unordered_map(); -static auto& g_pending_list = *new std::list(); -static std::atomic terminate_loop(false); -static bool main_thread_valid; -static uint64_t main_thread_id; - -static uint64_t fdevent_id; - -static bool run_needs_flush = false; -static auto& run_queue_notify_fd = *new unique_fd(); -static auto& run_queue_mutex = *new std::mutex(); -static auto& run_queue GUARDED_BY(run_queue_mutex) = *new std::deque>(); - -void check_main_thread() { - if (main_thread_valid) { - CHECK_EQ(main_thread_id, android::base::GetThreadId()); - } -} - -void set_main_thread() { - main_thread_valid = true; - main_thread_id = android::base::GetThreadId(); -} - static std::string dump_fde(const fdevent* fde) { std::string state; if (fde->state & FDE_ACTIVE) { @@ -124,40 +82,97 @@ static std::string dump_fde(const fdevent* fde) { state.c_str()); } -template -static fdevent* fdevent_create_impl(int fd, F func, void* arg) { - check_main_thread(); - CHECK_GE(fd, 0); +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->id = fdevent_id_++; fde->state = FDE_ACTIVE; - fde->fd.reset(fd); + fde->fd = std::move(fd); fde->func = func; fde->arg = arg; - if (!set_file_block_mode(fd, false)) { + 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 " << fd; + LOG(ERROR) << "failed to set non-blocking mode for fd " << fde->fd.get(); } - auto pair = g_poll_node_map.emplace(fde->fd.get(), PollNode(fde)); - CHECK(pair.second) << "install existing fd " << fd; + 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; } -fdevent* fdevent_create(int fd, fd_func func, void* arg) { - return fdevent_create_impl(fd, func, arg); -} - -fdevent* fdevent_create(int fd, fd_func2 func, void* arg) { - return fdevent_create_impl(fd, func, arg); -} - -unique_fd fdevent_release(fdevent* fde) { - check_main_thread(); +unique_fd fdevent_context_poll::Destroy(fdevent* fde) { + CheckMainThread(); if (!fde) { return {}; } @@ -168,10 +183,10 @@ unique_fd fdevent_release(fdevent* fde) { unique_fd result = std::move(fde->fd); if (fde->state & FDE_ACTIVE) { - g_poll_node_map.erase(result.get()); + poll_node_map_.erase(result.get()); if (fde->state & FDE_PENDING) { - g_pending_list.remove(fde); + pending_list_.remove(fde); } fde->state = 0; fde->events = 0; @@ -181,14 +196,16 @@ unique_fd fdevent_release(fdevent* fde) { return result; } -void fdevent_destroy(fdevent* fde) { - // Release, and then let unique_fd's destructor cleanup. - fdevent_release(fde); -} +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); -static void fdevent_update(fdevent* fde, unsigned events) { - auto it = g_poll_node_map.find(fde->fd.get()); - CHECK(it != g_poll_node_map.end()); + 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; @@ -202,42 +219,31 @@ static void fdevent_update(fdevent* fde, unsigned events) { node.pollfd.events &= ~POLLOUT; } fde->state = (fde->state & FDE_STATEMASK) | events; -} -void fdevent_set(fdevent* fde, unsigned events) { - check_main_thread(); - events &= FDE_EVENTMASK; - if ((fde->state & FDE_EVENTMASK) == events) { - return; - } - CHECK(fde->state & FDE_ACTIVE); - fdevent_update(fde, 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) { - g_pending_list.remove(fde); + pending_list_.remove(fde); fde->state &= ~FDE_PENDING; } } } -void fdevent_add(fdevent* fde, unsigned events) { - check_main_thread(); - CHECK(!(events & FDE_TIMEOUT)); - fdevent_set(fde, (fde->state & FDE_EVENTMASK) | events); +void fdevent_context_poll::Add(fdevent* fde, unsigned events) { + Set(fde, (fde->state & FDE_EVENTMASK) | events); } -void fdevent_del(fdevent* fde, unsigned events) { - check_main_thread(); +void fdevent_context_poll::Del(fdevent* fde, unsigned events) { CHECK(!(events & FDE_TIMEOUT)); - fdevent_set(fde, (fde->state & FDE_EVENTMASK) & ~events); + Set(fde, (fde->state & FDE_EVENTMASK) & ~events); } -void fdevent_set_timeout(fdevent* fde, std::optional timeout) { - check_main_thread(); +void fdevent_context_poll::SetTimeout(fdevent* fde, + std::optional timeout) { + CheckMainThread(); fde->timeout = timeout; fde->last_active = std::chrono::steady_clock::now(); } @@ -257,12 +263,12 @@ static std::string dump_pollfds(const std::vector& pollfds) { return result; } -static std::optional calculate_timeout() { +static std::optional calculate_timeout(fdevent_context_poll* ctx) { std::optional result = std::nullopt; auto now = std::chrono::steady_clock::now(); - check_main_thread(); + ctx->CheckMainThread(); - for (const auto& [fd, pollnode] : g_poll_node_map) { + for (const auto& [fd, pollnode] : ctx->poll_node_map_) { UNUSED(fd); auto timeout_opt = pollnode.fde->timeout; if (timeout_opt) { @@ -283,15 +289,15 @@ static std::optional calculate_timeout() { return result; } -static void fdevent_process() { +static void fdevent_process(fdevent_context_poll* ctx) { std::vector pollfds; - for (const auto& pair : g_poll_node_map) { + 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(); + auto timeout = calculate_timeout(ctx); int timeout_ms; if (!timeout) { timeout_ms = -1; @@ -328,8 +334,8 @@ static void fdevent_process() { events |= FDE_READ | FDE_ERROR; } #endif - auto it = g_poll_node_map.find(pollfd.fd); - CHECK(it != g_poll_node_map.end()); + auto it = ctx->poll_node_map_.find(pollfd.fd); + CHECK(it != ctx->poll_node_map_.end()); fdevent* fde = it->second.fde; if (events == 0) { @@ -348,7 +354,7 @@ static void fdevent_process() { fde->last_active = post_poll; D("%s got events %x", dump_fde(fde).c_str(), events); fde->state |= FDE_PENDING; - g_pending_list.push_back(fde); + ctx->pending_list_.push_back(fde); } } } @@ -376,27 +382,28 @@ static void fdevent_call_fdfunc(fdevent* fde) { fde->func); } -static void fdevent_run_flush() EXCLUDES(run_queue_mutex) { +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(run_queue_mutex); - if (run_queue.empty()) { + std::lock_guard lock(ctx->run_queue_mutex_); + if (ctx->run_queue_.empty()) { break; } - fn = run_queue.front(); - run_queue.pop_front(); + fn = ctx->run_queue_.front(); + ctx->run_queue_.pop_front(); } fn(); } } -static void fdevent_run_func(int fd, unsigned ev, void* /* userdata */) { +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. @@ -405,13 +412,13 @@ static void fdevent_run_func(int fd, unsigned ev, void* /* userdata */) { } // Mark that we need to flush, and then run it at the end of fdevent_loop. - run_needs_flush = true; + *run_needs_flush = true; } -static void fdevent_run_setup() { +static void fdevent_run_setup(fdevent_context_poll* ctx) { { - std::lock_guard lock(run_queue_mutex); - CHECK(run_queue_notify_fd.get() == -1); + 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"; @@ -421,23 +428,23 @@ static void fdevent_run_setup() { PLOG(FATAL) << "failed to make run queue notify socket nonblocking"; } - run_queue_notify_fd.reset(s[0]); - fdevent* fde = fdevent_create(s[1], fdevent_run_func, nullptr); + 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); - fdevent_add(fde, FDE_READ); + ctx->Add(fde, FDE_READ); } - fdevent_run_flush(); + fdevent_run_flush(ctx); } -void fdevent_run_on_main_thread(std::function fn) { - std::lock_guard lock(run_queue_mutex); - run_queue.push_back(std::move(fn)); +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); + 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) { @@ -448,7 +455,7 @@ void fdevent_run_on_main_thread(std::function fn) { } } -static void fdevent_check_spin(uint64_t cycle) { +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 { @@ -456,6 +463,8 @@ static void fdevent_check_spin(uint64_t cycle) { 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(); @@ -468,7 +477,7 @@ static void fdevent_check_spin(uint64_t cycle) { } last_cycle = now; - for (auto* fde : g_pending_list) { + 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] = @@ -503,51 +512,110 @@ static void fdevent_check_spin(uint64_t cycle) { } } -void fdevent_loop() { - set_main_thread(); - fdevent_run_setup(); +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) { + if (terminate_loop_) { return; } D("--- --- waiting for events"); - fdevent_process(); + fdevent_process(this); - fdevent_check_spin(cycle++); + fdevent_check_spin(this, cycle++); - while (!g_pending_list.empty()) { - fdevent* fde = g_pending_list.front(); - g_pending_list.pop_front(); + while (!pending_list_.empty()) { + fdevent* fde = pending_list_.front(); + pending_list_.pop_front(); fdevent_call_fdfunc(fde); } - if (run_needs_flush) { - fdevent_run_flush(); - run_needs_flush = false; + 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); +} + +fdevent* fdevent_create(int fd, fd_func2 func, void* arg) { + unique_fd ufd(fd); + return fdevent_get_ambient()->Create(std::move(ufd), func, arg); +} + +unique_fd fdevent_release(fdevent* fde) { + return fdevent_get_ambient()->Destroy(fde); +} + +void fdevent_destroy(fdevent* fde) { + fdevent_get_ambient()->Destroy(fde); +} + +void fdevent_set(fdevent* fde, unsigned events) { + fdevent_get_ambient()->Set(fde, events); +} + +void fdevent_add(fdevent* fde, unsigned events) { + fdevent_get_ambient()->Add(fde, events); +} + +void fdevent_del(fdevent* fde, unsigned events) { + fdevent_get_ambient()->Del(fde, events); +} + +void fdevent_set_timeout(fdevent* fde, std::optional timeout) { + fdevent_get_ambient()->SetTimeout(fde, timeout); +} + +void fdevent_run_on_main_thread(std::function fn) { + fdevent_get_ambient()->Run(std::move(fn)); +} + +void fdevent_loop() { + fdevent_get_ambient()->Loop(); +} + +void check_main_thread() { + fdevent_get_ambient()->CheckMainThread(); +} + void fdevent_terminate_loop() { - terminate_loop = true; + fdevent_get_ambient()->TerminateLoop(); } size_t fdevent_installed_count() { - return g_poll_node_map.size(); + return fdevent_get_ambient()->InstalledCount(); } void fdevent_reset() { - g_poll_node_map.clear(); - g_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; + return fdevent_get_ambient()->Reset(); } diff --git a/adb/fdevent/fdevent.h b/adb/fdevent/fdevent.h index 42dbb9e70..a953cc17d 100644 --- a/adb/fdevent/fdevent.h +++ b/adb/fdevent/fdevent.h @@ -36,6 +36,43 @@ typedef void (*fd_func)(int fd, unsigned events, void *userdata); typedef void (*fd_func2)(struct fdevent* fde, unsigned events, void* userdata); +struct fdevent; + +struct fdevent_context { + virtual ~fdevent_context() = default; + + // Allocate and initialize a new fdevent object. + virtual fdevent* Create(unique_fd fd, std::variant func, void* arg) = 0; + + // Deallocate an fdevent object, returning the file descriptor that was owned by it. + virtual unique_fd Destroy(fdevent* fde) = 0; + + // Change which events should cause notifications. + virtual void Set(fdevent* fde, unsigned events) = 0; + virtual void Add(fdevent* fde, unsigned events) = 0; + virtual void Del(fdevent* fde, unsigned events) = 0; + + // Set a timeout on an fdevent. + // If no events are triggered by the timeout, an FDE_TIMEOUT will be generated. + // Note timeouts are not defused automatically; if a timeout is set on an fdevent, it will + // trigger repeatedly every |timeout| ms. + virtual void SetTimeout(fdevent* fde, std::optional timeout) = 0; + + // Loop forever, handling events. + virtual void Loop() = 0; + + // Assert that the caller is running on the context's main thread. + virtual void CheckMainThread() = 0; + + // Queue an operation to be run on the main thread. + virtual void Run(std::function fn) = 0; + + // Test-only functionality: + virtual void TerminateLoop() = 0; + virtual size_t InstalledCount() = 0; + virtual void Reset() = 0; +}; + struct fdevent { uint64_t id; @@ -51,31 +88,18 @@ struct fdevent { void* arg = nullptr; }; -// Allocate and initialize a new fdevent object -// TODO: Switch these to unique_fd. -fdevent *fdevent_create(int fd, fd_func func, void *arg); +// Backwards compatibility shims that forward to the global fdevent_context. +fdevent* fdevent_create(int fd, fd_func func, void* arg); fdevent* fdevent_create(int fd, fd_func2 func, void* arg); -// Deallocate an fdevent object that was created by fdevent_create. -void fdevent_destroy(fdevent *fde); - -// fdevent_destroy, except releasing the file descriptor previously owned by the fdevent. unique_fd fdevent_release(fdevent* fde); +void fdevent_destroy(fdevent* fde); -// Change which events should cause notifications void fdevent_set(fdevent *fde, unsigned events); void fdevent_add(fdevent *fde, unsigned events); void fdevent_del(fdevent *fde, unsigned events); - -// Set a timeout on an fdevent. -// If no events are triggered by the timeout, an FDE_TIMEOUT will be generated. -// Note timeouts are not defused automatically; if a timeout is set on an fdevent, it will -// trigger repeatedly every |timeout| ms. void fdevent_set_timeout(fdevent* fde, std::optional timeout); - -// Loop forever, handling events. void fdevent_loop(); - void check_main_thread(); // Queue an operation to run on the main thread. @@ -85,6 +109,5 @@ void fdevent_run_on_main_thread(std::function fn); void fdevent_terminate_loop(); size_t fdevent_installed_count(); void fdevent_reset(); -void set_main_thread(); #endif From 95068bbd9a1b00020564e6cd32ff9abdd4115402 Mon Sep 17 00:00:00 2001 From: Josh Gao Date: Mon, 8 Jul 2019 15:23:17 -0700 Subject: [PATCH 04/11] adb: move fdevent implementation out to a separate file. Test: mma Change-Id: I8af945b9ae0accac75fa93c538e44cbab357bb5d --- adb/Android.bp | 1 + adb/fdevent/fdevent.cpp | 543 ++--------------------------------- adb/fdevent/fdevent.h | 9 + adb/fdevent/fdevent_poll.cpp | 472 ++++++++++++++++++++++++++++++ adb/fdevent/fdevent_poll.h | 81 ++++++ 5 files changed, 581 insertions(+), 525 deletions(-) create mode 100644 adb/fdevent/fdevent_poll.cpp create mode 100644 adb/fdevent/fdevent_poll.h 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; +}; From 7adca93fe9e7aa4e4c22ab03e39b6bcf8268c840 Mon Sep 17 00:00:00 2001 From: Josh Gao Date: Mon, 8 Jul 2019 17:31:47 -0700 Subject: [PATCH 05/11] adb: implement fdevent_reset by constructing a new context. Test: adb_test Change-Id: Ie67afafe2b73fb2a8fc08568560adac6f456eb9a --- adb/fdevent/fdevent.cpp | 7 ++++--- adb/fdevent/fdevent.h | 1 - adb/fdevent/fdevent_poll.cpp | 12 ------------ adb/fdevent/fdevent_poll.h | 1 - 4 files changed, 4 insertions(+), 17 deletions(-) diff --git a/adb/fdevent/fdevent.cpp b/adb/fdevent/fdevent.cpp index 82bb0a224..e80bb5a51 100644 --- a/adb/fdevent/fdevent.cpp +++ b/adb/fdevent/fdevent.cpp @@ -49,10 +49,11 @@ std::string dump_fde(const fdevent* fde) { state.c_str()); } -static fdevent_context* g_ambient_fdevent_context = new fdevent_context_poll(); +static auto& g_ambient_fdevent_context = + *new std::unique_ptr(new fdevent_context_poll()); static fdevent_context* fdevent_get_ambient() { - return g_ambient_fdevent_context; + return g_ambient_fdevent_context.get(); } fdevent* fdevent_create(int fd, fd_func func, void* arg) { @@ -110,5 +111,5 @@ size_t fdevent_installed_count() { } void fdevent_reset() { - return fdevent_get_ambient()->Reset(); + g_ambient_fdevent_context.reset(new fdevent_context_poll()); } diff --git a/adb/fdevent/fdevent.h b/adb/fdevent/fdevent.h index 7bc50f4b2..b46219c51 100644 --- a/adb/fdevent/fdevent.h +++ b/adb/fdevent/fdevent.h @@ -79,7 +79,6 @@ struct fdevent_context { // Test-only functionality: virtual void TerminateLoop() = 0; virtual size_t InstalledCount() = 0; - virtual void Reset() = 0; }; struct fdevent { diff --git a/adb/fdevent/fdevent_poll.cpp b/adb/fdevent/fdevent_poll.cpp index 63ef3f1fa..6e016f6cc 100644 --- a/adb/fdevent/fdevent_poll.cpp +++ b/adb/fdevent/fdevent_poll.cpp @@ -458,15 +458,3 @@ void fdevent_context_poll::TerminateLoop() { 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 index aa82ee52a..f5720ca3a 100644 --- a/adb/fdevent/fdevent_poll.h +++ b/adb/fdevent/fdevent_poll.h @@ -62,7 +62,6 @@ struct fdevent_context_poll : public fdevent_context { 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. From 95eef6b097df222ccb63c408e7aa87fa27c7953c Mon Sep 17 00:00:00 2001 From: Josh Gao Date: Mon, 8 Jul 2019 17:37:23 -0700 Subject: [PATCH 06/11] adb: fdevent: move run queue to fdevent_context. Make the run queue logic reusable between implementations of fdevent by moving it to the abstract base class. Test: adb_test Change-Id: If2f72e3ddc8007304bca63aa75446fa117267b25 --- adb/fdevent/fdevent.cpp | 26 ++++++++ adb/fdevent/fdevent.h | 19 +++++- adb/fdevent/fdevent_poll.cpp | 123 +++++++++++++---------------------- adb/fdevent/fdevent_poll.h | 17 ++--- adb/fdevent/fdevent_test.h | 4 +- 5 files changed, 99 insertions(+), 90 deletions(-) diff --git a/adb/fdevent/fdevent.cpp b/adb/fdevent/fdevent.cpp index e80bb5a51..c858f6be4 100644 --- a/adb/fdevent/fdevent.cpp +++ b/adb/fdevent/fdevent.cpp @@ -49,6 +49,32 @@ std::string dump_fde(const fdevent* fde) { state.c_str()); } +void fdevent_context::Run(std::function fn) { + { + std::lock_guard lock(run_queue_mutex_); + run_queue_.push_back(std::move(fn)); + } + + Interrupt(); +} + +void fdevent_context::FlushRunQueue() { + // 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(this->run_queue_mutex_); + if (this->run_queue_.empty()) { + break; + } + fn = this->run_queue_.front(); + this->run_queue_.pop_front(); + } + fn(); + } +} + static auto& g_ambient_fdevent_context = *new std::unique_ptr(new fdevent_context_poll()); diff --git a/adb/fdevent/fdevent.h b/adb/fdevent/fdevent.h index b46219c51..5a2f2c63f 100644 --- a/adb/fdevent/fdevent.h +++ b/adb/fdevent/fdevent.h @@ -21,10 +21,14 @@ #include #include +#include #include +#include #include #include +#include + #include "adb_unique_fd.h" // Events that may be observed @@ -48,6 +52,7 @@ struct fdevent; std::string dump_fde(const fdevent* fde); struct fdevent_context { + public: virtual ~fdevent_context() = default; // Allocate and initialize a new fdevent object. @@ -68,17 +73,29 @@ struct fdevent_context { virtual void SetTimeout(fdevent* fde, std::optional timeout) = 0; // Loop forever, handling events. + // Implementations should call FlushRunQueue on every iteration. virtual void Loop() = 0; // Assert that the caller is running on the context's main thread. virtual void CheckMainThread() = 0; // Queue an operation to be run on the main thread. - virtual void Run(std::function fn) = 0; + void Run(std::function fn); // Test-only functionality: virtual void TerminateLoop() = 0; virtual size_t InstalledCount() = 0; + + protected: + // Interrupt the run loop. + virtual void Interrupt() = 0; + + // Run all pending functions enqueued via Run(). + void FlushRunQueue() EXCLUDES(run_queue_mutex_); + + private: + std::mutex run_queue_mutex_; + std::deque> run_queue_ GUARDED_BY(run_queue_mutex_); }; struct fdevent { diff --git a/adb/fdevent/fdevent_poll.cpp b/adb/fdevent/fdevent_poll.cpp index 6e016f6cc..7615859dc 100644 --- a/adb/fdevent/fdevent_poll.cpp +++ b/adb/fdevent/fdevent_poll.cpp @@ -50,6 +50,35 @@ #include "fdevent.h" #include "sysdeps/chrono.h" +static void fdevent_interrupt(int fd, unsigned, void*) { + char buf[BUFSIZ]; + ssize_t rc = TEMP_FAILURE_RETRY(adb_read(fd, buf, sizeof(buf))); + if (rc == -1) { + PLOG(FATAL) << "failed to read from fdevent interrupt fd"; + } +} + +fdevent_context_poll::fdevent_context_poll() { + int s[2]; + if (adb_socketpair(s) != 0) { + PLOG(FATAL) << "failed to create fdevent interrupt socketpair"; + } + + if (!set_file_block_mode(s[0], false) || !set_file_block_mode(s[1], false)) { + PLOG(FATAL) << "failed to make fdevent interrupt socket nonblocking"; + } + + this->interrupt_fd_.reset(s[0]); + fdevent* fde = this->Create(unique_fd(s[1]), fdevent_interrupt, nullptr); + CHECK(fde != nullptr); + this->Add(fde, FDE_READ); +} + +fdevent_context_poll::~fdevent_context_poll() { + main_thread_valid_ = false; + this->Destroy(this->interrupt_fde_); +} + void fdevent_context_poll::CheckMainThread() { if (main_thread_valid_) { CHECK_EQ(main_thread_id_, android::base::GetThreadId()); @@ -291,79 +320,6 @@ static void fdevent_call_fdfunc(fdevent* fde) { 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. @@ -424,7 +380,6 @@ static void fdevent_check_spin(fdevent_context_poll* ctx, uint64_t cycle) { 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) { @@ -444,17 +399,27 @@ void fdevent_context_poll::Loop() { fdevent_call_fdfunc(fde); } - if (run_needs_flush_) { - fdevent_run_flush(this); - run_needs_flush_ = false; - } + this->FlushRunQueue(); } } void fdevent_context_poll::TerminateLoop() { terminate_loop_ = true; + Interrupt(); } size_t fdevent_context_poll::InstalledCount() { - return poll_node_map_.size(); + // We always have an installed fde for interrupt. + return poll_node_map_.size() - 1; +} + +void fdevent_context_poll::Interrupt() { + int rc = adb_write(this->interrupt_fd_, "", 1); + + // It's possible that we get EAGAIN here, if lots of notifications came in while handling. + if (rc == 0) { + PLOG(FATAL) << "fdevent interrupt fd was closed?"; + } else if (rc == -1 && errno != EAGAIN) { + PLOG(FATAL) << "failed to write to fdevent interrupt fd"; + } } diff --git a/adb/fdevent/fdevent_poll.h b/adb/fdevent/fdevent_poll.h index f5720ca3a..1b505a7e0 100644 --- a/adb/fdevent/fdevent_poll.h +++ b/adb/fdevent/fdevent_poll.h @@ -25,6 +25,7 @@ #include +#include "adb_unique_fd.h" #include "fdevent.h" struct PollNode { @@ -44,7 +45,8 @@ struct PollNode { }; struct fdevent_context_poll : public fdevent_context { - virtual ~fdevent_context_poll() = default; + fdevent_context_poll(); + virtual ~fdevent_context_poll(); virtual fdevent* Create(unique_fd fd, std::variant func, void* arg) final; virtual unique_fd Destroy(fdevent* fde) final; @@ -58,11 +60,13 @@ struct fdevent_context_poll : public fdevent_context { virtual void CheckMainThread() final; - virtual void Run(std::function fn) final; - virtual void TerminateLoop() final; virtual size_t InstalledCount() final; + protected: + virtual void Interrupt() final; + + public: // 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_; @@ -71,10 +75,7 @@ struct fdevent_context_poll : public fdevent_context { 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_); - + unique_fd interrupt_fd_; + fdevent* interrupt_fde_ = nullptr; std::atomic terminate_loop_ = false; }; diff --git a/adb/fdevent/fdevent_test.h b/adb/fdevent/fdevent_test.h index 24bce59a7..2139d0f66 100644 --- a/adb/fdevent/fdevent_test.h +++ b/adb/fdevent/fdevent_test.h @@ -78,8 +78,8 @@ class FdeventTest : public ::testing::Test { } size_t GetAdditionalLocalSocketCount() { - // dummy socket installed in PrepareThread() + fdevent_run_on_main_thread socket - return 2; + // dummy socket installed in PrepareThread() + return 1; } void TerminateThread() { From e546f6816e90faa31385ebc9eabbb04d49368bb2 Mon Sep 17 00:00:00 2001 From: Josh Gao Date: Mon, 8 Jul 2019 17:57:30 -0700 Subject: [PATCH 07/11] adb: fdevent: delete FDE_CREATED. All fdevents must be created by fdevent_create/fdevent_context::Create now, so this flag is meaningless. Test: none Change-Id: I8489afad07bdb267ddfdb0dbb25d40d3a62f55f6 --- adb/fdevent/fdevent.cpp | 3 --- adb/fdevent/fdevent.h | 1 - adb/fdevent/fdevent_poll.cpp | 5 ----- 3 files changed, 9 deletions(-) diff --git a/adb/fdevent/fdevent.cpp b/adb/fdevent/fdevent.cpp index c858f6be4..d6e3a660d 100644 --- a/adb/fdevent/fdevent.cpp +++ b/adb/fdevent/fdevent.cpp @@ -33,9 +33,6 @@ std::string dump_fde(const fdevent* fde) { if (fde->state & FDE_PENDING) { state += "P"; } - if (fde->state & FDE_CREATED) { - state += "C"; - } if (fde->state & FDE_READ) { state += "R"; } diff --git a/adb/fdevent/fdevent.h b/adb/fdevent/fdevent.h index 5a2f2c63f..0147ad5d6 100644 --- a/adb/fdevent/fdevent.h +++ b/adb/fdevent/fdevent.h @@ -43,7 +43,6 @@ #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); diff --git a/adb/fdevent/fdevent_poll.cpp b/adb/fdevent/fdevent_poll.cpp index 7615859dc..f7863379c 100644 --- a/adb/fdevent/fdevent_poll.cpp +++ b/adb/fdevent/fdevent_poll.cpp @@ -105,7 +105,6 @@ fdevent* fdevent_context_poll::Create(unique_fd fd, std::variantfd.get(), PollNode(fde)); CHECK(pair.second) << "install existing fd " << fde->fd.get(); - fde->state |= FDE_CREATED; return fde; } @@ -115,10 +114,6 @@ unique_fd fdevent_context_poll::Destroy(fdevent* 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()); From 2c95bf73a574fc7aa9a7df821efbcdedfa0401c4 Mon Sep 17 00:00:00 2001 From: Josh Gao Date: Mon, 8 Jul 2019 18:05:16 -0700 Subject: [PATCH 08/11] adb: fdevent: move CheckMainThread logic to fdevent_context. Test: adb_test Change-Id: I10b93293af22d54ec739dc0f95c95df3ab082fb6 --- adb/fdevent/fdevent.cpp | 8 ++++++++ adb/fdevent/fdevent.h | 7 +++++-- adb/fdevent/fdevent_poll.cpp | 14 ++++---------- adb/fdevent/fdevent_poll.h | 4 ---- 4 files changed, 17 insertions(+), 16 deletions(-) diff --git a/adb/fdevent/fdevent.cpp b/adb/fdevent/fdevent.cpp index d6e3a660d..698b29353 100644 --- a/adb/fdevent/fdevent.cpp +++ b/adb/fdevent/fdevent.cpp @@ -20,7 +20,9 @@ #include +#include #include +#include #include "fdevent.h" #include "fdevent_poll.h" @@ -46,6 +48,12 @@ std::string dump_fde(const fdevent* fde) { state.c_str()); } +void fdevent_context::CheckMainThread() { + if (main_thread_id_) { + CHECK_EQ(*main_thread_id_, android::base::GetThreadId()); + } +} + void fdevent_context::Run(std::function fn) { { std::lock_guard lock(run_queue_mutex_); diff --git a/adb/fdevent/fdevent.h b/adb/fdevent/fdevent.h index 0147ad5d6..c007e518b 100644 --- a/adb/fdevent/fdevent.h +++ b/adb/fdevent/fdevent.h @@ -75,8 +75,9 @@ struct fdevent_context { // Implementations should call FlushRunQueue on every iteration. virtual void Loop() = 0; - // Assert that the caller is running on the context's main thread. - virtual void CheckMainThread() = 0; + // Assert that the caller is either running on the context's main thread, or that there is no + // active main thread. + void CheckMainThread(); // Queue an operation to be run on the main thread. void Run(std::function fn); @@ -92,6 +93,8 @@ struct fdevent_context { // Run all pending functions enqueued via Run(). void FlushRunQueue() EXCLUDES(run_queue_mutex_); + std::optional main_thread_id_ = std::nullopt; + private: std::mutex run_queue_mutex_; std::deque> run_queue_ GUARDED_BY(run_queue_mutex_); diff --git a/adb/fdevent/fdevent_poll.cpp b/adb/fdevent/fdevent_poll.cpp index f7863379c..8a1216283 100644 --- a/adb/fdevent/fdevent_poll.cpp +++ b/adb/fdevent/fdevent_poll.cpp @@ -75,16 +75,9 @@ fdevent_context_poll::fdevent_context_poll() { } fdevent_context_poll::~fdevent_context_poll() { - main_thread_valid_ = false; this->Destroy(this->interrupt_fde_); } -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(); @@ -373,13 +366,12 @@ static void fdevent_check_spin(fdevent_context_poll* ctx, uint64_t cycle) { } void fdevent_context_poll::Loop() { - this->main_thread_id_ = android::base::GetThreadId(); - this->main_thread_valid_ = true; + main_thread_id_ = android::base::GetThreadId(); uint64_t cycle = 0; while (true) { if (terminate_loop_) { - return; + break; } D("--- --- waiting for events"); @@ -396,6 +388,8 @@ void fdevent_context_poll::Loop() { this->FlushRunQueue(); } + + main_thread_id_.reset(); } void fdevent_context_poll::TerminateLoop() { diff --git a/adb/fdevent/fdevent_poll.h b/adb/fdevent/fdevent_poll.h index 1b505a7e0..b0465b959 100644 --- a/adb/fdevent/fdevent_poll.h +++ b/adb/fdevent/fdevent_poll.h @@ -58,8 +58,6 @@ struct fdevent_context_poll : public fdevent_context { virtual void Loop() final; - virtual void CheckMainThread() final; - virtual void TerminateLoop() final; virtual size_t InstalledCount() final; @@ -71,8 +69,6 @@ struct fdevent_context_poll : public fdevent_context { // 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; unique_fd interrupt_fd_; From ebaa348d3268c8d2a8d7f3bb92e234a4693a8de4 Mon Sep 17 00:00:00 2001 From: Josh Gao Date: Mon, 8 Jul 2019 18:08:06 -0700 Subject: [PATCH 09/11] adb: fdevent: move TerminateLoop to fdevent_context. Test: adb_test Change-Id: Ib1f20aefdb36603e2ceac4197c02551f557056ee --- adb/fdevent/fdevent.cpp | 5 +++++ adb/fdevent/fdevent.h | 8 +++++--- adb/fdevent/fdevent_poll.cpp | 5 ----- adb/fdevent/fdevent_poll.h | 2 -- 4 files changed, 10 insertions(+), 10 deletions(-) diff --git a/adb/fdevent/fdevent.cpp b/adb/fdevent/fdevent.cpp index 698b29353..d2f81e004 100644 --- a/adb/fdevent/fdevent.cpp +++ b/adb/fdevent/fdevent.cpp @@ -63,6 +63,11 @@ void fdevent_context::Run(std::function fn) { Interrupt(); } +void fdevent_context::TerminateLoop() { + terminate_loop_ = true; + Interrupt(); +} + void fdevent_context::FlushRunQueue() { // We need to be careful around reentrancy here, since a function we call can queue up another // function. diff --git a/adb/fdevent/fdevent.h b/adb/fdevent/fdevent.h index c007e518b..e84cff799 100644 --- a/adb/fdevent/fdevent.h +++ b/adb/fdevent/fdevent.h @@ -71,8 +71,9 @@ struct fdevent_context { // trigger repeatedly every |timeout| ms. virtual void SetTimeout(fdevent* fde, std::optional timeout) = 0; - // Loop forever, handling events. - // Implementations should call FlushRunQueue on every iteration. + // Loop until TerminateLoop is called, handling events. + // Implementations should call FlushRunQueue on every iteration, and check the value of + // terminate_loop_ to determine whether to stop. virtual void Loop() = 0; // Assert that the caller is either running on the context's main thread, or that there is no @@ -83,7 +84,7 @@ struct fdevent_context { void Run(std::function fn); // Test-only functionality: - virtual void TerminateLoop() = 0; + void TerminateLoop(); virtual size_t InstalledCount() = 0; protected: @@ -94,6 +95,7 @@ struct fdevent_context { void FlushRunQueue() EXCLUDES(run_queue_mutex_); std::optional main_thread_id_ = std::nullopt; + std::atomic terminate_loop_ = false; private: std::mutex run_queue_mutex_; diff --git a/adb/fdevent/fdevent_poll.cpp b/adb/fdevent/fdevent_poll.cpp index 8a1216283..43fd462a0 100644 --- a/adb/fdevent/fdevent_poll.cpp +++ b/adb/fdevent/fdevent_poll.cpp @@ -392,11 +392,6 @@ void fdevent_context_poll::Loop() { main_thread_id_.reset(); } -void fdevent_context_poll::TerminateLoop() { - terminate_loop_ = true; - Interrupt(); -} - size_t fdevent_context_poll::InstalledCount() { // We always have an installed fde for interrupt. return poll_node_map_.size() - 1; diff --git a/adb/fdevent/fdevent_poll.h b/adb/fdevent/fdevent_poll.h index b0465b959..9ae47baaf 100644 --- a/adb/fdevent/fdevent_poll.h +++ b/adb/fdevent/fdevent_poll.h @@ -58,7 +58,6 @@ struct fdevent_context_poll : public fdevent_context { virtual void Loop() final; - virtual void TerminateLoop() final; virtual size_t InstalledCount() final; protected: @@ -73,5 +72,4 @@ struct fdevent_context_poll : public fdevent_context { unique_fd interrupt_fd_; fdevent* interrupt_fde_ = nullptr; - std::atomic terminate_loop_ = false; }; From 33944a27424e1657615c143125d68d21d3506fac Mon Sep 17 00:00:00 2001 From: Josh Gao Date: Mon, 8 Jul 2019 18:16:19 -0700 Subject: [PATCH 10/11] adb: fdevent: extract Create/Destroy from fdevent_context_poll. Test: adb_test Change-Id: Ida308e8e281cbb2954277196a51945f124ce5823 --- adb/fdevent/fdevent.cpp | 35 +++++++++++++++++++++++++++++++++++ adb/fdevent/fdevent.h | 13 +++++++++++-- adb/fdevent/fdevent_poll.cpp | 33 +++------------------------------ adb/fdevent/fdevent_poll.h | 5 ++--- 4 files changed, 51 insertions(+), 35 deletions(-) diff --git a/adb/fdevent/fdevent.cpp b/adb/fdevent/fdevent.cpp index d2f81e004..26ebc49d7 100644 --- a/adb/fdevent/fdevent.cpp +++ b/adb/fdevent/fdevent.cpp @@ -24,6 +24,7 @@ #include #include +#include "adb_utils.h" #include "fdevent.h" #include "fdevent_poll.h" @@ -48,6 +49,40 @@ std::string dump_fde(const fdevent* fde) { state.c_str()); } +fdevent* fdevent_context::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(); + } + + this->Register(fde); + return fde; +} + +unique_fd fdevent_context::Destroy(fdevent* fde) { + CheckMainThread(); + if (!fde) { + return {}; + } + + this->Unregister(fde); + + unique_fd result = std::move(fde->fd); + delete fde; + return result; +} + void fdevent_context::CheckMainThread() { if (main_thread_id_) { CHECK_EQ(*main_thread_id_, android::base::GetThreadId()); diff --git a/adb/fdevent/fdevent.h b/adb/fdevent/fdevent.h index e84cff799..3a3682fcc 100644 --- a/adb/fdevent/fdevent.h +++ b/adb/fdevent/fdevent.h @@ -55,11 +55,19 @@ struct fdevent_context { virtual ~fdevent_context() = default; // Allocate and initialize a new fdevent object. - virtual fdevent* Create(unique_fd fd, std::variant func, void* arg) = 0; + fdevent* Create(unique_fd fd, std::variant func, void* arg); // Deallocate an fdevent object, returning the file descriptor that was owned by it. - virtual unique_fd Destroy(fdevent* fde) = 0; + unique_fd Destroy(fdevent* fde); + protected: + // Register an fdevent that is being created by Create with the fdevent_context. + virtual void Register(fdevent* fde) = 0; + + // Unregister an fdevent that is being destroyed by Destroy with the fdevent_context. + virtual void Unregister(fdevent* fde) = 0; + + public: // Change which events should cause notifications. virtual void Set(fdevent* fde, unsigned events) = 0; virtual void Add(fdevent* fde, unsigned events) = 0; @@ -98,6 +106,7 @@ struct fdevent_context { std::atomic terminate_loop_ = false; private: + uint64_t fdevent_id_ = 0; std::mutex run_queue_mutex_; std::deque> run_queue_ GUARDED_BY(run_queue_mutex_); }; diff --git a/adb/fdevent/fdevent_poll.cpp b/adb/fdevent/fdevent_poll.cpp index 43fd462a0..59b6b933f 100644 --- a/adb/fdevent/fdevent_poll.cpp +++ b/adb/fdevent/fdevent_poll.cpp @@ -78,38 +78,14 @@ fdevent_context_poll::~fdevent_context_poll() { this->Destroy(this->interrupt_fde_); } -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(); - } +void fdevent_context_poll::Register(fdevent* fde) { auto pair = poll_node_map_.emplace(fde->fd.get(), PollNode(fde)); CHECK(pair.second) << "install existing fd " << fde->fd.get(); - - return fde; } -unique_fd fdevent_context_poll::Destroy(fdevent* fde) { - CheckMainThread(); - if (!fde) { - return {}; - } - - unique_fd result = std::move(fde->fd); +void fdevent_context_poll::Unregister(fdevent* fde) { if (fde->state & FDE_ACTIVE) { - poll_node_map_.erase(result.get()); + poll_node_map_.erase(fde->fd.get()); if (fde->state & FDE_PENDING) { pending_list_.remove(fde); @@ -117,9 +93,6 @@ unique_fd fdevent_context_poll::Destroy(fdevent* fde) { fde->state = 0; fde->events = 0; } - - delete fde; - return result; } void fdevent_context_poll::Set(fdevent* fde, unsigned events) { diff --git a/adb/fdevent/fdevent_poll.h b/adb/fdevent/fdevent_poll.h index 9ae47baaf..bffc873c6 100644 --- a/adb/fdevent/fdevent_poll.h +++ b/adb/fdevent/fdevent_poll.h @@ -48,8 +48,8 @@ struct fdevent_context_poll : public fdevent_context { fdevent_context_poll(); virtual ~fdevent_context_poll(); - virtual fdevent* Create(unique_fd fd, std::variant func, void* arg) final; - virtual unique_fd Destroy(fdevent* fde) final; + virtual void Register(fdevent* fde) final; + virtual void Unregister(fdevent* fde) final; virtual void Set(fdevent* fde, unsigned events) final; virtual void Add(fdevent* fde, unsigned events) final; @@ -68,7 +68,6 @@ struct fdevent_context_poll : public fdevent_context { // That's why we don't need a lock for fdevent. std::unordered_map poll_node_map_; std::list pending_list_; - uint64_t fdevent_id_ = 0; unique_fd interrupt_fd_; fdevent* interrupt_fde_ = nullptr; From 35b29360a29a10c71441d550e215984be6b88b65 Mon Sep 17 00:00:00 2001 From: Josh Gao Date: Mon, 8 Jul 2019 18:17:41 -0700 Subject: [PATCH 11/11] adb: fdevent: extract Add/Del/SetTimeout. Test: adb_test Change-Id: Ibed301f8de73289837153fdfed370a681c24ac55 --- adb/fdevent/fdevent.cpp | 15 +++++++++++++++ adb/fdevent/fdevent.h | 6 +++--- adb/fdevent/fdevent_poll.cpp | 16 ---------------- adb/fdevent/fdevent_poll.h | 3 --- 4 files changed, 18 insertions(+), 22 deletions(-) diff --git a/adb/fdevent/fdevent.cpp b/adb/fdevent/fdevent.cpp index 26ebc49d7..28b8f37ce 100644 --- a/adb/fdevent/fdevent.cpp +++ b/adb/fdevent/fdevent.cpp @@ -83,6 +83,21 @@ unique_fd fdevent_context::Destroy(fdevent* fde) { return result; } +void fdevent_context::Add(fdevent* fde, unsigned events) { + Set(fde, (fde->state & FDE_EVENTMASK) | events); +} + +void fdevent_context::Del(fdevent* fde, unsigned events) { + CHECK(!(events & FDE_TIMEOUT)); + Set(fde, (fde->state & FDE_EVENTMASK) & ~events); +} + +void fdevent_context::SetTimeout(fdevent* fde, std::optional timeout) { + CheckMainThread(); + fde->timeout = timeout; + fde->last_active = std::chrono::steady_clock::now(); +} + void fdevent_context::CheckMainThread() { if (main_thread_id_) { CHECK_EQ(*main_thread_id_, android::base::GetThreadId()); diff --git a/adb/fdevent/fdevent.h b/adb/fdevent/fdevent.h index 3a3682fcc..ccb0c9221 100644 --- a/adb/fdevent/fdevent.h +++ b/adb/fdevent/fdevent.h @@ -70,14 +70,14 @@ struct fdevent_context { public: // Change which events should cause notifications. virtual void Set(fdevent* fde, unsigned events) = 0; - virtual void Add(fdevent* fde, unsigned events) = 0; - virtual void Del(fdevent* fde, unsigned events) = 0; + void Add(fdevent* fde, unsigned events); + void Del(fdevent* fde, unsigned events); // Set a timeout on an fdevent. // If no events are triggered by the timeout, an FDE_TIMEOUT will be generated. // Note timeouts are not defused automatically; if a timeout is set on an fdevent, it will // trigger repeatedly every |timeout| ms. - virtual void SetTimeout(fdevent* fde, std::optional timeout) = 0; + void SetTimeout(fdevent* fde, std::optional timeout); // Loop until TerminateLoop is called, handling events. // Implementations should call FlushRunQueue on every iteration, and check the value of diff --git a/adb/fdevent/fdevent_poll.cpp b/adb/fdevent/fdevent_poll.cpp index 59b6b933f..75ea08178 100644 --- a/adb/fdevent/fdevent_poll.cpp +++ b/adb/fdevent/fdevent_poll.cpp @@ -131,22 +131,6 @@ void fdevent_context_poll::Set(fdevent* fde, unsigned events) { } } -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) { diff --git a/adb/fdevent/fdevent_poll.h b/adb/fdevent/fdevent_poll.h index bffc873c6..db083018b 100644 --- a/adb/fdevent/fdevent_poll.h +++ b/adb/fdevent/fdevent_poll.h @@ -52,9 +52,6 @@ struct fdevent_context_poll : public fdevent_context { virtual void Unregister(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;