ACPR/android_system_core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

adb: move fdevent implementation out to a separate file.

Browse source 
Test: mma
Change-Id: I8af945b9ae0accac75fa93c538e44cbab357bb5d
This commit is contained in:
Josh Gao 2019-07-08 15:23:17 -07:00
parent c2cf121174
commit 95068bbd9a
5 changed files with 581 additions and 525 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
adb/Android.bp
View file

@ -125,6 +125,7 @@ libadb_srcs = [
"adb_unique_fd.cpp", "adb_unique_fd.cpp",
"adb_utils.cpp", "adb_utils.cpp",
"fdevent/fdevent.cpp", "fdevent/fdevent.cpp",
"fdevent/fdevent_poll.cpp",
"services.cpp", "services.cpp",
"sockets.cpp", "sockets.cpp",
"socket_spec.cpp", "socket_spec.cpp",

543
adb/fdevent/fdevent.cpp
View file

@ -1,64 +1,31 @@
/* http://frotznet.googlecode.com/svn/trunk/utils/fdevent.c /*
** * Copyright 2006, Brian Swetland <swetland@frotz.net>
** Copyright 2006, Brian Swetland <swetland@frotz.net> *
** * Licensed under the Apache License, Version 2.0 (the "License");
** Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License.
** you may not use this file except in compliance with the License. * You may obtain a copy of the License at
** You may obtain a copy of the License at *
** * http://www.apache.org/licenses/LICENSE-2.0
** http://www.apache.org/licenses/LICENSE-2.0 *
** * Unless required by applicable law or agreed to in writing, software
** Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS,
** distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
** WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and
** See the License for the specific language governing permissions and * limitations under the License.
** limitations under the License. */
*/
#define TRACE_TAG FDEVENT #define TRACE_TAG FDEVENT
#include "sysdeps.h" #include "sysdeps.h"
#include "fdevent.h"
#include <fcntl.h>
#include <inttypes.h> #include <inttypes.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <atomic>
#include <deque>
#include <functional>
#include <list>
#include <mutex>
#include <optional>
#include <unordered_map>
#include <utility>
#include <variant>
#include <vector>
#include <android-base/chrono_utils.h>
#include <android-base/file.h>
#include <android-base/logging.h>
#include <android-base/stringprintf.h> #include <android-base/stringprintf.h>
#include <android-base/thread_annotations.h>
#include <android-base/threads.h>
#include "adb_io.h" #include "fdevent.h"
#include "adb_trace.h" #include "fdevent_poll.h"
#include "adb_unique_fd.h"
#include "adb_utils.h"
#include "sysdeps/chrono.h"
#define FDE_EVENTMASK 0x00ff std::string dump_fde(const fdevent* fde) {
#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 state; std::string state;
if (fde->state & FDE_ACTIVE) { if (fde->state & FDE_ACTIVE) {
state += "A"; state += "A";
@ -82,486 +49,12 @@ static std::string dump_fde(const fdevent* fde) {
state.c_str()); 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<fd_func, fd_func2> 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<std::chrono::milliseconds> timeout) final;
virtual void Loop() final;
virtual void CheckMainThread() final;
virtual void Run(std::function<void()> 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<int, PollNode> poll_node_map_;
std::list<fdevent*> 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<std::function<void()>> run_queue_ GUARDED_BY(run_queue_mutex_);
std::atomic<bool> terminate_loop_ = false;
};
static fdevent_context* g_ambient_fdevent_context = new fdevent_context_poll(); static fdevent_context* g_ambient_fdevent_context = new fdevent_context_poll();
static fdevent_context* fdevent_get_ambient() { static fdevent_context* fdevent_get_ambient() {
return g_ambient_fdevent_context; 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<fd_func, fd_func2> 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<std::chrono::milliseconds> timeout) {
CheckMainThread();
fde->timeout = timeout;
fde->last_active = std::chrono::steady_clock::now();
}
static std::string dump_pollfds(const std::vector<adb_pollfd>& 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<std::chrono::milliseconds> calculate_timeout(fdevent_context_poll* ctx) {
std::optional<std::chrono::milliseconds> 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<std::chrono::milliseconds>(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<adb_pollfd> 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 <class T>
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<decltype(f)>;
if constexpr (std::is_same_v<fd_func, F>) {
f(fde->fd.get(), events, fde->arg);
} else if constexpr (std::is_same_v<fd_func2, F>) {
f(fde, events, fde->arg);
} else {
static_assert(always_false<F>::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<void()> fn;
{
std::lock_guard<std::mutex> 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<bool*>(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<std::mutex> 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<void()> fn) {
std::lock_guard<std::mutex> 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<uint64_t, SpinCheck>();
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<std::mutex> 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) { fdevent* fdevent_create(int fd, fd_func func, void* arg) {
unique_fd ufd(fd); unique_fd ufd(fd);
return fdevent_get_ambient()->Create(std::move(ufd), func, arg); return fdevent_get_ambient()->Create(std::move(ufd), func, arg);

9
adb/fdevent/fdevent.h
View file

@ -33,10 +33,19 @@
#define FDE_ERROR 0x0004 #define FDE_ERROR 0x0004
#define FDE_TIMEOUT 0x0008 #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_func)(int fd, unsigned events, void *userdata);
typedef void (*fd_func2)(struct fdevent* fde, unsigned events, void* userdata); typedef void (*fd_func2)(struct fdevent* fde, unsigned events, void* userdata);
struct fdevent; struct fdevent;
std::string dump_fde(const fdevent* fde);
struct fdevent_context { struct fdevent_context {
virtual ~fdevent_context() = default; virtual ~fdevent_context() = default;

472
adb/fdevent/fdevent_poll.cpp Normal file
View file

@ -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 <fcntl.h>
#include <inttypes.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <atomic>
#include <deque>
#include <functional>
#include <list>
#include <mutex>
#include <optional>
#include <unordered_map>
#include <utility>
#include <variant>
#include <vector>
#include <android-base/chrono_utils.h>
#include <android-base/file.h>
#include <android-base/logging.h>
#include <android-base/stringprintf.h>
#include <android-base/threads.h>
#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<fd_func, fd_func2> 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<std::chrono::milliseconds> timeout) {
CheckMainThread();
fde->timeout = timeout;
fde->last_active = std::chrono::steady_clock::now();
}
static std::string dump_pollfds(const std::vector<adb_pollfd>& 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<std::chrono::milliseconds> calculate_timeout(fdevent_context_poll* ctx) {
std::optional<std::chrono::milliseconds> 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<std::chrono::milliseconds>(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<adb_pollfd> 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 <class T>
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<decltype(f)>;
if constexpr (std::is_same_v<fd_func, F>) {
f(fde->fd.get(), events, fde->arg);
} else if constexpr (std::is_same_v<fd_func2, F>) {
f(fde, events, fde->arg);
} else {
static_assert(always_false<F>::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<void()> fn;
{
std::lock_guard<std::mutex> 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<bool*>(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<std::mutex> 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<void()> fn) {
std::lock_guard<std::mutex> 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<uint64_t, SpinCheck>();
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<std::mutex> lock(run_queue_mutex_);
run_queue_notify_fd_.reset();
run_queue_.clear();
main_thread_valid_ = false;
terminate_loop_ = false;
}

81
adb/fdevent/fdevent_poll.h Normal file
View file

@ -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 <deque>
#include <list>
#include <mutex>
#include <unordered_map>
#include <android-base/thread_annotations.h>
#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<fd_func, fd_func2> 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<std::chrono::milliseconds> timeout) final;
virtual void Loop() final;
virtual void CheckMainThread() final;
virtual void Run(std::function<void()> 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<int, PollNode> poll_node_map_;
std::list<fdevent*> 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<std::function<void()>> run_queue_ GUARDED_BY(run_queue_mutex_);
std::atomic<bool> terminate_loop_ = false;
};