samiandClaude Sonnet 5
6d7c4f7bc4
Introduces the four rpc-layer seams docs/08-porting.md calls for this lane: platform::Wakeup (eventfd), platform::peer_of (SO_PEERCRED/struct ucred, same-UID check preserved unchanged), platform::acquire_instance_lock (abstract-namespace socket), platform::runtime_base_dir/data_base_dir (XDG lookups). Today's Linux code moves unchanged into daemon/src/rpc/platform/linux/; the seam headers carry no OS types and no #ifdef. No fifth interface for timerfd: EventLoop gains a portable add_timer() that folds the next deadline into poll()'s own timeout, replacing both timerfd instances in main.cpp — the loop already computes a deadline, so this needs no per-OS backend at all. Full suite 59/59 green; no #ifdef outside platform/linux/, no behaviour change. Co-Authored-By: Claude Sonnet 5 <[email protected]>
104 lines
3.5 KiB
C++
104 lines
3.5 KiB
C++
#pragma once
|
|
|
|
// A single-threaded poll(2) reactor. Every RPC listener and connection registers its fd
|
|
// here; the loop never blocks on disk or DNS (AGENT-DAEMON.md build step 1 — "Never block
|
|
// the RPC loop"). Long work is handed to CORE's pools later; this class only multiplexes
|
|
// readiness.
|
|
//
|
|
// Thread model: run() executes on one thread. add_fd/mod_fd/del_fd are called from
|
|
// callbacks on that same thread. stop() and wake() are async-signal-safe and safe to call
|
|
// from any thread or a signal handler — they only write() a byte to an internal eventfd.
|
|
|
|
#include <atomic>
|
|
#include <chrono>
|
|
#include <cstdint>
|
|
#include <functional>
|
|
#include <mutex>
|
|
#include <unordered_map>
|
|
#include <vector>
|
|
|
|
#include "rpc/platform/wakeup.hpp"
|
|
|
|
namespace velox::daemon::rpc {
|
|
|
|
enum Interest : unsigned {
|
|
kNone = 0,
|
|
kRead = 1u << 0,
|
|
kWrite = 1u << 1,
|
|
};
|
|
|
|
class EventLoop {
|
|
public:
|
|
// Called when the fd is readable and/or writable. `events` is the subset of the fd's
|
|
// registered Interest that fired. A callback may add/modify/remove any fd, including
|
|
// its own, and may call stop().
|
|
using Callback = std::function<void(int fd, unsigned events)>;
|
|
|
|
EventLoop();
|
|
~EventLoop();
|
|
|
|
EventLoop(const EventLoop&) = delete;
|
|
EventLoop& operator=(const EventLoop&) = delete;
|
|
|
|
// Register `fd` (must be non-blocking) for `interest`. Replaces any prior registration.
|
|
void add_fd(int fd, unsigned interest, Callback cb);
|
|
// Change the interest mask for an already-registered fd.
|
|
void mod_fd(int fd, unsigned interest);
|
|
// Stop watching `fd`. Does not close it — ownership stays with the caller.
|
|
void del_fd(int fd);
|
|
|
|
// Run until stop() is called. Re-entrant calls are not supported.
|
|
void run();
|
|
|
|
// Ask run() to return after the current poll wakeup. Async-signal-safe.
|
|
void stop() noexcept;
|
|
|
|
// Force one poll() wakeup without stopping — used when interest changed from outside a
|
|
// callback. Async-signal-safe.
|
|
void wake() noexcept;
|
|
|
|
// Run `fn` on the loop thread at the next iteration. Thread-safe; the intended way to
|
|
// marshal an engine-thread callback back onto the RPC loop.
|
|
void post(std::function<void()> fn);
|
|
|
|
using TimerId = std::uint64_t;
|
|
|
|
// Fire `cb` roughly every `interval` for as long as the loop runs, expressed as a
|
|
// poll(2) timeout rather than a platform timer fd (docs/08-porting.md: "the loop
|
|
// already has a deadline set" — no per-OS backend needed for this one). Not
|
|
// reentrant-safe to call from inside a timer callback other than the one running.
|
|
TimerId add_timer(std::chrono::milliseconds interval, std::function<void()> cb);
|
|
void remove_timer(TimerId id);
|
|
|
|
private:
|
|
struct Entry {
|
|
unsigned interest;
|
|
Callback cb;
|
|
};
|
|
|
|
struct Timer {
|
|
std::chrono::steady_clock::time_point next;
|
|
std::chrono::milliseconds interval;
|
|
std::function<void()> cb;
|
|
};
|
|
|
|
void drain_posts();
|
|
// Milliseconds until the next timer is due, or -1 if there are none (poll()'s "block
|
|
// forever" convention).
|
|
int next_timeout_ms() const;
|
|
void run_due_timers();
|
|
|
|
platform::Wakeup wakeup_;
|
|
bool running_ = false;
|
|
std::atomic<bool> stop_requested_ = false; // set from stop(), read by run()
|
|
std::unordered_map<int, Entry> fds_;
|
|
|
|
std::mutex post_mu_;
|
|
std::vector<std::function<void()>> posts_;
|
|
|
|
TimerId next_timer_id_ = 1;
|
|
std::unordered_map<TimerId, Timer> timers_;
|
|
};
|
|
|
|
} // namespace velox::daemon::rpc
|