From bc97c80eedccdc0d3a6c22076541c4455500a450 Mon Sep 17 00:00:00 2001 From: Mirko Brombin Date: Sun, 20 Sep 2026 20:45:03 +0200 Subject: [PATCH] feat: add sleep on the reactor's descriptors and a wake event when idle, and poll other processes' frames only for windows they draw into --- native/desktop/foreign_surface.cpp | 33 +++++++++---- native/desktop/foreign_surface.h | 2 + native/desktop/windows.cpp | 11 ++++- native/desktop/windows.h | 1 + native/nt/completion_port.h | 26 ++++++++-- native/nt/pipe.cpp | 4 +- native/nt/providers.cpp | 1 + native/nt/shared_file.h | 79 +++++++++++++++++++++++++++--- native/nt/thread_api.cpp | 15 +++++- native/nt/threads.cpp | 31 ++++++++---- native/nt/threads.h | 9 ++-- native/nt/winsock_api.cpp | 8 +-- native/nt/x86_process.cpp | 1 + 13 files changed, 180 insertions(+), 41 deletions(-) diff --git a/native/desktop/foreign_surface.cpp b/native/desktop/foreign_surface.cpp index 024a610..5e2588a 100644 --- a/native/desktop/foreign_surface.cpp +++ b/native/desktop/foreign_surface.cpp @@ -27,16 +27,20 @@ namespace { constexpr size_t header_size = 4096; constexpr uint32_t surface_magic = 0x52574653; -std::filesystem::path window_directory() +// The directory is the same for the whole life of the process; it is made the first time it is asked for. +const std::filesystem::path &window_directory() { - std::filesystem::path base; - if (const char *runtime = std::getenv("XDG_RUNTIME_DIR"); runtime && *runtime) - base = runtime; - else - base = NativePrefix::from_environment().root(); - auto directory = base / ("raven-windows-" + prefix_tag()); - std::error_code error; - std::filesystem::create_directories(directory, error); + static const std::filesystem::path directory = [] { + std::filesystem::path base; + if (const char *runtime = std::getenv("XDG_RUNTIME_DIR"); runtime && *runtime) + base = runtime; + else + base = NativePrefix::from_environment().root(); + auto made = base / ("raven-windows-" + prefix_tag()); + std::error_code error; + std::filesystem::create_directories(made, error); + return made; + }(); return directory; } @@ -52,7 +56,8 @@ std::string window_name(uint64_t window) uint64_t window_handle_base() { // Sixteen bits of window number under a tag from the process id; the tag stays clear of the low handles other objects use. - return uint64_t((uint32_t(::getpid()) % 0x7ef0) + 0x100) << 16; + static const uint64_t base = uint64_t((uint32_t(::getpid()) % 0x7ef0) + 0x100) << 16; + return base; } // Removes what processes that ended without withdrawing their windows left: a window whose owner is gone, and pixels without a window. @@ -164,6 +169,14 @@ void ForeignSurface::publish_frame() std::atomic_ref(header[3]).fetch_add(1, std::memory_order_release); } +uint64_t foreign_frames_stamp() +{ + struct stat status{}; + if (::stat(window_directory().c_str(), &status) != 0) + return 0; + return uint64_t(status.st_mtim.tv_sec) * 1000000000ULL + uint64_t(status.st_mtim.tv_nsec); +} + bool take_foreign_frame(uint64_t window, uint32_t &seen, const std::function &use) { const auto path = window_directory() / (window_name(window) + ".pixels"); diff --git a/native/desktop/foreign_surface.h b/native/desktop/foreign_surface.h index e60e8f5..2ddf8f6 100644 --- a/native/desktop/foreign_surface.h +++ b/native/desktop/foreign_surface.h @@ -41,5 +41,7 @@ private: // Owner side: when another process finished a new frame for the window, hands its pixels (32-bit rows, top down) to `use` and returns // true; `seen` remembers the last frame taken. bool take_foreign_frame(uint64_t window, uint32_t &seen, const std::function &use); +// Changes whenever another process starts or stops drawing into a window of the prefix (its pixels file is made or removed). +uint64_t foreign_frames_stamp(); } diff --git a/native/desktop/windows.cpp b/native/desktop/windows.cpp index fbd24d2..b503c4a 100644 --- a/native/desktop/windows.cpp +++ b/native/desktop/windows.cpp @@ -249,11 +249,20 @@ void NativeWindows::import_foreign_frames() const bool poll=now-foreign_checked>=std::chrono::milliseconds(8); if (poll) foreign_checked=now; if (!poll && dirty_roots.empty()) return; + // Only windows another process has drawn into are looked at on every poll; all of them only when the set of drawn windows changed. + bool every=false; + if (poll) { + const auto stamp=foreign_frames_stamp(); + every=stamp!=foreign_stamp; + foreign_stamp=stamp; + } for (const auto handle : objects.window_handles()) { if (handlewindow_handle_base()+0xffff) continue; const auto known=foreign_frames.find(handle); - const bool redraw=known!=foreign_frames.end() && known->second && dirty_roots.contains(objects.window_root(handle)); + const bool drawn=known!=foreign_frames.end() && known->second; + const bool redraw=drawn && dirty_roots.contains(objects.window_root(handle)); if (!poll && !redraw) continue; + if (poll && !redraw && !every && !drawn) continue; auto &seen=foreign_frames[handle]; if (redraw) seen=0; take_foreign_frame(handle,seen,[&](const uint8_t *pixels,uint32_t width,uint32_t height) { diff --git a/native/desktop/windows.h b/native/desktop/windows.h index 4311d4a..82b27c1 100644 --- a/native/desktop/windows.h +++ b/native/desktop/windows.h @@ -86,6 +86,7 @@ private: uint64_t foreign_dc(uint64_t window); std::map foreign_frames; std::chrono::steady_clock::time_point foreign_checked{}; + uint64_t foreign_stamp=UINT64_MAX; void import_foreign_frames(); // Host input collected while threads waited, delivered by the next input_ready. std::deque pending_events; diff --git a/native/nt/completion_port.h b/native/nt/completion_port.h index a4a74db..b364c0b 100644 --- a/native/nt/completion_port.h +++ b/native/nt/completion_port.h @@ -6,6 +6,7 @@ #include #include #include +#include #include namespace raven { @@ -40,14 +41,19 @@ private: class Reactor { public: using Operation = std::function; - // A pending operation counts as work only the host can finish, so waiting threads are not taken for deadlocked. - void add(Operation operation) { operations.push_back({std::move(operation), std::make_unique()}); } + // A pending operation counts as work only the host can finish, so waiting threads are not taken for deadlocked. The descriptor and + // events it waits for, when it names them, let an idle process sleep until the operation can go on instead of trying it again and + // again. + void add(Operation operation, int descriptor = -1, short events = 0) + { + operations.emplace_back(Pending{std::move(operation), descriptor, events}, std::make_unique()); + } // Gives every pending operation a chance to finish; true when any did. bool poll() { bool finished = false; for (size_t index = 0; index < operations.size();) { - if (operations[index].first()) { + if (operations[index].first.run()) { operations.erase(operations.begin() + ptrdiff_t(index)); finished = true; } else { @@ -57,9 +63,21 @@ public: return finished; } size_t pending() const { return operations.size(); } + // The descriptors pending operations wait on. + void watch(std::vector &descriptors) const + { + for (const auto &[operation, wait] : operations) + if (operation.descriptor >= 0) + descriptors.push_back({operation.descriptor, operation.events, 0}); + } private: - std::vector>> operations; + struct Pending { + Operation run; + int descriptor; + short events; + }; + std::vector>> operations; }; } diff --git a/native/nt/pipe.cpp b/native/nt/pipe.cpp index 218fb0d..e53bb50 100644 --- a/native/nt/pipe.cpp +++ b/native/nt/pipe.cpp @@ -257,7 +257,7 @@ void start_overlapped(KernelContext &context, ApiCall &call, const std::shared_p state.pending.erase(std::remove(state.pending.begin(), state.pending.end(), io), state.pending.end()); finish_io(context, port, key, overlapped, width, error, bytes); return true; - }); + }, pipe->descriptor, short(shared->reading ? POLLIN : POLLOUT)); call.fail(error_io_pending); } @@ -527,7 +527,7 @@ void register_pipe_api(ApiRegistry ®istry, KernelContext &context) state.pending.erase(std::remove(state.pending.begin(), state.pending.end(), io), state.pending.end()); finish_io(context, port, key, overlapped, width, error, 0); return true; - }); + }, pipe->listener ? pipe->listener->descriptor : -1, short(POLLIN)); call.fail(error_io_pending); }); registry.add("DisconnectNamedPipe", 1, [pipe_of](ApiCall &call) { diff --git a/native/nt/providers.cpp b/native/nt/providers.cpp index 2ab7437..8c75f3f 100644 --- a/native/nt/providers.cpp +++ b/native/nt/providers.cpp @@ -1164,6 +1164,7 @@ NativeResult run_native_process(ModuleSet &modules, PEModule &entry, object_api.set_threads(&threads); windows.set_yield([&threads] { return threads.live() > 1 && threads.yield(); }); KernelContext kernel{process, environment, memory, allocator, kernel_objects, threads, paths, modules, entry, *launch, invoke}; + threads.set_watched([reactor = kernel.reactor](std::vector &descriptors) { reactor->watch(descriptors); }); windows.set_global_allocator([&allocator](uint64_t size) { return allocator.allocate(size,true); }); kernel.message_ready=[&windows](uint32_t thread) { return windows.input_pending(thread); }; windows.set_block([&threads](std::function ready) { diff --git a/native/nt/shared_file.h b/native/nt/shared_file.h index bbb5f40..79e5ebe 100644 --- a/native/nt/shared_file.h +++ b/native/nt/shared_file.h @@ -6,14 +6,18 @@ #include #include #include +#include #include #include #include +#include #include #include #include #include #include +#include +#include #include namespace raven { @@ -21,6 +25,71 @@ namespace raven { // The number of shared objects this process holds; waits on them are polled, so the scheduler looks more often while any exists. extern std::atomic shared_object_count; +// The directory that holds the shared objects of the prefix. +inline std::filesystem::path shared_objects_directory() +{ + std::filesystem::path base; + if (const char *runtime = std::getenv("XDG_RUNTIME_DIR"); runtime && *runtime) + base = runtime; + else + base = NativePrefix::from_environment().root(); + return base / ("raven-objects-" + prefix_tag()); +} + +// A word every process of the prefix maps and bumps when it changes a shared object another process may be waiting for: an idle process +// sleeps until it changes instead of looking at its shared objects every few milliseconds. Null when it cannot be made. +inline std::atomic *shared_changes() +{ + static std::atomic *word = [] () -> std::atomic * { + const auto directory = shared_objects_directory(); + std::error_code failure; + std::filesystem::create_directories(directory, failure); + const int descriptor = ::open((directory / ".changes").c_str(), O_CREAT | O_RDWR | O_CLOEXEC, 0600); + if (descriptor < 0) + return nullptr; + if (::ftruncate(descriptor, 4096) != 0) { + ::close(descriptor); + return nullptr; + } + void *mapped = ::mmap(nullptr, 4096, PROT_READ | PROT_WRITE, MAP_SHARED, descriptor, 0); + ::close(descriptor); + return mapped == MAP_FAILED ? nullptr : static_cast *>(mapped); + }(); + return word; +} + +inline void announce_shared_change() +{ + if (auto *word = shared_changes()) { + word->fetch_add(1); + ::syscall(SYS_futex, word, FUTEX_WAKE, INT32_MAX, nullptr, nullptr, 0); + } +} + +// The eventfd of the process's scheduler, written whenever the shared word changes. +inline std::atomic shared_change_listener{-1}; + +// Starts, once, the thread that waits for changes to the shared word and wakes the scheduler. +inline void listen_shared_changes() +{ + static std::once_flag started; + std::call_once(started, [] { + auto *word = shared_changes(); + if (!word) + return; + std::thread([word] { + for (;;) { + const uint32_t seen = word->load(); + ::syscall(SYS_futex, word, FUTEX_WAIT, seen, nullptr, nullptr, 0); + if (const int descriptor = shared_change_listener.load(); descriptor >= 0) { + const uint64_t one = 1; + [[maybe_unused]] const auto written = ::write(descriptor, &one, sizeof one); + } + } + }).detach(); + }); +} + // Holds the lock of a directory of shared object files until the returned descriptor is closed; -1 when the lock file cannot be made. inline int lock_directory(const std::filesystem::path &directory) { @@ -36,7 +105,7 @@ inline void sweep_directory(const std::filesystem::path &directory) { std::error_code error; for (const auto &entry : std::filesystem::directory_iterator(directory, error)) { - if (entry.path().filename() == ".lock") + if (entry.path().filename() == ".lock" || entry.path().filename() == ".changes") continue; const int probe = ::open(entry.path().c_str(), O_RDWR | O_CLOEXEC); if (probe < 0) @@ -82,12 +151,8 @@ public: const std::function &init = {}) { existed = false; - std::filesystem::path base; - if (const char *runtime = std::getenv("XDG_RUNTIME_DIR"); runtime && *runtime) - base = runtime; - else - base = NativePrefix::from_environment().root(); - auto directory = base / ("raven-objects-" + prefix_tag()); + listen_shared_changes(); + auto directory = shared_objects_directory(); std::error_code failure; std::filesystem::create_directories(directory, failure); static std::once_flag swept; diff --git a/native/nt/thread_api.cpp b/native/nt/thread_api.cpp index 9859882..7b854c9 100644 --- a/native/nt/thread_api.cpp +++ b/native/nt/thread_api.cpp @@ -83,7 +83,13 @@ public: int signaled(uint32_t) const override { return state->signaled.load() ? 1 : 0; } bool try_consume(uint32_t) override { return state->manual || state->signaled.exchange(0) == 1; } void consume(uint32_t thread) override { try_consume(thread); } - int set_state(bool value) override { return int(state->signaled.exchange(value ? 1 : 0)); } + int set_state(bool value) override + { + const int previous = int(state->signaled.exchange(value ? 1 : 0)); + if (value) + announce_shared_change(); + return previous; + } EventState *state = nullptr; std::shared_ptr shared; }; @@ -619,8 +625,11 @@ void register_thread_api(ApiRegistry ®istry, KernelContext &context) call.fail(288); return; } - if (--mutex->state->recursion == 0) + if (--mutex->state->recursion == 0) { mutex->state->owner = 0; + if (mutex->shared) + announce_shared_change(); + } call.ret(1); }); const auto make_semaphore = [&objects](ApiCall &call, int64_t initial, int64_t maximum, uint64_t name) { @@ -688,6 +697,8 @@ void register_thread_api(ApiRegistry ®istry, KernelContext &context) if (semaphore->state->count.compare_exchange_weak(current, current + amount)) break; } + if (semaphore->shared) + announce_shared_change(); if (call[2]) call.store(call[2], current); call.ret(1); }); diff --git a/native/nt/threads.cpp b/native/nt/threads.cpp index b173dc9..746e10a 100644 --- a/native/nt/threads.cpp +++ b/native/nt/threads.cpp @@ -10,8 +10,11 @@ #include #include #include +#include +#include #include #include +#include namespace raven { @@ -40,16 +43,23 @@ NativeThreads::NativeThreads(NativeProcess &process, NativeEnvironment &environm if (const char *text = std::getenv("RAVEN_SLICE_MS")) slice = std::strtol(text, nullptr, 10); process.set_time_slicing(std::chrono::milliseconds(slice < 0 ? 0 : slice), [this] { return !tearing_down && live() > 1; }, [this] { yield(); }); + wake = ::eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK); + if (wake < 0) + throw std::runtime_error("cannot create the scheduler's wake descriptor"); + shared_change_listener = wake; } NativeThreads::~NativeThreads() { + if (shared_change_listener == wake) + shared_change_listener = -1; process.set_time_slicing(std::chrono::milliseconds(0), nullptr, nullptr); // Host frames of threads that never finished hold objects that only their unwinding releases; a failure here must not escape. try { unwind_all(); } catch (...) { } + ::close(wake); } void NativeThreads::stop_time_slicing() @@ -315,24 +325,27 @@ void NativeThreads::schedule() throw std::runtime_error("native guest threads deadlocked: every thread waits without a timeout"); const auto limit = std::min(wake, deadline); auto slice = std::chrono::duration_cast(limit > now ? limit - now : std::chrono::nanoseconds(0)); - slice = std::min(slice, std::chrono::milliseconds(shared_object_count.load() > 0 ? 2 : 20)); + // The process sleeps until a host thread nudges it, another process changes a shared object, a pending operation's descriptor is + // ready or the earliest timeout; conditions nothing announces, such as a child process ending, are looked at every 20 ms. + slice = std::min(slice, std::chrono::milliseconds(20)); if (poll) { poll(slice); } else if (slice.count() > 0) { - std::unique_lock hold(idle_lock); - idle_signal.wait_for(hold, slice, [this] { return nudged; }); - nudged = false; + std::vector descriptors{{this->wake, POLLIN, 0}}; + if (watched) + watched(descriptors); + const timespec timeout{slice.count() / 1000, (slice.count() % 1000) * 1000000}; + ::ppoll(descriptors.data(), descriptors.size(), &timeout, nullptr); + uint64_t drained = 0; + while (::read(this->wake, &drained, sizeof drained) > 0) {} } } } void NativeThreads::nudge() { - { - std::lock_guard hold(idle_lock); - nudged = true; - } - idle_signal.notify_one(); + const uint64_t one = 1; + [[maybe_unused]] const auto written = ::write(wake, &one, sizeof one); } bool NativeThreads::block(std::function ready, std::optional wake) diff --git a/native/nt/threads.h b/native/nt/threads.h index 72c95d2..5eca7bb 100644 --- a/native/nt/threads.h +++ b/native/nt/threads.h @@ -13,6 +13,7 @@ #include #include #include +#include #include namespace raven { @@ -77,6 +78,8 @@ public: } // Host polling used while every guest thread waits; it may make waiting predicates true (window events). void set_idle(std::function idle) { poll = std::move(idle); } + // Adds the descriptors whose readiness can make a waiting thread's condition true; the idle wait sleeps on them. + void set_watched(std::function &)> descriptors) { watched = std::move(descriptors); } // True while something outside the process, such as a child program, can still make a waiting thread's condition true. void add_external(std::function pending) { externals.push_back(std::move(pending)); } // Runs whenever every thread waits, so work the host does for the guest (finishing started operations) can make a wait succeed. @@ -119,9 +122,9 @@ private: std::vector> externals; std::vector> pumps; std::vector> services; - std::mutex idle_lock; - std::condition_variable idle_signal; - bool nudged = false; + // Written to end the idle wait: by nudge() and by the thread that sees other processes change shared objects. + int wake = -1; + std::function &)> watched; std::vector> threads; GuestThread *running = nullptr; // Set while the process ends: threads leave through their frames and the deadline no longer interrupts them. diff --git a/native/nt/winsock_api.cpp b/native/nt/winsock_api.cpp index 5dbe3c3..829e784 100644 --- a/native/nt/winsock_api.cpp +++ b/native/nt/winsock_api.cpp @@ -657,6 +657,8 @@ void register_winsock_api(ApiRegistry ®istry, KernelContext &context) call.write(transfer.overlapped, pending); } auto shared = std::make_shared(std::move(transfer)); + const int descriptor = shared->socket->descriptor; + const short events = shared->receiving ? POLLIN : POLLOUT; reactor.add([shared, attempt, complete]() { uint32_t result = 0; uint64_t count = 0; @@ -664,7 +666,7 @@ void register_winsock_api(ApiRegistry ®istry, KernelContext &context) return false; complete(*shared, result, count, false); return true; - }); + }, descriptor, events); call.fail(997, UINT32_MAX); }; // Without an OVERLAPPED the call is an ordinary blocking or non-blocking transfer. @@ -883,7 +885,7 @@ void register_winsock_api(ApiRegistry ®istry, KernelContext &context) ::getsockopt(socket->descriptor, SOL_SOCKET, SO_ERROR, &error, &length); done(error ? wsa_error(error) : 0, false); return true; - }); + }, socket->descriptor, short(POLLOUT)); fail(call, 997, 0); }); registry.add("AcceptEx", 8, [socket_of, fail, finish, &reactor, &process](ApiCall &call) { @@ -944,7 +946,7 @@ void register_winsock_api(ApiRegistry ®istry, KernelContext &context) return false; finish(listener, overlapped, width, uint32_t(result), 0, false); return true; - }); + }, listener->descriptor, short(POLLIN)); fail(call, 997, 0); }); registry.add("GetAcceptExSockaddrs", 8, [](ApiCall &call) { diff --git a/native/nt/x86_process.cpp b/native/nt/x86_process.cpp index efee434..811d73d 100644 --- a/native/nt/x86_process.cpp +++ b/native/nt/x86_process.cpp @@ -1045,6 +1045,7 @@ NativeResult run_native_process_x86(ModuleSet &modules, PEModule &entry, GuestAllocator allocator(memory, guest_heap); compat.set_global_blocks([&allocator](uint64_t address) { return allocator.size(address); }, [&allocator](uint64_t address) { return allocator.release(address); }); KernelContext kernel{process, environment, memory, allocator, kernel_objects, guest_threads, paths, modules, entry, *launch, invoke}; + guest_threads.set_watched([reactor = kernel.reactor](std::vector &descriptors) { reactor->watch(descriptors); }); register_file_api(shared_api, kernel); register_memory_api(shared_api, kernel); register_text_api(shared_api, kernel);