Project reorganization
This commit is contained in:
parent
d6f95d89fa
commit
a1ec9f731d
GPG key ID:
9B203B252A63EE38
38 changed files with 182 additions and 87 deletions
|
|
@ -1,208 +0,0 @@
|
|||
// SPDX-FileCopyrightText: 2024 Yury Gubich <blue@macaw.me>
|
||||
// SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
#include "loop.h"
|
||||
|
||||
#include <fcntl.h>
|
||||
#include <unistd.h>
|
||||
#include <stdio.h>
|
||||
#include <string.h>
|
||||
#include <csignal>
|
||||
|
||||
Loop::Loop(const Shared::Logger& logger):
|
||||
Shared::Loggable(logger, {"Loop"}),
|
||||
wakePipe(),
|
||||
handlers(),
|
||||
handlersToAdd(),
|
||||
descriptorsToRemove(),
|
||||
mutex(),
|
||||
running(false)
|
||||
{
|
||||
if (pipe(wakePipe) < 0) {
|
||||
fatal(std::string("pipe: ") + strerror(errno));
|
||||
std::exit(1);
|
||||
}
|
||||
|
||||
fcntl(wakePipe[0], F_SETFL, O_NONBLOCK);
|
||||
fcntl(wakePipe[1], F_SETFL, O_NONBLOCK);
|
||||
|
||||
registerInstance(this);
|
||||
}
|
||||
|
||||
Loop::~Loop() {
|
||||
unregisterInstance(this);
|
||||
|
||||
close(wakePipe[0]);
|
||||
close(wakePipe[1]);
|
||||
}
|
||||
|
||||
void Loop::run() {
|
||||
if (running.exchange(true)) {
|
||||
warn("an attempt to run already running event loop, skipping");
|
||||
return;
|
||||
}
|
||||
|
||||
debug("entering the loop");
|
||||
while (running) {
|
||||
syncHandlers();
|
||||
|
||||
fd_set readfds;
|
||||
int maxFD = setFDsAndFindMax(&readfds);
|
||||
|
||||
int result = select(maxFD + 1, &readfds, nullptr, nullptr, nullptr);
|
||||
|
||||
if (result < 0) {
|
||||
if (errno == EINTR) {
|
||||
debug("interrupted");
|
||||
// drain(&readfds); looks like it's not a good idea to drain here
|
||||
continue;
|
||||
};
|
||||
fatal(std::string("select: ") + strerror(errno));
|
||||
running = false;
|
||||
break;
|
||||
}
|
||||
|
||||
drain(&readfds);
|
||||
runCallbacks(&readfds);
|
||||
}
|
||||
|
||||
debug("left the loop");
|
||||
}
|
||||
|
||||
void Loop::stop() {
|
||||
if (!running.exchange(false))
|
||||
return;
|
||||
|
||||
debug("stopping the loop");
|
||||
wake();
|
||||
}
|
||||
|
||||
void Loop::addDescriptor(int descriptor, const Callback& handler) {
|
||||
std::lock_guard lock(mutex);
|
||||
|
||||
if (handlers.count(descriptor) != 0) {
|
||||
warn("an attempt to add descriptor " + std::to_string(descriptor) + " for the second time");
|
||||
return;
|
||||
}
|
||||
|
||||
if (!handlersToAdd.emplace(descriptor, handler).second) {
|
||||
warn("an attempt to add descriptor " + std::to_string(descriptor) + " for the second time");
|
||||
return;
|
||||
}
|
||||
|
||||
wake();
|
||||
}
|
||||
|
||||
void Loop::removeDescriptor(int descriptor) {
|
||||
std::lock_guard lock(mutex);
|
||||
|
||||
if (handlersToAdd.erase(descriptor) != 0)
|
||||
return;
|
||||
|
||||
if (handlers.count(descriptor) == 0) {
|
||||
warn("an attempt to remove unknown descriptor " + std::to_string(descriptor));
|
||||
return;
|
||||
}
|
||||
|
||||
if (!descriptorsToRemove.insert(descriptor).second) {
|
||||
warn("an attempt to remove descriptor " + std::to_string(descriptor) + " for the second time");
|
||||
return;
|
||||
}
|
||||
|
||||
wake();
|
||||
}
|
||||
|
||||
void Loop::wake() {
|
||||
char w = 'w';
|
||||
if (write(wakePipe[1], &w, 1) < 0 && errno != EAGAIN)
|
||||
error("failed to wake up event loop: " + std::string(strerror(errno)));
|
||||
}
|
||||
|
||||
void Loop::drain(fd_set* readfds) {
|
||||
if (!FD_ISSET(wakePipe[0], readfds))
|
||||
return;
|
||||
|
||||
char buf[64];
|
||||
while (true) {
|
||||
ssize_t r = read(wakePipe[0], buf, sizeof(buf));
|
||||
if (r > 0)
|
||||
continue;
|
||||
|
||||
if (r == -1) {
|
||||
if (errno == EAGAIN)
|
||||
break;
|
||||
|
||||
error("failed to drain wake pipe: " + std::string(strerror(errno)));
|
||||
break;
|
||||
}
|
||||
|
||||
if (r == 0)
|
||||
break; // pipe closed
|
||||
}
|
||||
}
|
||||
|
||||
void Loop::runCallbacks(fd_set* readfds) {
|
||||
for (const std::pair<const int, Callback>& pair : handlers) {
|
||||
if (!FD_ISSET(pair.first, readfds))
|
||||
continue;
|
||||
|
||||
try {
|
||||
pair.second();
|
||||
} catch (const std::exception& e) {
|
||||
error("exception in event loop from handler with file descriptor " + std::to_string(pair.first) + ":\n" + e.what());
|
||||
} catch (...) {
|
||||
error("unhandled throw in event loop from handler with file descriptor " + std::to_string(pair.first));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
int Loop::setFDsAndFindMax(fd_set* readfds) {
|
||||
FD_ZERO(readfds);
|
||||
FD_SET(wakePipe[0], readfds);
|
||||
|
||||
int maxFD = wakePipe[0];
|
||||
for (const std::pair<const int, Callback>& pair : handlers) {
|
||||
FD_SET(pair.first, readfds);
|
||||
maxFD = std::max(maxFD, pair.first);
|
||||
}
|
||||
|
||||
return maxFD;
|
||||
}
|
||||
|
||||
void Loop::syncHandlers() {
|
||||
std::lock_guard lock(mutex);
|
||||
|
||||
for (int descriptor : descriptorsToRemove)
|
||||
handlers.erase(descriptor);
|
||||
|
||||
for (const std::pair<const int, Callback>& pair : handlersToAdd)
|
||||
handlers.insert(pair);
|
||||
|
||||
descriptorsToRemove.clear();
|
||||
handlersToAdd.clear();
|
||||
}
|
||||
|
||||
void Loop::registerInstance(Loop* loop) {
|
||||
std::lock_guard lock(instanceMutex);
|
||||
if (instances.size() == 0) {
|
||||
std::signal(SIGINT, signalHandler);
|
||||
std::signal(SIGTERM, signalHandler);
|
||||
}
|
||||
|
||||
instances.insert(loop);
|
||||
}
|
||||
|
||||
void Loop::unregisterInstance(Loop* loop) {
|
||||
std::lock_guard lock(instanceMutex);
|
||||
instances.erase(loop);
|
||||
|
||||
if (instances.size() == 0) {
|
||||
std::signal(SIGINT, SIG_DFL);
|
||||
std::signal(SIGTERM, SIG_DFL);
|
||||
}
|
||||
}
|
||||
|
||||
void Loop::signalHandler(int signum) {
|
||||
for (Loop* loop : instances)
|
||||
loop->stop();
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue