adb/fdevent/fdevent.cpp

/* http://frotznet.googlecode.com/svn/trunk/utils/fdevent.c
**
** Copyright 2006, Brian Swetland <swetland@frotz.net>
**
** 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 <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/thread_annotations.h>
#include <android-base/threads.h>

#include "adb_io.h"
#include "adb_trace.h"
#include "adb_unique_fd.h"
#include "adb_utils.h"
#include "sysdeps/chrono.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 state;
    if (fde->state & FDE_ACTIVE) {
        state += "A";
    }
    if (fde->state & FDE_PENDING) {
        state += "P";
    }
    if (fde->state & FDE_CREATED) {
        state += "C";
    }
    if (fde->state & FDE_READ) {
        state += "R";
    }
    if (fde->state & FDE_WRITE) {
        state += "W";
    }
    if (fde->state & FDE_ERROR) {
        state += "E";
    }
    return android::base::StringPrintf("(fdevent %" PRIu64 ": fd %d %s)", fde->id, fde->fd.get(),
                                       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* 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<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) {
    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<std::chrono::milliseconds> timeout) {
    fdevent_get_ambient()->SetTimeout(fde, timeout);
}

void fdevent_run_on_main_thread(std::function<void()> 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() {
    fdevent_get_ambient()->TerminateLoop();
}

size_t fdevent_installed_count() {
    return fdevent_get_ambient()->InstalledCount();
}

void fdevent_reset() {
    return fdevent_get_ambient()->Reset();
}