/*
* Copyright (C) 2017 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 "perfetto/base/unix_task_runner.h"
#include "perfetto/base/build_config.h"
#include <errno.h>
#include <stdlib.h>
#include <unistd.h>
#include <limits>
namespace perfetto {
namespace base {
UnixTaskRunner::UnixTaskRunner() {
AddFileDescriptorWatch(event_.fd(), [] {
// Not reached -- see PostFileDescriptorWatches().
PERFETTO_DFATAL("Should be unreachable.");
});
}
UnixTaskRunner::~UnixTaskRunner() = default;
void UnixTaskRunner::WakeUp() {
event_.Notify();
}
void UnixTaskRunner::Run() {
PERFETTO_DCHECK_THREAD(thread_checker_);
created_thread_id_ = GetThreadId();
quit_ = false;
for (;;) {
int poll_timeout_ms;
{
std::lock_guard<std::mutex> lock(lock_);
if (quit_)
return;
poll_timeout_ms = GetDelayMsToNextTaskLocked();
UpdateWatchTasksLocked();
}
int ret = PERFETTO_EINTR(poll(
&poll_fds_[0], static_cast<nfds_t>(poll_fds_.size()), poll_timeout_ms));
PERFETTO_CHECK(ret >= 0);
// To avoid starvation we always interleave all types of tasks -- immediate,
// delayed and file descriptor watches.
PostFileDescriptorWatches();
RunImmediateAndDelayedTask();
}
}
void UnixTaskRunner::Quit() {
std::lock_guard<std::mutex> lock(lock_);
quit_ = true;
WakeUp();
}
bool UnixTaskRunner::QuitCalled() {
std::lock_guard<std::mutex> lock(lock_);
return quit_;
}
bool UnixTaskRunner::IsIdleForTesting() {
std::lock_guard<std::mutex> lock(lock_);
return immediate_tasks_.empty();
}
void UnixTaskRunner::UpdateWatchTasksLocked() {
PERFETTO_DCHECK_THREAD(thread_checker_);
if (!watch_tasks_changed_)
return;
watch_tasks_changed_ = false;
poll_fds_.clear();
for (auto& it : watch_tasks_) {
it.second.poll_fd_index = poll_fds_.size();
poll_fds_.push_back({it.first, POLLIN | POLLHUP, 0});
}
}
void UnixTaskRunner::RunImmediateAndDelayedTask() {
// If locking overhead becomes an issue, add a separate work queue.
std::function<void()> immediate_task;
std::function<void()> delayed_task;
TimeMillis now = GetWallTimeMs();
{
std::lock_guard<std::mutex> lock(lock_);
if (!immediate_tasks_.empty()) {
immediate_task = std::move(immediate_tasks_.front());
immediate_tasks_.pop_front();
}
if (!delayed_tasks_.empty()) {
auto it = delayed_tasks_.begin();
if (now >= it->first) {
delayed_task = std::move(it->second);
delayed_tasks_.erase(it);
}
}
}
errno = 0;
if (immediate_task)
RunTask(immediate_task);
errno = 0;
if (delayed_task)
RunTask(delayed_task);
}
void UnixTaskRunner::PostFileDescriptorWatches() {
PERFETTO_DCHECK_THREAD(thread_checker_);
for (size_t i = 0; i < poll_fds_.size(); i++) {
if (!(poll_fds_[i].revents & (POLLIN | POLLHUP)))
continue;
poll_fds_[i].revents = 0;
// The wake-up event is handled inline to avoid an infinite recursion of
// posted tasks.
if (poll_fds_[i].fd == event_.fd()) {
event_.Clear();
continue;
}
// Binding to |this| is safe since we are the only object executing the
// task.
PostTask(std::bind(&UnixTaskRunner::RunFileDescriptorWatch, this,
poll_fds_[i].fd));
// Make the fd negative while a posted task is pending. This makes poll(2)
// ignore the fd.
PERFETTO_DCHECK(poll_fds_[i].fd >= 0);
poll_fds_[i].fd = -poll_fds_[i].fd;
}
}
void UnixTaskRunner::RunFileDescriptorWatch(int fd) {
std::function<void()> task;
{
std::lock_guard<std::mutex> lock(lock_);
auto it = watch_tasks_.find(fd);
if (it == watch_tasks_.end())
return;
// Make poll(2) pay attention to the fd again. Since another thread may have
// updated this watch we need to refresh the set first.
UpdateWatchTasksLocked();
size_t fd_index = it->second.poll_fd_index;
PERFETTO_DCHECK(fd_index < poll_fds_.size());
PERFETTO_DCHECK(::abs(poll_fds_[fd_index].fd) == fd);
poll_fds_[fd_index].fd = fd;
task = it->second.callback;
}
errno = 0;
RunTask(task);
}
int UnixTaskRunner::GetDelayMsToNextTaskLocked() const {
PERFETTO_DCHECK_THREAD(thread_checker_);
if (!immediate_tasks_.empty())
return 0;
if (!delayed_tasks_.empty()) {
TimeMillis diff = delayed_tasks_.begin()->first - GetWallTimeMs();
return std::max(0, static_cast<int>(diff.count()));
}
return -1;
}
void UnixTaskRunner::PostTask(std::function<void()> task) {
bool was_empty;
{
std::lock_guard<std::mutex> lock(lock_);
was_empty = immediate_tasks_.empty();
immediate_tasks_.push_back(std::move(task));
}
if (was_empty)
WakeUp();
}
void UnixTaskRunner::PostDelayedTask(std::function<void()> task,
uint32_t delay_ms) {
TimeMillis runtime = GetWallTimeMs() + TimeMillis(delay_ms);
{
std::lock_guard<std::mutex> lock(lock_);
delayed_tasks_.insert(std::make_pair(runtime, std::move(task)));
}
WakeUp();
}
void UnixTaskRunner::AddFileDescriptorWatch(int fd,
std::function<void()> task) {
PERFETTO_DCHECK(fd >= 0);
{
std::lock_guard<std::mutex> lock(lock_);
PERFETTO_DCHECK(!watch_tasks_.count(fd));
watch_tasks_[fd] = {std::move(task), SIZE_MAX};
watch_tasks_changed_ = true;
}
WakeUp();
}
void UnixTaskRunner::RemoveFileDescriptorWatch(int fd) {
PERFETTO_DCHECK(fd >= 0);
{
std::lock_guard<std::mutex> lock(lock_);
PERFETTO_DCHECK(watch_tasks_.count(fd));
watch_tasks_.erase(fd);
watch_tasks_changed_ = true;
}
// No need to schedule a wake-up for this.
}
bool UnixTaskRunner::RunsTasksOnCurrentThread() const {
return GetThreadId() == created_thread_id_;
}
} // namespace base
} // namespace perfetto