system/serenity
system/serenity
1
0
Fork 0
mirror of https://github.com/SerenityOS/serenity synced 2024-07-22 10:36:24 +00:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

LibCore: Big first step towards pluggable Core::EventLoop

Browse source 
The EventLoop is now a wrapper around an EventLoopImplementation.
Our old EventLoop code has moved into EventLoopImplementationUnix and
continues to work as before.

The main difference is that all the separate thread_local variables have
been collected into a file-local ThreadData data structure.

The goal here is to allow running Core::EventLoop with a totally
different backend, such as Qt for Ladybird.
This commit is contained in:
Andreas Kling 2023-04-24 12:25:14 +02:00
parent 3de8dd921e
commit 16c47ccff6
11 changed files with 718 additions and 562 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Userland/Libraries/LibConfig/Client.cpp
View file

@ -14,7 +14,6 @@ static RefPtr<Client> s_the = nullptr;
Client& Client::the()
{
if (!s_the || !s_the->is_open()) {
VERIFY(Core::EventLoop::has_been_instantiated());
s_the = Client::try_create().release_value_but_fixme_should_propagate_errors();
}
return *s_the;

2
Userland/Libraries/LibCore/CMakeLists.txt
View file

@ -11,6 +11,8 @@ set(SOURCES
ElapsedTimer.cpp
Event.cpp
EventLoop.cpp
EventLoopImplementation.cpp
EventLoopImplementationUnix.cpp
File.cpp
IODevice.cpp
LockFile.cpp

588
Userland/Libraries/LibCore/EventLoop.cpp
View file

@ -6,252 +6,76 @@
* SPDX-License-Identifier: BSD-2-Clause
*/
#include <AK/Assertions.h>
#include <AK/Badge.h>
#include <AK/Debug.h>
#include <AK/Format.h>
#include <AK/IDAllocator.h>
#include <AK/JsonObject.h>
#include <AK/JsonValue.h>
#include <AK/NeverDestroyed.h>
#include <AK/Singleton.h>
#include <AK/TemporaryChange.h>
#include <AK/Time.h>
#include <LibCore/Event.h>
#include <LibCore/EventLoop.h>
#include <LibCore/LocalServer.h>
#include <LibCore/Notifier.h>
#include <LibCore/EventLoopImplementationUnix.h>
#include <LibCore/Object.h>
#include <LibCore/Promise.h>
#include <LibCore/SessionManagement.h>
#include <LibCore/Socket.h>
#include <LibCore/ThreadEventQueue.h>
#include <LibThreading/Mutex.h>
#include <LibThreading/MutexProtected.h>
#include <errno.h>
#include <fcntl.h>
#include <signal.h>
#include <stdio.h>
#include <string.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <sys/types.h>
#include <time.h>
#include <unistd.h>
#ifdef AK_OS_SERENITY
# include <LibCore/Account.h>
extern bool s_global_initializers_ran;
#endif
namespace Core {
struct EventLoopTimer {
int timer_id { 0 };
Time interval;
Time fire_time;
bool should_reload { false };
TimerShouldFireWhenNotVisible fire_when_not_visible { TimerShouldFireWhenNotVisible::No };
WeakPtr<Object> owner;
void reload(Time const& now);
bool has_expired(Time const& now) const;
};
struct EventLoop::Private {
ThreadEventQueue& thread_event_queue;
Private()
: thread_event_queue(ThreadEventQueue::current())
{
}
};
static Threading::MutexProtected<NeverDestroyed<IDAllocator>> s_id_allocator;
// Each thread has its own event loop stack, its own timers, notifiers and a wake pipe.
static thread_local Vector<EventLoop&>* s_event_loop_stack;
static thread_local HashMap<int, NonnullOwnPtr<EventLoopTimer>>* s_timers;
static thread_local HashTable<Notifier*>* s_notifiers;
// The wake pipe is both responsible for notifying us when someone calls wake(), as well as POSIX signals.
// While wake() pushes zero into the pipe, signal numbers (by defintion nonzero, see signal_numbers.h) are pushed into the pipe verbatim.
thread_local int EventLoop::s_wake_pipe_fds[2];
thread_local bool EventLoop::s_wake_pipe_initialized { false };
void EventLoop::initialize_wake_pipes()
namespace {
thread_local Vector<EventLoop&>* s_event_loop_stack;
Vector<EventLoop&>& event_loop_stack()
{
if (!s_wake_pipe_initialized) {
#if defined(SOCK_NONBLOCK)
int rc = pipe2(s_wake_pipe_fds, O_CLOEXEC);
#else
int rc = pipe(s_wake_pipe_fds);
fcntl(s_wake_pipe_fds[0], F_SETFD, FD_CLOEXEC);
fcntl(s_wake_pipe_fds[1], F_SETFD, FD_CLOEXEC);
#endif
VERIFY(rc == 0);
s_wake_pipe_initialized = true;
}
if (!s_event_loop_stack)
s_event_loop_stack = new Vector<EventLoop&>;
return *s_event_loop_stack;
}
bool EventLoop::has_been_instantiated()
bool has_event_loop()
{
return s_event_loop_stack != nullptr && !s_event_loop_stack->is_empty();
return !event_loop_stack().is_empty();
}
class SignalHandlers : public RefCounted<SignalHandlers> {
AK_MAKE_NONCOPYABLE(SignalHandlers);
AK_MAKE_NONMOVABLE(SignalHandlers);
public:
SignalHandlers(int signo, void (*handle_signal)(int));
~SignalHandlers();
void dispatch();
int add(Function<void(int)>&& handler);
bool remove(int handler_id);
bool is_empty() const
{
if (m_calling_handlers) {
for (auto& handler : m_handlers_pending) {
if (handler.value)
return false; // an add is pending
}
}
return m_handlers.is_empty();
}
bool have(int handler_id) const
{
if (m_calling_handlers) {
auto it = m_handlers_pending.find(handler_id);
if (it != m_handlers_pending.end()) {
if (!it->value)
return false; // a deletion is pending
}
}
return m_handlers.contains(handler_id);
}
int m_signo;
void (*m_original_handler)(int); // TODO: can't use sighandler_t?
HashMap<int, Function<void(int)>> m_handlers;
HashMap<int, Function<void(int)>> m_handlers_pending;
bool m_calling_handlers { false };
};
struct SignalHandlersInfo {
HashMap<int, NonnullRefPtr<SignalHandlers>> signal_handlers;
int next_signal_id { 0 };
};
static Singleton<SignalHandlersInfo> s_signals;
template<bool create_if_null = true>
inline SignalHandlersInfo* signals_info()
{
return s_signals.ptr();
}
pid_t EventLoop::s_pid;
EventLoop::EventLoop()
: m_wake_pipe_fds(&s_wake_pipe_fds)
, m_private(make<Private>())
: m_impl(make<EventLoopImplementationUnix>())
{
#ifdef AK_OS_SERENITY
if (!s_global_initializers_ran) {
// NOTE: Trying to have an event loop as a global variable will lead to initialization-order fiascos,
// as the event loop constructor accesses and/or sets other global variables.
// Therefore, we crash the program before ASAN catches us.
// If you came here because of the assertion failure, please redesign your program to not have global event loops.
// The common practice is to initialize the main event loop in the main function, and if necessary,
// pass event loop references around or access them with EventLoop::with_main_locked() and EventLoop::current().
VERIFY_NOT_REACHED();
if (event_loop_stack().is_empty()) {
event_loop_stack().append(*this);
}
#endif
if (!s_event_loop_stack) {
s_event_loop_stack = new Vector<EventLoop&>;
s_timers = new HashMap<int, NonnullOwnPtr<EventLoopTimer>>;
s_notifiers = new HashTable<Notifier*>;
}
if (s_event_loop_stack->is_empty()) {
s_pid = getpid();
s_event_loop_stack->append(*this);
}
initialize_wake_pipes();
dbgln_if(EVENTLOOP_DEBUG, "{} Core::EventLoop constructed :)", getpid());
}
EventLoop::~EventLoop()
{
if (!s_event_loop_stack->is_empty() && &s_event_loop_stack->last() == this)
s_event_loop_stack->take_last();
if (!event_loop_stack().is_empty() && &event_loop_stack().last() == this) {
event_loop_stack().take_last();
}
}
#define VERIFY_EVENT_LOOP_INITIALIZED() \
do { \
if (!s_event_loop_stack) { \
warnln("EventLoop static API was called without prior EventLoop init!"); \
VERIFY_NOT_REACHED(); \
} \
} while (0)
EventLoop& EventLoop::current()
{
VERIFY_EVENT_LOOP_INITIALIZED();
return s_event_loop_stack->last();
return event_loop_stack().last();
}
void EventLoop::quit(int code)
{
dbgln_if(EVENTLOOP_DEBUG, "Core::EventLoop::quit({})", code);
m_exit_requested = true;
m_exit_code = code;
m_impl->quit(code);
}
void EventLoop::unquit()
{
dbgln_if(EVENTLOOP_DEBUG, "Core::EventLoop::unquit()");
m_exit_requested = false;
m_exit_code = 0;
m_impl->unquit();
}
struct EventLoopPusher {
public:
EventLoopPusher(EventLoop& event_loop)
: m_event_loop(event_loop)
{
if (EventLoop::has_been_instantiated()) {
s_event_loop_stack->append(event_loop);
}
event_loop_stack().append(event_loop);
}
~EventLoopPusher()
{
if (EventLoop::has_been_instantiated()) {
s_event_loop_stack->take_last();
}
event_loop_stack().take_last();
}
private:
EventLoop& m_event_loop;
};
int EventLoop::exec()
{
EventLoopPusher pusher(*this);
for (;;) {
if (m_exit_requested)
return m_exit_code;
pump();
}
VERIFY_NOT_REACHED();
return m_impl->exec();
}
void EventLoop::spin_until(Function<bool()> goal_condition)
@ -263,16 +87,12 @@ void EventLoop::spin_until(Function<bool()> goal_condition)
size_t EventLoop::pump(WaitMode mode)
{
// Pumping the event loop from another thread is not allowed.
VERIFY(&m_private->thread_event_queue == &ThreadEventQueue::current());
wait_for_event(mode);
return m_private->thread_event_queue.process();
return m_impl->pump(mode == WaitMode::WaitForEvents ? EventLoopImplementation::PumpMode::WaitForEvents : EventLoopImplementation::PumpMode::DontWaitForEvents);
}
void EventLoop::post_event(Object& receiver, NonnullOwnPtr<Event>&& event)
{
m_private->thread_event_queue.post_event(receiver, move(event));
m_impl->post_event(receiver, move(event));
}
void EventLoop::add_job(NonnullRefPtr<Promise<NonnullRefPtr<Object>>> job_promise)
@ -280,373 +100,56 @@ void EventLoop::add_job(NonnullRefPtr<Promise<NonnullRefPtr<Object>>> job_promis
ThreadEventQueue::current().add_job(move(job_promise));
}
SignalHandlers::SignalHandlers(int signo, void (*handle_signal)(int))
: m_signo(signo)
, m_original_handler(signal(signo, handle_signal))
int EventLoop::register_signal(int signal_number, Function<void(int)> handler)
{
dbgln_if(EVENTLOOP_DEBUG, "Core::EventLoop: Registered handler for signal {}", m_signo);
}
SignalHandlers::~SignalHandlers()
{
dbgln_if(EVENTLOOP_DEBUG, "Core::EventLoop: Unregistering handler for signal {}", m_signo);
signal(m_signo, m_original_handler);
}
void SignalHandlers::dispatch()
{
TemporaryChange change(m_calling_handlers, true);
for (auto& handler : m_handlers)
handler.value(m_signo);
if (!m_handlers_pending.is_empty()) {
// Apply pending adds/removes
for (auto& handler : m_handlers_pending) {
if (handler.value) {
auto result = m_handlers.set(handler.key, move(handler.value));
VERIFY(result == AK::HashSetResult::InsertedNewEntry);
} else {
m_handlers.remove(handler.key);
}
}
m_handlers_pending.clear();
}
}
int SignalHandlers::add(Function<void(int)>&& handler)
{
int id = ++signals_info()->next_signal_id; // TODO: worry about wrapping and duplicates?
if (m_calling_handlers)
m_handlers_pending.set(id, move(handler));
else
m_handlers.set(id, move(handler));
return id;
}
bool SignalHandlers::remove(int handler_id)
{
VERIFY(handler_id != 0);
if (m_calling_handlers) {
auto it = m_handlers.find(handler_id);
if (it != m_handlers.end()) {
// Mark pending remove
m_handlers_pending.set(handler_id, {});
return true;
}
it = m_handlers_pending.find(handler_id);
if (it != m_handlers_pending.end()) {
if (!it->value)
return false; // already was marked as deleted
it->value = nullptr;
return true;
}
return false;
}
return m_handlers.remove(handler_id);
}
void EventLoop::dispatch_signal(int signo)
{
auto& info = *signals_info();
auto handlers = info.signal_handlers.find(signo);
if (handlers != info.signal_handlers.end()) {
// Make sure we bump the ref count while dispatching the handlers!
// This allows a handler to unregister/register while the handlers
// are being called!
auto handler = handlers->value;
dbgln_if(EVENTLOOP_DEBUG, "Core::EventLoop: dispatching signal {}", signo);
handler->dispatch();
}
}
void EventLoop::handle_signal(int signo)
{
VERIFY(signo != 0);
// We MUST check if the current pid still matches, because there
// is a window between fork() and exec() where a signal delivered
// to our fork could be inadvertently routed to the parent process!
if (getpid() == s_pid) {
int nwritten = write(s_wake_pipe_fds[1], &signo, sizeof(signo));
if (nwritten < 0) {
perror("EventLoop::register_signal: write");
VERIFY_NOT_REACHED();
}
} else {
// We're a fork who received a signal, reset s_pid
s_pid = 0;
}
}
int EventLoop::register_signal(int signo, Function<void(int)> handler)
{
VERIFY(signo != 0);
auto& info = *signals_info();
auto handlers = info.signal_handlers.find(signo);
if (handlers == info.signal_handlers.end()) {
auto signal_handlers = adopt_ref(*new SignalHandlers(signo, EventLoop::handle_signal));
auto handler_id = signal_handlers->add(move(handler));
info.signal_handlers.set(signo, move(signal_handlers));
return handler_id;
} else {
return handlers->value->add(move(handler));
}
if (!has_event_loop())
return 0;
return current().m_impl->register_signal(signal_number, move(handler));
}
void EventLoop::unregister_signal(int handler_id)
{
VERIFY(handler_id != 0);
int remove_signo = 0;
auto& info = *signals_info();
for (auto& h : info.signal_handlers) {
auto& handlers = *h.value;
if (handlers.remove(handler_id)) {
if (handlers.is_empty())
remove_signo = handlers.m_signo;
break;
}
}
if (remove_signo != 0)
info.signal_handlers.remove(remove_signo);
}
void EventLoop::notify_forked(ForkEvent event)
{
VERIFY_EVENT_LOOP_INITIALIZED();
switch (event) {
case ForkEvent::Child:
s_event_loop_stack->clear();
s_timers->clear();
s_notifiers->clear();
s_wake_pipe_initialized = false;
initialize_wake_pipes();
if (auto* info = signals_info<false>()) {
info->signal_handlers.clear();
info->next_signal_id = 0;
}
s_pid = 0;
if (!has_event_loop())
return;
}
VERIFY_NOT_REACHED();
current().m_impl->unregister_signal(handler_id);
}
void EventLoop::wait_for_event(WaitMode mode)
void EventLoop::notify_forked(ForkEvent)
{
fd_set rfds;
fd_set wfds;
retry:
// Set up the file descriptors for select().
// Basically, we translate high-level event information into low-level selectable file descriptors.
FD_ZERO(&rfds);
FD_ZERO(&wfds);
int max_fd = 0;
auto add_fd_to_set = [&max_fd](int fd, fd_set& set) {
FD_SET(fd, &set);
if (fd > max_fd)
max_fd = fd;
};
int max_fd_added = -1;
// The wake pipe informs us of POSIX signals as well as manual calls to wake()
add_fd_to_set(s_wake_pipe_fds[0], rfds);
max_fd = max(max_fd, max_fd_added);
for (auto& notifier : *s_notifiers) {
if (notifier->type() == Notifier::Type::Read)
add_fd_to_set(notifier->fd(), rfds);
if (notifier->type() == Notifier::Type::Write)
add_fd_to_set(notifier->fd(), wfds);
if (notifier->type() == Notifier::Type::Exceptional)
VERIFY_NOT_REACHED();
}
bool has_pending_events = m_private->thread_event_queue.has_pending_events();
// Figure out how long to wait at maximum.
// This mainly depends on the WaitMode and whether we have pending events, but also the next expiring timer.
Time now;
struct timeval timeout = { 0, 0 };
bool should_wait_forever = false;
if (mode == WaitMode::WaitForEvents && !has_pending_events) {
auto next_timer_expiration = get_next_timer_expiration();
if (next_timer_expiration.has_value()) {
now = Time::now_monotonic_coarse();
auto computed_timeout = next_timer_expiration.value() - now;
if (computed_timeout.is_negative())
computed_timeout = Time::zero();
timeout = computed_timeout.to_timeval();
} else {
should_wait_forever = true;
}
}
try_select_again:
// select() and wait for file system events, calls to wake(), POSIX signals, or timer expirations.
int marked_fd_count = select(max_fd + 1, &rfds, &wfds, nullptr, should_wait_forever ? nullptr : &timeout);
// Because POSIX, we might spuriously return from select() with EINTR; just select again.
if (marked_fd_count < 0) {
int saved_errno = errno;
if (saved_errno == EINTR) {
if (m_exit_requested)
return;
goto try_select_again;
}
dbgln("Core::EventLoop::wait_for_event: {} ({}: {})", marked_fd_count, saved_errno, strerror(saved_errno));
VERIFY_NOT_REACHED();
}
// We woke up due to a call to wake() or a POSIX signal.
// Handle signals and see whether we need to handle events as well.
if (FD_ISSET(s_wake_pipe_fds[0], &rfds)) {
int wake_events[8];
ssize_t nread;
// We might receive another signal while read()ing here. The signal will go to the handle_signal properly,
// but we get interrupted. Therefore, just retry while we were interrupted.
do {
errno = 0;
nread = read(s_wake_pipe_fds[0], wake_events, sizeof(wake_events));
if (nread == 0)
break;
} while (nread < 0 && errno == EINTR);
if (nread < 0) {
perror("Core::EventLoop::wait_for_event: read from wake pipe");
VERIFY_NOT_REACHED();
}
VERIFY(nread > 0);
bool wake_requested = false;
int event_count = nread / sizeof(wake_events[0]);
for (int i = 0; i < event_count; i++) {
if (wake_events[i] != 0)
dispatch_signal(wake_events[i]);
else
wake_requested = true;
}
if (!wake_requested && nread == sizeof(wake_events))
goto retry;
}
if (!s_timers->is_empty()) {
now = Time::now_monotonic_coarse();
}
// Handle expired timers.
for (auto& it : *s_timers) {
auto& timer = *it.value;
if (!timer.has_expired(now))
continue;
auto owner = timer.owner.strong_ref();
if (timer.fire_when_not_visible == TimerShouldFireWhenNotVisible::No
&& owner && !owner->is_visible_for_timer_purposes()) {
continue;
}
dbgln_if(EVENTLOOP_DEBUG, "Core::EventLoop: Timer {} has expired, sending Core::TimerEvent to {}", timer.timer_id, *owner);
if (owner)
post_event(*owner, make<TimerEvent>(timer.timer_id));
if (timer.should_reload) {
timer.reload(now);
} else {
// FIXME: Support removing expired timers that don't want to reload.
VERIFY_NOT_REACHED();
}
}
if (!marked_fd_count)
return;
// Handle file system notifiers by making them normal events.
for (auto& notifier : *s_notifiers) {
if (notifier->type() == Notifier::Type::Read && FD_ISSET(notifier->fd(), &rfds)) {
post_event(*notifier, make<NotifierActivationEvent>(notifier->fd()));
}
if (notifier->type() == Notifier::Type::Write && FD_ISSET(notifier->fd(), &wfds)) {
post_event(*notifier, make<NotifierActivationEvent>(notifier->fd()));
}
}
}
bool EventLoopTimer::has_expired(Time const& now) const
{
return now > fire_time;
}
void EventLoopTimer::reload(Time const& now)
{
fire_time = now + interval;
}
Optional<Time> EventLoop::get_next_timer_expiration()
{
auto now = Time::now_monotonic_coarse();
Optional<Time> soonest {};
for (auto& it : *s_timers) {
auto& fire_time = it.value->fire_time;
auto owner = it.value->owner.strong_ref();
if (it.value->fire_when_not_visible == TimerShouldFireWhenNotVisible::No
&& owner && !owner->is_visible_for_timer_purposes()) {
continue;
}
// OPTIMIZATION: If we have a timer that needs to fire right away, we can stop looking here.
// FIXME: This whole operation could be O(1) with a better data structure.
if (fire_time < now)
return now;
if (!soonest.has_value() || fire_time < soonest.value())
soonest = fire_time;
}
return soonest;
current().m_impl->notify_forked_and_in_child();
}
int EventLoop::register_timer(Object& object, int milliseconds, bool should_reload, TimerShouldFireWhenNotVisible fire_when_not_visible)
{
VERIFY_EVENT_LOOP_INITIALIZED();
VERIFY(milliseconds >= 0);
auto timer = make<EventLoopTimer>();
timer->owner = object;
timer->interval = Time::from_milliseconds(milliseconds);
timer->reload(Time::now_monotonic_coarse());
timer->should_reload = should_reload;
timer->fire_when_not_visible = fire_when_not_visible;
int timer_id = s_id_allocator.with_locked([](auto& allocator) { return allocator->allocate(); });
timer->timer_id = timer_id;
s_timers->set(timer_id, move(timer));
return timer_id;
if (!has_event_loop())
return 0;
return current().m_impl->register_timer(object, milliseconds, should_reload, fire_when_not_visible);
}
bool EventLoop::unregister_timer(int timer_id)
{
VERIFY_EVENT_LOOP_INITIALIZED();
s_id_allocator.with_locked([&](auto& allocator) { allocator->deallocate(timer_id); });
auto it = s_timers->find(timer_id);
if (it == s_timers->end())
if (!has_event_loop())
return false;
s_timers->remove(it);
return true;
return current().m_impl->unregister_timer(timer_id);
}
void EventLoop::register_notifier(Badge<Notifier>, Notifier& notifier)
{
VERIFY_EVENT_LOOP_INITIALIZED();
s_notifiers->set(&notifier);
if (!has_event_loop())
return;
current().m_impl->register_notifier(notifier);
}
void EventLoop::unregister_notifier(Badge<Notifier>, Notifier& notifier)
{
VERIFY_EVENT_LOOP_INITIALIZED();
s_notifiers->remove(&notifier);
if (!has_event_loop())
return;
current().m_impl->unregister_notifier(notifier);
}
void EventLoop::wake()
{
dbgln_if(EVENTLOOP_DEBUG, "Core::EventLoop::wake()");
int wake_event = 0;
int nwritten = write((*m_wake_pipe_fds)[1], &wake_event, sizeof(wake_event));
if (nwritten < 0) {
perror("EventLoop::wake: write");
VERIFY_NOT_REACHED();
}
m_impl->wake();
}
void EventLoop::deferred_invoke(Function<void()> invokee)
@ -660,4 +163,9 @@ void deferred_invoke(Function<void()> invokee)
EventLoop::current().deferred_invoke(move(invokee));
}
bool EventLoop::was_exit_requested() const
{
return m_impl->was_exit_requested();
}
}

19
Userland/Libraries/LibCore/EventLoop.h
View file

@ -18,6 +18,8 @@
namespace Core {
class EventLoopImplementation;
// The event loop enables asynchronous (not parallel or multi-threaded) computing by efficiently handling events from various sources.
// Event loops are most important for GUI programs, where the various GUI updates and action callbacks run on the EventLoop,
// as well as services, where asynchronous remote procedure calls of multiple clients are handled.
@ -48,9 +50,6 @@ public:
EventLoop();
~EventLoop();
static void initialize_wake_pipes();
static bool has_been_instantiated();
// Pump the event loop until its exit is requested.
int exec();
@ -73,7 +72,7 @@ public:
void quit(int);
void unquit();
bool was_exit_requested() const { return m_exit_requested; }
bool was_exit_requested() const;
// The registration functions act upon the current loop of the current thread.
static int register_timer(Object&, int milliseconds, bool should_reload, TimerShouldFireWhenNotVisible);
@ -102,17 +101,7 @@ private:
static pid_t s_pid;
bool m_exit_requested { false };
int m_exit_code { 0 };
static thread_local int s_wake_pipe_fds[2];
static thread_local bool s_wake_pipe_initialized;
// The wake pipe of this event loop needs to be accessible from other threads.
int (*m_wake_pipe_fds)[2];
struct Private;
NonnullOwnPtr<Private> m_private;
NonnullOwnPtr<EventLoopImplementation> m_impl;
};
void deferred_invoke(Function<void()>);

30
Userland/Libraries/LibCore/EventLoopImplementation.cpp Normal file
View file

@ -0,0 +1,30 @@
/*
* Copyright (c) 2023, Andreas Kling <kling@serenityos.org>
*
* SPDX-License-Identifier: BSD-2-Clause
*/
#include <AK/NonnullOwnPtr.h>
#include <LibCore/Event.h>
#include <LibCore/EventLoopImplementation.h>
#include <LibCore/ThreadEventQueue.h>
namespace Core {
EventLoopImplementation::EventLoopImplementation()
: m_thread_event_queue(ThreadEventQueue::current())
{
}
EventLoopImplementation::~EventLoopImplementation() = default;
void EventLoopImplementation::post_event(Object& receiver, NonnullOwnPtr<Event>&& event)
{
m_thread_event_queue.post_event(receiver, move(event));
// Wake up this EventLoopImplementation if this is a cross-thread event posting.
if (&ThreadEventQueue::current() != &m_thread_event_queue)
wake();
}
}

53
Userland/Libraries/LibCore/EventLoopImplementation.h Normal file
View file

@ -0,0 +1,53 @@
/*
* Copyright (c) 2022-2023, Andreas Kling <kling@serenityos.org>
*
* SPDX-License-Identifier: BSD-2-Clause
*/
#pragma once
#include <AK/Function.h>
#include <LibCore/Forward.h>
namespace Core {
class ThreadEventQueue;
class EventLoopImplementation {
public:
virtual ~EventLoopImplementation();
enum class PumpMode {
WaitForEvents,
DontWaitForEvents,
};
void post_event(Object& receiver, NonnullOwnPtr<Event>&&);
virtual int exec() = 0;
virtual size_t pump(PumpMode) = 0;
virtual void quit(int) = 0;
virtual void wake() = 0;
virtual void deferred_invoke(Function<void()>) = 0;
virtual int register_timer(Object&, int milliseconds, bool should_reload, TimerShouldFireWhenNotVisible) = 0;
virtual bool unregister_timer(int timer_id) = 0;
virtual void register_notifier(Notifier&) = 0;
virtual void unregister_notifier(Notifier&) = 0;
// FIXME: These APIs only exist for obscure use-cases inside SerenityOS. Try to get rid of them.
virtual void unquit() = 0;
virtual bool was_exit_requested() const = 0;
virtual void notify_forked_and_in_child() = 0;
virtual int register_signal(int signal_number, Function<void(int)> handler) = 0;
virtual void unregister_signal(int handler_id) = 0;
protected:
EventLoopImplementation();
ThreadEventQueue& m_thread_event_queue;
};
}

525
Userland/Libraries/LibCore/EventLoopImplementationUnix.cpp Normal file
View file

@ -0,0 +1,525 @@
/*
* Copyright (c) 2023, Andreas Kling <kling@serenityos.org>
*
* SPDX-License-Identifier: BSD-2-Clause
*/
#include <AK/IDAllocator.h>
#include <AK/Singleton.h>
#include <AK/TemporaryChange.h>
#include <AK/Time.h>
#include <AK/WeakPtr.h>
#include <LibCore/Event.h>
#include <LibCore/EventLoopImplementationUnix.h>
#include <LibCore/Notifier.h>
#include <LibCore/Object.h>
#include <LibCore/Socket.h>
#include <LibCore/System.h>
#include <LibCore/ThreadEventQueue.h>
#include <sys/select.h>
#include <unistd.h>
namespace Core {
struct ThreadData;
namespace {
thread_local ThreadData* s_thread_data;
}
struct EventLoopTimer {
int timer_id { 0 };
Time interval;
Time fire_time;
bool should_reload { false };
TimerShouldFireWhenNotVisible fire_when_not_visible { TimerShouldFireWhenNotVisible::No };
WeakPtr<Object> owner;
void reload(Time const& now) { fire_time = now + interval; }
bool has_expired(Time const& now) const { return now > fire_time; }
};
struct ThreadData {
static ThreadData& the()
{
if (!s_thread_data) {
// FIXME: Don't leak this.
s_thread_data = new ThreadData;
}
return *s_thread_data;
}
ThreadData()
{
pid = getpid();
initialize_wake_pipe();
}
void initialize_wake_pipe()
{
if (wake_pipe_fds[0] != -1)
close(wake_pipe_fds[0]);
if (wake_pipe_fds[1] != -1)
close(wake_pipe_fds[1]);
#if defined(SOCK_NONBLOCK)
int rc = pipe2(wake_pipe_fds, O_CLOEXEC);
#else
int rc = pipe(wake_pipe_fds);
fcntl(wake_pipe_fds[0], F_SETFD, FD_CLOEXEC);
fcntl(wake_pipe_fds[1], F_SETFD, FD_CLOEXEC);
#endif
VERIFY(rc == 0);
}
// Each thread has its own timers, notifiers and a wake pipe.
HashMap<int, NonnullOwnPtr<EventLoopTimer>> timers;
HashTable<Notifier*> notifiers;
// The wake pipe is used to notify another event loop that someone has called wake(), or a signal has been received.
// wake() writes 0i32 into the pipe, signals write the signal number (guaranteed non-zero).
int wake_pipe_fds[2] { -1, -1 };
pid_t pid { 0 };
IDAllocator id_allocator;
};
EventLoopImplementationUnix::EventLoopImplementationUnix()
: m_wake_pipe_fds(&ThreadData::the().wake_pipe_fds)
{
}
EventLoopImplementationUnix::~EventLoopImplementationUnix() = default;
int EventLoopImplementationUnix::exec()
{
for (;;) {
if (m_exit_requested)
return m_exit_code;
pump(PumpMode::WaitForEvents);
}
VERIFY_NOT_REACHED();
}
size_t EventLoopImplementationUnix::pump(PumpMode mode)
{
// We can only pump the event loop of the current thread.
VERIFY(&m_thread_event_queue == &ThreadEventQueue::current());
wait_for_events(mode);
return m_thread_event_queue.process();
}
void EventLoopImplementationUnix::quit(int code)
{
m_exit_requested = true;
m_exit_code = code;
}
void EventLoopImplementationUnix::unquit()
{
m_exit_requested = false;
m_exit_code = 0;
}
bool EventLoopImplementationUnix::was_exit_requested() const
{
return m_exit_requested;
}
void EventLoopImplementationUnix::wake()
{
int wake_event = 0;
MUST(Core::System::write((*m_wake_pipe_fds)[1], { &wake_event, sizeof(wake_event) }));
}
void EventLoopImplementationUnix::deferred_invoke(Function<void()> invokee)
{
// FIXME: Get rid of the useless DeferredInvocationContext object.
auto context = DeferredInvocationContext::construct();
post_event(context, make<DeferredInvocationEvent>(context, move(invokee)));
}
void EventLoopImplementationUnix::wait_for_events(PumpMode mode)
{
auto& thread_data = ThreadData::the();
fd_set read_fds {};
fd_set write_fds {};
retry:
int max_fd = 0;
auto add_fd_to_set = [&max_fd](int fd, fd_set& set) {
FD_SET(fd, &set);
if (fd > max_fd)
max_fd = fd;
};
int max_fd_added = -1;
// The wake pipe informs us of POSIX signals as well as manual calls to wake()
add_fd_to_set(thread_data.wake_pipe_fds[0], read_fds);
max_fd = max(max_fd, max_fd_added);
for (auto& notifier : thread_data.notifiers) {
if (notifier->type() == Notifier::Type::Read)
add_fd_to_set(notifier->fd(), read_fds);
if (notifier->type() == Notifier::Type::Write)
add_fd_to_set(notifier->fd(), write_fds);
if (notifier->type() == Notifier::Type::Exceptional)
TODO();
}
bool has_pending_events = ThreadEventQueue::current().has_pending_events();
// Figure out how long to wait at maximum.
// This mainly depends on the PumpMode and whether we have pending events, but also the next expiring timer.
Time now;
struct timeval timeout = { 0, 0 };
bool should_wait_forever = false;
if (mode == PumpMode::WaitForEvents && !has_pending_events) {
auto next_timer_expiration = get_next_timer_expiration();
if (next_timer_expiration.has_value()) {
now = Time::now_monotonic_coarse();
auto computed_timeout = next_timer_expiration.value() - now;
if (computed_timeout.is_negative())
computed_timeout = Time::zero();
timeout = computed_timeout.to_timeval();
} else {
should_wait_forever = true;
}
}
try_select_again:
// select() and wait for file system events, calls to wake(), POSIX signals, or timer expirations.
int marked_fd_count = select(max_fd + 1, &read_fds, &write_fds, nullptr, should_wait_forever ? nullptr : &timeout);
// Because POSIX, we might spuriously return from select() with EINTR; just select again.
if (marked_fd_count < 0) {
int saved_errno = errno;
if (saved_errno == EINTR) {
if (m_exit_requested)
return;
goto try_select_again;
}
dbgln("EventLoopImplementationUnix::wait_for_events: {} ({}: {})", marked_fd_count, saved_errno, strerror(saved_errno));
VERIFY_NOT_REACHED();
}
// We woke up due to a call to wake() or a POSIX signal.
// Handle signals and see whether we need to handle events as well.
if (FD_ISSET(thread_data.wake_pipe_fds[0], &read_fds)) {
int wake_events[8];
ssize_t nread;
// We might receive another signal while read()ing here. The signal will go to the handle_signal properly,
// but we get interrupted. Therefore, just retry while we were interrupted.
do {
errno = 0;
nread = read(thread_data.wake_pipe_fds[0], wake_events, sizeof(wake_events));
if (nread == 0)
break;
} while (nread < 0 && errno == EINTR);
if (nread < 0) {
perror("EventLoopImplementationUnix::wait_for_events: read from wake pipe");
VERIFY_NOT_REACHED();
}
VERIFY(nread > 0);
bool wake_requested = false;
int event_count = nread / sizeof(wake_events[0]);
for (int i = 0; i < event_count; i++) {
if (wake_events[i] != 0)
dispatch_signal(wake_events[i]);
else
wake_requested = true;
}
if (!wake_requested && nread == sizeof(wake_events))
goto retry;
}
if (!thread_data.timers.is_empty()) {
now = Time::now_monotonic_coarse();
}
// Handle expired timers.
for (auto& it : thread_data.timers) {
auto& timer = *it.value;
if (!timer.has_expired(now))
continue;
auto owner = timer.owner.strong_ref();
if (timer.fire_when_not_visible == TimerShouldFireWhenNotVisible::No
&& owner && !owner->is_visible_for_timer_purposes()) {
continue;
}
if (owner)
ThreadEventQueue::current().post_event(*owner, make<TimerEvent>(timer.timer_id));
if (timer.should_reload) {
timer.reload(now);
} else {
// FIXME: Support removing expired timers that don't want to reload.
VERIFY_NOT_REACHED();
}
}
if (!marked_fd_count)
return;
// Handle file system notifiers by making them normal events.
for (auto& notifier : thread_data.notifiers) {
if (notifier->type() == Notifier::Type::Read && FD_ISSET(notifier->fd(), &read_fds)) {
ThreadEventQueue::current().post_event(*notifier, make<NotifierActivationEvent>(notifier->fd()));
}
if (notifier->type() == Notifier::Type::Write && FD_ISSET(notifier->fd(), &write_fds)) {
ThreadEventQueue::current().post_event(*notifier, make<NotifierActivationEvent>(notifier->fd()));
}
}
}
class SignalHandlers : public RefCounted<SignalHandlers> {
AK_MAKE_NONCOPYABLE(SignalHandlers);
AK_MAKE_NONMOVABLE(SignalHandlers);
public:
SignalHandlers(int signal_number, void (*handle_signal)(int));
~SignalHandlers();
void dispatch();
int add(Function<void(int)>&& handler);
bool remove(int handler_id);
bool is_empty() const
{
if (m_calling_handlers) {
for (auto& handler : m_handlers_pending) {
if (handler.value)
return false; // an add is pending
}
}
return m_handlers.is_empty();
}
bool have(int handler_id) const
{
if (m_calling_handlers) {
auto it = m_handlers_pending.find(handler_id);
if (it != m_handlers_pending.end()) {
if (!it->value)
return false; // a deletion is pending
}
}
return m_handlers.contains(handler_id);
}
int m_signal_number;
void (*m_original_handler)(int); // TODO: can't use sighandler_t?
HashMap<int, Function<void(int)>> m_handlers;
HashMap<int, Function<void(int)>> m_handlers_pending;
bool m_calling_handlers { false };
};
struct SignalHandlersInfo {
HashMap<int, NonnullRefPtr<SignalHandlers>> signal_handlers;
int next_signal_id { 0 };
};
static Singleton<SignalHandlersInfo> s_signals;
template<bool create_if_null = true>
inline SignalHandlersInfo* signals_info()
{
return s_signals.ptr();
}
void EventLoopImplementationUnix::dispatch_signal(int signal_number)
{
auto& info = *signals_info();
auto handlers = info.signal_handlers.find(signal_number);
if (handlers != info.signal_handlers.end()) {
// Make sure we bump the ref count while dispatching the handlers!
// This allows a handler to unregister/register while the handlers
// are being called!
auto handler = handlers->value;
handler->dispatch();
}
}
void EventLoopImplementationUnix::notify_forked_and_in_child()
{
auto& thread_data = ThreadData::the();
thread_data.timers.clear();
thread_data.notifiers.clear();
thread_data.initialize_wake_pipe();
if (auto* info = signals_info<false>()) {
info->signal_handlers.clear();
info->next_signal_id = 0;
}
thread_data.pid = getpid();
}
Optional<Time> EventLoopImplementationUnix::get_next_timer_expiration()
{
auto now = Time::now_monotonic_coarse();
Optional<Time> soonest {};
for (auto& it : ThreadData::the().timers) {
auto& fire_time = it.value->fire_time;
auto owner = it.value->owner.strong_ref();
if (it.value->fire_when_not_visible == TimerShouldFireWhenNotVisible::No
&& owner && !owner->is_visible_for_timer_purposes()) {
continue;
}
// OPTIMIZATION: If we have a timer that needs to fire right away, we can stop looking here.
// FIXME: This whole operation could be O(1) with a better data structure.
if (fire_time < now)
return now;
if (!soonest.has_value() || fire_time < soonest.value())
soonest = fire_time;
}
return soonest;
}
SignalHandlers::SignalHandlers(int signal_number, void (*handle_signal)(int))
: m_signal_number(signal_number)
, m_original_handler(signal(signal_number, handle_signal))
{
}
SignalHandlers::~SignalHandlers()
{
signal(m_signal_number, m_original_handler);
}
void SignalHandlers::dispatch()
{
TemporaryChange change(m_calling_handlers, true);
for (auto& handler : m_handlers)
handler.value(m_signal_number);
if (!m_handlers_pending.is_empty()) {
// Apply pending adds/removes
for (auto& handler : m_handlers_pending) {
if (handler.value) {
auto result = m_handlers.set(handler.key, move(handler.value));
VERIFY(result == AK::HashSetResult::InsertedNewEntry);
} else {
m_handlers.remove(handler.key);
}
}
m_handlers_pending.clear();
}
}
int SignalHandlers::add(Function<void(int)>&& handler)
{
int id = ++signals_info()->next_signal_id; // TODO: worry about wrapping and duplicates?
if (m_calling_handlers)
m_handlers_pending.set(id, move(handler));
else
m_handlers.set(id, move(handler));
return id;
}
bool SignalHandlers::remove(int handler_id)
{
VERIFY(handler_id != 0);
if (m_calling_handlers) {
auto it = m_handlers.find(handler_id);
if (it != m_handlers.end()) {
// Mark pending remove
m_handlers_pending.set(handler_id, {});
return true;
}
it = m_handlers_pending.find(handler_id);
if (it != m_handlers_pending.end()) {
if (!it->value)
return false; // already was marked as deleted
it->value = nullptr;
return true;
}
return false;
}
return m_handlers.remove(handler_id);
}
void EventLoopImplementationUnix::handle_signal(int signal_number)
{
VERIFY(signal_number != 0);
auto& thread_data = ThreadData::the();
// We MUST check if the current pid still matches, because there
// is a window between fork() and exec() where a signal delivered
// to our fork could be inadvertently routed to the parent process!
if (getpid() == thread_data.pid) {
int nwritten = write(thread_data.wake_pipe_fds[1], &signal_number, sizeof(signal_number));
if (nwritten < 0) {
perror("EventLoopImplementationUnix::register_signal: write");
VERIFY_NOT_REACHED();
}
} else {
// We're a fork who received a signal, reset thread_data.pid.
thread_data.pid = getpid();
}
}
int EventLoopImplementationUnix::register_signal(int signal_number, Function<void(int)> handler)
{
VERIFY(signal_number != 0);
auto& info = *signals_info();
auto handlers = info.signal_handlers.find(signal_number);
if (handlers == info.signal_handlers.end()) {
auto signal_handlers = adopt_ref(*new SignalHandlers(signal_number, EventLoopImplementationUnix::handle_signal));
auto handler_id = signal_handlers->add(move(handler));
info.signal_handlers.set(signal_number, move(signal_handlers));
return handler_id;
} else {
return handlers->value->add(move(handler));
}
}
void EventLoopImplementationUnix::unregister_signal(int handler_id)
{
VERIFY(handler_id != 0);
int remove_signal_number = 0;
auto& info = *signals_info();
for (auto& h : info.signal_handlers) {
auto& handlers = *h.value;
if (handlers.remove(handler_id)) {
if (handlers.is_empty())
remove_signal_number = handlers.m_signal_number;
break;
}
}
if (remove_signal_number != 0)
info.signal_handlers.remove(remove_signal_number);
}
int EventLoopImplementationUnix::register_timer(Object& object, int milliseconds, bool should_reload, TimerShouldFireWhenNotVisible fire_when_not_visible)
{
VERIFY(milliseconds >= 0);
auto& thread_data = ThreadData::the();
auto timer = make<EventLoopTimer>();
timer->owner = object;
timer->interval = Time::from_milliseconds(milliseconds);
timer->reload(Time::now_monotonic_coarse());
timer->should_reload = should_reload;
timer->fire_when_not_visible = fire_when_not_visible;
int timer_id = thread_data.id_allocator.allocate();
timer->timer_id = timer_id;
thread_data.timers.set(timer_id, move(timer));
return timer_id;
}
bool EventLoopImplementationUnix::unregister_timer(int timer_id)
{
auto& thread_data = ThreadData::the();
thread_data.id_allocator.deallocate(timer_id);
return thread_data.timers.remove(timer_id);
}
void EventLoopImplementationUnix::register_notifier(Notifier& notifier)
{
ThreadData::the().notifiers.set(&notifier);
}
void EventLoopImplementationUnix::unregister_notifier(Notifier& notifier)
{
ThreadData::the().notifiers.remove(&notifier);
}
}

51
Userland/Libraries/LibCore/EventLoopImplementationUnix.h Normal file
View file

@ -0,0 +1,51 @@
/*
* Copyright (c) 2023, Andreas Kling <kling@serenityos.org>
*
* SPDX-License-Identifier: BSD-2-Clause
*/
#pragma once
#include <LibCore/EventLoopImplementation.h>
namespace Core {
class EventLoopImplementationUnix final : public EventLoopImplementation {
public:
EventLoopImplementationUnix();
virtual ~EventLoopImplementationUnix();
virtual int exec() override;
virtual size_t pump(PumpMode) override;
virtual void quit(int) override;
virtual void wake() override;
virtual void deferred_invoke(Function<void()>) override;
virtual int register_timer(Object&, int milliseconds, bool should_reload, TimerShouldFireWhenNotVisible) override;
virtual bool unregister_timer(int timer_id) override;
virtual void register_notifier(Notifier&) override;
virtual void unregister_notifier(Notifier&) override;
virtual void unquit() override;
virtual bool was_exit_requested() const override;
virtual void notify_forked_and_in_child() override;
virtual int register_signal(int signal_number, Function<void(int)> handler) override;
virtual void unregister_signal(int handler_id) override;
private:
void wait_for_events(PumpMode);
void dispatch_signal(int signal_number);
static void handle_signal(int signal_number);
static Optional<Time> get_next_timer_expiration();
bool m_exit_requested { false };
int m_exit_code { 0 };
// The wake pipe of this event loop needs to be accessible from other threads.
int (*m_wake_pipe_fds)[2];
};
}

4
Userland/Libraries/LibCore/Object.cpp
View file

@ -149,7 +149,9 @@ void Object::stop_timer()
if (!m_timer_id)
return;
bool success = Core::EventLoop::unregister_timer(m_timer_id);
VERIFY(success);
if (!success) {
dbgln("{} {:p} could not unregister timer {}", class_name(), this, m_timer_id);
}
m_timer_id = 0;
}

2
Userland/Libraries/LibCore/ThreadEventQueue.cpp
View file

@ -92,7 +92,7 @@ size_t ThreadEventQueue::process()
case Event::Quit:
VERIFY_NOT_REACHED();
default:
dbgln("ThreadEventQueue::process: Event of type {} with no receiver", event.type());
// Receiver disappeared, drop the event on the floor.
break;
}
} else if (event.type() == Event::Type::DeferredInvoke) {

5
Userland/Libraries/LibIPC/Connection.cpp
View file

@ -106,10 +106,7 @@ ErrorOr<void> ConnectionBase::post_message(MessageBuffer buffer)
dbgln("LibIPC::Connection FIXME Warning, needed {} writes needed to send message of size {}B, this is pretty bad, as it spins on the EventLoop", writes_done, initial_size);
}
// Note: This disables responsiveness detection when an event loop is absent.
// There are no users which both need this feature but don't have an event loop.
if (Core::EventLoop::has_been_instantiated())
m_responsiveness_timer->start();
m_responsiveness_timer->start();
return {};
}