* Copyright (c) 2024-2025 Huawei Device Co., Ltd.
* 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 "epoll_manager.h"
#include <algorithm>
#include <atomic>
#include <limits>
#include <thread>
#include <vector>
#include <unistd.h>
#include <sys/epoll.h>
#include <sys/timerfd.h>
#include "dfx_define.h"
#include "dfx_log.h"
#ifdef LOG_DOMAIN
#undef LOG_DOMAIN
#define LOG_DOMAIN 0xD002D11
#endif
namespace OHOS {
namespace HiviewDFX {
namespace {
constexpr const char *const EPOLL_MANAGER = "EPOLL_MANAGER";
constexpr uint64_t NS_PER_S = MS_PER_S * US_PER_MS * NS_PER_US;
constexpr uint64_t US_PER_S = MS_PER_S * US_PER_MS;
constexpr uint8_t TIMESTAMP_BITS = 60;
constexpr uint64_t TIMESTAMP_MASK = (static_cast<uint64_t>(1) << TIMESTAMP_BITS) - 1;
constexpr uint8_t SEQUENCE_START_BIT = TIMESTAMP_BITS;
bool SetTimeOptionForTimeFd(int32_t fd, uint64_t delayTimeInNs, uint64_t intervalTimeInNs)
{
if (fd < 0) {
return false;
}
struct itimerspec timeOption{};
timeOption.it_value.tv_sec =
static_cast<decltype(timeOption.it_value.tv_sec)>(delayTimeInNs / NS_PER_S);
timeOption.it_value.tv_nsec =
static_cast<decltype(timeOption.it_value.tv_nsec)>(delayTimeInNs % NS_PER_S);
timeOption.it_interval.tv_sec =
static_cast<decltype(timeOption.it_interval.tv_sec)>(intervalTimeInNs / NS_PER_S);
timeOption.it_interval.tv_nsec =
static_cast<decltype(timeOption.it_interval.tv_nsec)>(intervalTimeInNs % NS_PER_S);
if (timerfd_settime(fd, 0, &timeOption, nullptr) == -1) {
DFXLOGE("%{public}s :: failed to set delay time for fd, errno: %{public}d.", EPOLL_MANAGER, errno);
return false;
}
return true;
}
SmartFd CreateTimeFd()
{
SmartFd timefd{timerfd_create(CLOCK_MONOTONIC, 0)};
if (!timefd) {
DFXLOGE("%{public}s :: failed to create time fd, errno: %{public}d", EPOLL_MANAGER, errno);
}
return timefd;
}
* @brief Combined uint64_t value with dual-field encoding
* @details Bit allocation for the 64-bit unsigned integer:
* - Bits 60-63 (4 high-order bits): Sequence number (range: 0~15, for batch/instance identification)
* - Bits 0-59 (60 low-order bits): microSecond-level timestamp (unsigned 60-bit value)
* - Timestamp range: 0 to ~36500 years (max value: 2^60 - 1 us ≈ 1.15×10^18 us = 36534 years)
* @note To parse:
* - Sequence number = (value >> 60) & 0x0F;
* - Timestamp (us) = value & TIMESTAMP_MASK; (mask for 60 low bits)
*/
inline uint64_t CalculateDelayTaskId(uint64_t executeTimeInMicroSecond)
{
static std::atomic<uint8_t> delaySequence{0};
return (static_cast<uint64_t>(delaySequence.fetch_add(1)) << SEQUENCE_START_BIT) | executeTimeInMicroSecond;
}
inline uint64_t CalculatePeriodicTaskId()
{
static std::atomic<uint64_t> periodicSequence{1};
return periodicSequence.fetch_add(1);
}
}
uint64_t GetMicroSecondsSinceBoot()
{
struct timespec times{};
if (clock_gettime(CLOCK_BOOTTIME, ×) == -1) {
DFXLOGE("%{public}s :: failed get time for %{public}d", EPOLL_MANAGER, errno);
return 0;
}
return static_cast<uint64_t>(times.tv_sec * US_PER_S + times.tv_nsec / NS_PER_US);
}
EpollListener::EpollListener(SmartFd fd, int64_t timeoutInMs) : fd_(std::move(fd))
{
if (timeoutInMs >= 0) {
timeoutTime_ = static_cast<int64_t>(GetMicroSecondsSinceBoot()) + timeoutInMs * static_cast<int64_t>(US_PER_MS);
} else {
timeoutTime_ = std::numeric_limits<int64_t>::max();
}
}
int64_t EpollListener::GetTimeOutTime() const
{
return timeoutTime_;
}
int32_t EpollListener::GetFd() const
{
return fd_.GetFd();
}
EpollManager &EpollManager::GetInstance()
{
static thread_local EpollManager mainEpollManager;
return mainEpollManager;
}
EpollManager::~EpollManager()
{
StopEpoll();
}
bool EpollManager::AddEpollEvent(EpollListener& epollListener) const
{
if (!eventFd_) {
return false;
}
epoll_event ev{};
ev.events = EPOLLIN;
ev.data.fd = epollListener.GetFd();
if (epoll_ctl(eventFd_.GetFd(), EPOLL_CTL_ADD, ev.data.fd, &ev) < 0) {
DFXLOGE("%s :: Failed to epoll ctl add fd %{public}d, errno %{public}d",
EPOLL_MANAGER, epollListener.GetFd(), errno);
return false;
}
return true;
}
bool EpollManager::DelEpollEvent(int32_t fd) const
{
if (!eventFd_) {
return false;
}
epoll_event ev{};
ev.events = EPOLLIN;
ev.data.fd = fd;
if (epoll_ctl(eventFd_.GetFd(), EPOLL_CTL_DEL, fd, &ev) < 0) {
DFXLOGW("%s :: Failed to epoll ctl delete Fd %{public}d, errno %{public}d", EPOLL_MANAGER, fd, errno);
return false;
}
return true;
}
bool EpollManager::AddListener(std::unique_ptr<EpollListener> epollListener)
{
if (!epollListener || epollListener->GetFd() < 0 || !AddEpollEvent(*epollListener)) {
return false;
}
auto timeoutTime = epollListener->GetTimeOutTime();
auto iter = std::find_if(listeners_.begin(), listeners_.end(),
[timeoutTime](const std::unique_ptr<EpollListener>& listener) {
return listener->GetTimeOutTime() > timeoutTime;
});
listeners_.insert(iter, std::move(epollListener));
return true;
}
bool EpollManager::RemoveListener(int32_t fd)
{
if (fd < 0 || !DelEpollEvent(fd)) {
return false;
}
listeners_.remove_if([fd](const std::unique_ptr<EpollListener>& epollLister) {
return epollLister->GetFd() == fd;
});
return true;
}
EpollListener* EpollManager::GetTargetListener(int32_t fd) const
{
auto iter = std::find_if(listeners_.begin(), listeners_.end(),
[fd](const std::unique_ptr<EpollListener>& listener) {
return listener->GetFd() == fd;
});
return iter == listeners_.end() ? nullptr : iter->get();
}
int32_t EpollManager::GetNextWaitTime() const
{
if (listeners_.empty() || listeners_.front()->GetTimeOutTime() == std::numeric_limits<int64_t>::max()) {
return -1;
}
auto nextWaitTime = listeners_.front()->GetTimeOutTime() - static_cast<int64_t>(GetMicroSecondsSinceBoot());
if (nextWaitTime < 0) {
return 0;
}
constexpr auto minCheckTime = 30 * 1000;
return std::min(static_cast<int32_t>(nextWaitTime / US_PER_MS), minCheckTime);
}
void EpollManager::HandleTimeOut()
{
if (listeners_.empty()) {
return;
}
auto listener = listeners_.front().get();
if (static_cast<uint64_t>(listener->GetTimeOutTime()) <= GetMicroSecondsSinceBoot()) {
listener->OnTimeOut();
RemoveListener(listener->GetFd());
}
}
bool EpollManager::Init(int maxPollEvent)
{
eventFd_ = SmartFd{epoll_create(maxPollEvent)};
if (!eventFd_) {
DFXLOGE("%s :: Failed to create eventFd.", EPOLL_MANAGER);
return false;
}
return true;
}
void EpollManager::StartEpoll(int maxConnection)
{
std::vector<epoll_event> events(maxConnection);
while (eventFd_) {
int32_t timeOut = GetNextWaitTime();
int epollNum = OHOS_TEMP_FAILURE_RETRY(epoll_wait(eventFd_.GetFd(), events.data(), maxConnection, timeOut));
if (epollNum < 0 || !eventFd_) {
continue;
}
if (epollNum == 0) {
HandleTimeOut();
continue;
}
for (int i = 0; i < epollNum; i++) {
if (!(events[i].events & EPOLLIN)) {
DFXLOGE("%{public}s :: client fd %{public}d disconnected", EPOLL_MANAGER, events[i].data.fd);
RemoveListener(events[i].data.fd);
continue;
}
const auto listener = GetTargetListener(events[i].data.fd);
if (listener == nullptr) {
DelEpollEvent(events[i].data.fd);
continue;
}
EventResult result = listener->OnEventPoll();
if (result == EventResult::REMOVE) {
RemoveListener(events[i].data.fd);
}
}
}
}
void EpollManager::StopEpoll()
{
if (eventFd_) {
for (const auto& listener : listeners_) {
(void)DelEpollEvent(listener->GetFd());
}
eventFd_.Reset();
}
}
TimerTask::TimerTask() : EpollListener(CreateTimeFd()) {}
bool TimerTask::SetTimeOption(int32_t delayTimeInS, int32_t intervalTimeInS)
{
if (delayTimeInS < 0 || intervalTimeInS < 0) {
return false;
}
return SetTimeOptionForTimeFd(GetFd(), static_cast<uint64_t>(delayTimeInS) * NS_PER_S,
static_cast<uint64_t>(intervalTimeInS) * NS_PER_S);
}
EventResult TimerTask::OnEventPoll()
{
uint64_t exp = 0;
auto ret = OHOS_TEMP_FAILURE_RETRY(read(GetFd(), &exp, sizeof(exp)));
if (ret < 0 || static_cast<uint64_t>(ret) != sizeof(exp)) {
DFXLOGE("%{public}s :: failed read time fd %{public}" PRId32, EPOLL_MANAGER, GetFd());
return EventResult::REMOVE;
}
return OnTimer() ? EventResult::KEEP : EventResult::REMOVE;
}
std::unique_ptr<TimerTask> TimerTaskAdapter::CreateInstance(std::function<void()> workFunc,
int32_t delayTimeInS, int32_t intervalTimeInS)
{
if (!workFunc) {
return nullptr;
}
auto task = std::unique_ptr<TimerTaskAdapter>(new (std::nothrow)TimerTaskAdapter(workFunc, intervalTimeInS > 0));
if (task == nullptr || !task->SetTimeOption(delayTimeInS, intervalTimeInS)) {
return nullptr;
}
return task;
}
TimerTaskAdapter::TimerTaskAdapter(std::function<void()>& workFunc, bool isIntervalTask)
: work_(std::move(workFunc)), isIntervalTask_(isIntervalTask) {}
bool TimerTaskAdapter::OnTimer()
{
work_();
return isIntervalTask_;
}
DelayTaskQueue::~DelayTaskQueue()
{
if (executor_ != nullptr) {
EpollManager::GetInstance().RemoveListener(executor_->GetFd());
}
}
DelayTaskQueue& DelayTaskQueue::GetInstance()
{
static thread_local DelayTaskQueue queue;
return queue;
}
bool DelayTaskQueue::InitExecutor(uint32_t delayTimeInS)
{
auto executor = std::unique_ptr<Executor>(new(std::nothrow) Executor(*this));
if (executor == nullptr) {
return false;
}
if (!SetTimeOptionForTimeFd(executor->GetFd(), delayTimeInS * NS_PER_S, NS_PER_S)) {
return false;
}
executor_ = executor.get();
if (!EpollManager::GetInstance().AddListener(std::move(executor))) {
return false;
}
return true;
}
uint64_t DelayTaskQueue::AddDelayTask(std::function<void()> workFunc, uint32_t delayTimeInS)
{
if (delayTimeInS == 0 || !workFunc) {
return 0;
}
if (executor_ == nullptr && !InitExecutor(delayTimeInS)) {
return 0;
}
const auto delayTimeInMicroSeconds = static_cast<uint64_t>(delayTimeInS) * MS_PER_S * US_PER_MS;
if (GetMicroSecondsSinceBoot() == 0) {
return 0;
}
const auto executeTimeInMicroSecond = GetMicroSecondsSinceBoot() + delayTimeInMicroSeconds;
auto insertPos = std::find_if(delayTasks_.begin(), delayTasks_.end(),
[executeTimeInMicroSecond](const std::pair<uint64_t, std::function<void()>>& existingTask) {
return (existingTask.first & TIMESTAMP_MASK) > executeTimeInMicroSecond;
});
if (insertPos == delayTasks_.begin()) {
SetTimeOptionForTimeFd(executor_->GetFd(), delayTimeInMicroSeconds * NS_PER_US, NS_PER_S);
}
auto delayTaskId = CalculateDelayTaskId(executeTimeInMicroSecond);
delayTasks_.insert(insertPos, std::make_pair(delayTaskId, std::move(workFunc)));
return delayTaskId;
}
bool DelayTaskQueue::RemoveDelayTask(uint64_t delayTaskId)
{
auto it = std::find_if(delayTasks_.begin(), delayTasks_.end(),
[&](const std::pair<uint64_t, std::function<void()>>& item) {
return item.first == delayTaskId;
});
if (it == delayTasks_.end()) {
return false;
}
bool isBegin = (it == delayTasks_.begin());
delayTasks_.erase(it);
if (isBegin && executor_ != nullptr && delayTasks_.empty()) {
SetTimeOptionForTimeFd(executor_->GetFd(), 1, NS_PER_S);
}
return true;
}
DelayTaskQueue::Executor::~Executor()
{
* Set the executor_ in the outer class to null to indicate that the existing executor has been destroyed.
*/
delayTaskQueue_.executor_ = nullptr;
}
bool DelayTaskQueue::Executor::OnTimer()
{
while (!delayTaskQueue_.delayTasks_.empty()) {
auto currentTimeInMicroSecond = GetMicroSecondsSinceBoot();
auto delayTaskId = delayTaskQueue_.delayTasks_.front().first;
auto firstTaskExecuteTime = delayTaskId & TIMESTAMP_MASK;
if (firstTaskExecuteTime > currentTimeInMicroSecond) {
auto nextTaskDelayTime = firstTaskExecuteTime - currentTimeInMicroSecond;
SetTimeOptionForTimeFd(GetFd(), nextTaskDelayTime * NS_PER_US, NS_PER_S);
return true;
}
delayTaskQueue_.delayTasks_.front().second();
delayTaskQueue_.RemoveDelayTask(delayTaskId);
}
return false;
}
PeriodicTaskQueue::~PeriodicTaskQueue()
{
if (currentDelayTaskId_ != 0) {
DelayTaskQueue::GetInstance().RemoveDelayTask(currentDelayTaskId_);
}
}
PeriodicTaskQueue& PeriodicTaskQueue::GetInstance()
{
static thread_local PeriodicTaskQueue queue;
return queue;
}
void PeriodicTaskQueue::Execute()
{
auto currentTime = GetMicroSecondsSinceBoot();
while (!periodicTasks_.empty()) {
auto& frontTask = periodicTasks_.front();
if (std::get<0>(frontTask) > currentTime) {
break;
}
uint32_t interval = std::get<2>(frontTask);
auto task = std::get<3>(frontTask);
if (task && task()) {
auto nextTime = currentTime + static_cast<uint64_t>(interval) * MS_PER_S * US_PER_MS;
std::get<0>(frontTask) = nextTime;
auto it = periodicTasks_.begin();
auto newPos = std::find_if(periodicTasks_.begin(), periodicTasks_.end(),
[nextTime](const std::tuple<uint64_t, uint64_t, uint32_t, std::function<bool()>>& item) {
return std::get<0>(item) > nextTime;
});
periodicTasks_.splice(newPos, periodicTasks_, it);
} else {
periodicTasks_.pop_front();
}
}
ScheduleNextTask(currentTime);
}
void PeriodicTaskQueue::ScheduleNextTask(uint64_t currentTime)
{
if (periodicTasks_.empty()) {
return;
}
auto nextExecuteTime = std::get<0>(periodicTasks_.front());
uint64_t delayTime = 0;
if (nextExecuteTime > currentTime) {
delayTime = (nextExecuteTime - currentTime + MS_PER_S * US_PER_MS - 1) / (MS_PER_S * US_PER_MS);
}
if (delayTime == 0) {
delayTime++;
}
currentDelayTaskId_ = DelayTaskQueue::GetInstance().AddDelayTask(
[this]() { Execute(); }, delayTime);
}
uint64_t PeriodicTaskQueue::AddPeriodicTask(std::function<bool()> workFunc, uint32_t intervalTimeInS)
{
if (!workFunc || intervalTimeInS == 0) {
return 0;
}
auto taskId = CalculatePeriodicTaskId();
uint64_t currentTime = GetMicroSecondsSinceBoot();
auto nextExecuteTime = currentTime + static_cast<uint64_t>(intervalTimeInS) * MS_PER_S * US_PER_MS;
auto insertPos = std::find_if(periodicTasks_.begin(), periodicTasks_.end(),
[nextExecuteTime](const std::tuple<uint64_t, uint64_t, uint32_t, std::function<bool()>>& item) {
return std::get<0>(item) > nextExecuteTime;
});
auto isBegin = insertPos == periodicTasks_.begin();
periodicTasks_.insert(insertPos, std::make_tuple(nextExecuteTime, taskId, intervalTimeInS, std::move(workFunc)));
if (isBegin) {
ScheduleNextTask(currentTime);
}
return taskId;
}
bool PeriodicTaskQueue::RemovePeriodicTask(uint64_t taskId)
{
auto it = std::find_if(periodicTasks_.begin(), periodicTasks_.end(),
[taskId](const std::tuple<uint64_t, uint64_t, uint32_t, std::function<bool()>>& item) {
return std::get<1>(item) == taskId;
});
if (it == periodicTasks_.end()) {
return false;
}
bool isBegin = (it == periodicTasks_.begin());
periodicTasks_.erase(it);
if (isBegin) {
auto currentTime = GetMicroSecondsSinceBoot();
ScheduleNextTask(currentTime);
}
return true;
}
}
}