wehub-resource-sync
134 lines
3.3 KiB
C++
134 lines
3.3 KiB
C++
// SPDX-License-Identifier: Apache-2.0
|
|
|
|
#include "periodic_event_notifier.h"
|
|
|
|
#include <cerrno>
|
|
#include <cstdio>
|
|
#include <algorithm>
|
|
|
|
namespace lmcache {
|
|
namespace storage_manager {
|
|
|
|
std::mutex PeriodicEventNotifier::create_mutex_;
|
|
std::unique_ptr<PeriodicEventNotifier> PeriodicEventNotifier::instance_;
|
|
|
|
PeriodicEventNotifier::PeriodicEventNotifier(int interval_ms, bool use_eventfd)
|
|
: interval_ms_(std::max(1, interval_ms)), use_eventfd_(use_eventfd) {
|
|
thread_ = std::thread(&PeriodicEventNotifier::thread_func, this);
|
|
}
|
|
|
|
PeriodicEventNotifier::~PeriodicEventNotifier() {
|
|
stop_.store(true, std::memory_order_release);
|
|
cv_.notify_all();
|
|
if (thread_.joinable()) {
|
|
thread_.join();
|
|
}
|
|
}
|
|
|
|
void PeriodicEventNotifier::create(int interval_ms, bool use_eventfd) {
|
|
std::lock_guard<std::mutex> lock(create_mutex_);
|
|
if (instance_) {
|
|
return;
|
|
}
|
|
instance_.reset(new PeriodicEventNotifier(interval_ms, use_eventfd));
|
|
}
|
|
|
|
PeriodicEventNotifier* PeriodicEventNotifier::get() { return instance_.get(); }
|
|
|
|
void PeriodicEventNotifier::shutdown() {
|
|
std::lock_guard<std::mutex> lock(create_mutex_);
|
|
if (!instance_) {
|
|
return;
|
|
}
|
|
instance_->stop_.store(true, std::memory_order_release);
|
|
instance_->cv_.notify_all();
|
|
if (instance_->thread_.joinable()) {
|
|
instance_->thread_.join();
|
|
}
|
|
instance_.reset();
|
|
}
|
|
|
|
void PeriodicEventNotifier::register_fd(int fd) {
|
|
{
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
fds_.insert(fd);
|
|
}
|
|
cv_.notify_one();
|
|
}
|
|
|
|
void PeriodicEventNotifier::unregister_fd(int fd) {
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
fds_.erase(fd);
|
|
}
|
|
|
|
void PeriodicEventNotifier::set_interval_ms(int interval_ms) {
|
|
interval_ms_.store(std::max(1, interval_ms), std::memory_order_release);
|
|
cv_.notify_one();
|
|
}
|
|
|
|
void PeriodicEventNotifier::signal_fd(int fd) {
|
|
if (use_eventfd_) {
|
|
uint64_t one = 1;
|
|
for (;;) {
|
|
ssize_t w = ::write(fd, &one, sizeof(one));
|
|
if (w == static_cast<ssize_t>(sizeof(one))) return;
|
|
if (w < 0) {
|
|
if (errno == EINTR) continue;
|
|
if (errno == EAGAIN || errno == EWOULDBLOCK) return;
|
|
return;
|
|
}
|
|
}
|
|
} else {
|
|
uint8_t one = 1;
|
|
for (;;) {
|
|
ssize_t w = ::write(fd, &one, sizeof(one));
|
|
if (w == 1) return;
|
|
if (w < 0) {
|
|
if (errno == EINTR) continue;
|
|
if (errno == EAGAIN || errno == EWOULDBLOCK) return;
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
void PeriodicEventNotifier::thread_func() {
|
|
while (true) {
|
|
{
|
|
std::unique_lock<std::mutex> lock(mutex_);
|
|
cv_.wait(lock, [this] {
|
|
return stop_.load(std::memory_order_acquire) || !fds_.empty();
|
|
});
|
|
if (stop_.load(std::memory_order_acquire)) {
|
|
return;
|
|
}
|
|
}
|
|
|
|
while (!stop_.load(std::memory_order_acquire)) {
|
|
std::vector<int> snapshot;
|
|
{
|
|
std::lock_guard<std::mutex> lock(mutex_);
|
|
if (fds_.empty()) {
|
|
break;
|
|
}
|
|
snapshot.assign(fds_.begin(), fds_.end());
|
|
}
|
|
|
|
for (int fd : snapshot) {
|
|
signal_fd(fd);
|
|
}
|
|
|
|
{
|
|
std::unique_lock<std::mutex> lock(mutex_);
|
|
cv_.wait_for(lock,
|
|
std::chrono::milliseconds(
|
|
interval_ms_.load(std::memory_order_acquire)),
|
|
[this] { return stop_.load(std::memory_order_acquire); });
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
} // namespace storage_manager
|
|
} // namespace lmcache
|