cheatah
Source BigBrain
Source

stdlib/memory/owner.hpp

1// Copyright (c) 2026 BigBrain LLC. MIT-licensed (see LICENSE).
2// Original work; see ACKNOWLEDGMENTS.md for the open-source ideas we build upon.
3#pragma once
5/**
6 * @file owner.hpp
7 * @brief `memory::Owner<T>` — the sole owner + scheduling coordinator — and the `own()` factory.
8 *
9 * `T` is the only class template argument; scheduling `policy` is a constructor value, a write's
10 * priority is a compile-time argument on `rwrite`. Non-copyable and pinned (so `&value_` is stable
11 * forever — a renewed reader never dangles even when the value's internal buffer reallocates).
12 *
13 * The coordinator is a hand-rolled priority reader/writer lock (one `std::mutex` + `condition_variable`
14 * over explicit state — `std::shared_mutex` can't honor write priorities). Guarantees:
15 * - readers share; a writer is exclusive (no torn reads, no lost updates) — synchronized via the
16 * coordinator mutex at every grant/release, so object access is race-free with no lock held during use.
17 * - **drain-before-write**: a write request flips the current read generation's gate, so readers that
18 * loop on `valid()` yield; the write proceeds only once `readers_ == 0`.
19 * - **priority**: waiting writes are ordered by `(priority desc, arrival asc)` in a priority_queue.
20 * - **immediate-write** (`rwrite<memory::immediate>()`, i.e. priority < 0): bypasses the queue; if a
21 * writer is active AND cooperating (loops on `valid()`), it preempts that writer, does its write,
22 * then the writer resumes; against a non-looping writer it simply waits for it to finish, then goes
23 * ahead of the queue.
24 */
26#include <atomic>
27#include <condition_variable>
28#include <cstdint>
29#include <functional>
30#include <future>
31#include <memory>
32#include <mutex>
33#include <queue>
34#include <utility>
35#include <vector>
37#include "lease.hpp"
38#include "mode.hpp"
39#include "ownable.hpp"
40#include "policy.hpp"
41#include "request.hpp"
43namespace cheatah::memory {
45/**
46 * @brief The sole owner + scheduling coordinator for one `T`. Non-copyable and pinned, so the object
47 * never moves; every access goes through a request → acquire → lease. @tparam T the owned type.
48 */
49template <Ownable T>
50class Owner {
51public:
52 /// Take sole ownership by MOVING @p value in — the object is consumed (its resources move into the
53 /// Owner), never copied. An lvalue won't bind here (bind an rvalue: `own(std::move(x))`).
54 /// @param value the object to take ownership of (moved in).
55 /// @param pol the scheduling policy (interleave / writes_first).
56 /// @complexity O(1) plus moving @p value.
57 /// @alloc one read-generation gate (`std::make_shared`); the value's own storage just moves.
58 /// @concurrency construct before sharing; the pinned Owner must outlive every lease and every
59 /// thread that uses it.
60 /// @test Memory.OwnerConsumesAndMovesTheObjectInNeverCopies
61 explicit Owner(T&& value, policy pol = policy::interleave)
62 : value_(std::move(value)), policy_(pol) { read_gate_ = make_gate(); }
63 Owner(const Owner&) = delete; // copying an owner is forbidden — there is one owner.
64 Owner& operator=(const Owner&) = delete; // sole ownership; pinned (mutex is non-movable).
65 Owner(Owner&&) = delete; // pinned: &value_ must stay stable for live leases.
66 Owner& operator=(Owner&&) = delete;
68 /// Request a shared READ lease. Blocks — in THIS call: the grant is synchronous, so the returned
69 /// request is already fulfilled — only while a write is pending/active or queued; otherwise many
70 /// read leases coexist. @return a request for a read lease.
71 /// @complexity O(1) amortized. @alloc the request's promise/future.
72 /// @concurrency callable from any thread; readers share. Writer-preference: this waits while any
73 /// write is active, suspended, or queued, so writers cannot starve.
74 /// @warning requesting while the SAME thread still holds a lease on this owner can deadlock (a
75 /// queued write makes the grant wait on that very lease) — renewal is release, then re-request.
76 /// @test Memory.ReadLeasesCoexist
77 /// @test MemoryConcurrency.ManyReadersCoexistThenAWriteDrains
78 /// @systest MemoryCheatah.ReadLeaseValidState
79 Request<Lease<T, read>> rread() {
80 std::promise<Lease<T, read>> p;
81 auto fut = p.get_future();
82 {
83 std::unique_lock<std::mutex> lk(mtx_);
84 cv_.wait(lk, [&] { return can_read(); });
85 ++readers_;
86 auto gate = read_gate_;
87 p.set_value(Lease<T, read>(&value_, std::move(gate), [this] { release_read(); }));
88 }
89 return Request<Lease<T, read>>(std::move(fut));
90 }
92 /// Request an exclusive WRITE lease at compile-time `priority` (a plain int or the caller's enum;
93 /// higher = served first, ties FIFO). `priority < 0` (spell it `memory::immediate`) is an
94 /// immediate-write. Blocks in THIS call — the grant is synchronous, so the returned request is
95 /// already fulfilled — until the readers drain and this write wins the queue.
96 /// @tparam priority the compile-time write priority. @return a request for a write
97 /// lease. @complexity O(log k) to enqueue among k waiters (O(1) immediate), plus the blocking wait.
98 /// @alloc the request's promise/future, two fresh gates (`std::make_shared`: this write's own +
99 /// the next read generation's), plus one queue-ticket slot (amortized) for a non-immediate write.
100 /// @concurrency callable from any thread. Drain-before-write: flips the current read generation's
101 /// gate and waits until every reader has released and no other write is active; an immediate-write
102 /// skips the queue and additionally preempts a cooperating active writer (which resumes after).
103 /// @warning requesting while the SAME thread still holds a lease on this owner deadlocks (the
104 /// drain waits on that very lease) — release first, then re-request.
105 /// @test Memory.WriteWaitsForReadersToDrain
106 /// @test Memory.HigherPriorityWriteServedFirst
107 /// @test Memory.NegativePriorityImmediateWritePreemptsTheActiveWriterWhichThenResumes
108 /// @test MemoryConcurrency.ManyWritersDeterministicSum
109 /// @systest MemoryCheatah.ConcurrentSumOverSharedOwner
110 template <auto priority = 0>
111 Request<Lease<T, write>> rwrite() {
112 constexpr long long P = static_cast<long long>(priority);
113 if constexpr (P < 0) return grant_immediate();
114 else return grant_write(P);
115 }
117private:
118 // ── the object + coordinator state (all guarded by mtx_) ──
119 T value_;
120 policy policy_;
121 std::mutex mtx_;
122 std::condition_variable cv_;
123 long long readers_ = 0; ///< active read leases.
124 bool writer_ = false; ///< an active (non-suspended) write lease.
125 bool writer_suspended_ = false; ///< active writer paused for an immediate-write.
126 bool immediate_ = false;///< an immediate-write holds exclusive access.
127 std::shared_ptr<detail::Gate> read_gate_; ///< current read generation's yield gate.
128 std::shared_ptr<detail::Gate> writer_gate_; ///< the active writer's gate (for preempt/resume).
130 struct Ticket { long long prio; std::uint64_t seq; };
131 struct ServedLater { // priority_queue is a max-heap: top = highest priority, then earliest arrival.
132 /**
133 * The heap ordering: is @p a served later than @p b? Higher priority wins; within a
134 * priority, the earlier arrival (lower seq) wins — FIFO among equals.
135 * @param a one waiting write's ticket.
136 * @param b the other waiting write's ticket.
137 * @return true iff @p a is served after @p b.
138 * @complexity O(1).
139 * @alloc none.
140 * @test Memory.HigherPriorityWriteServedFirst
141 */
142 bool operator()(const Ticket& a, const Ticket& b) const {
143 return a.prio != b.prio ? a.prio < b.prio : a.seq > b.seq;
144 }
145 };
146 std::priority_queue<Ticket, std::vector<Ticket>, ServedLater> wq_; ///< waiting non-immediate writes.
147 std::uint64_t seq_ = 0;
149 // Readers proceed only when no writer/immediate is active or paused and no writer is queued
150 // (writer-preference — this is what forces the drain and prevents writer starvation).
151 bool can_read() const { return !writer_ && !immediate_ && !writer_suspended_ && wq_.empty(); }
153 // A gate whose holder, on observing !valid, wakes our cv_. The wake must synchronize on
154 // mtx_ before notifying: the holder acks from outside the lock, so a bare notify_all could
155 // land while grant_immediate() still holds mtx_ evaluating its wait predicate (acked read
156 // as false, waiter not yet blocked) — and since the ack is one-shot, that lost wakeup left
157 // the preempting write asleep forever against a writer spinning on valid(). Taking and
158 // releasing mtx_ first pins the wake after the waiter is actually waiting.
159 std::shared_ptr<detail::Gate> make_gate() {
160 auto g = std::make_shared<detail::Gate>();
161 g->wake = [this] {
162 { std::lock_guard<std::mutex> lk(mtx_); } // serialize with a waiter mid-predicate
163 cv_.notify_all();
164 };
165 return g;
166 }
168 Request<Lease<T, write>> grant_write(long long prio) {
169 std::promise<Lease<T, write>> p;
170 auto fut = p.get_future();
171 {
172 std::unique_lock<std::mutex> lk(mtx_);
173 const std::uint64_t my = ++seq_;
174 wq_.push({prio, my});
175 read_gate_->valid.store(false); // ask current readers to yield (drain)
176 cv_.notify_all();
177 cv_.wait(lk, [&] {
178 return !writer_ && !immediate_ && !writer_suspended_ && readers_ == 0 &&
179 !wq_.empty() && wq_.top().seq == my; // no active access + I'm the winner
180 });
181 wq_.pop();
182 writer_ = true;
183 writer_gate_ = make_gate(); // fresh, valid — this writer is preemptible
184 read_gate_ = make_gate(); // fresh read generation for future readers
185 p.set_value(Lease<T, write>(&value_, writer_gate_, [this] { release_write(); }));
186 }
187 return Request<Lease<T, write>>(std::move(fut));
188 }
190 Request<Lease<T, write>> grant_immediate() {
191 std::promise<Lease<T, write>> p;
192 auto fut = p.get_future();
193 {
194 std::unique_lock<std::mutex> lk(mtx_);
195 if (writer_) { // preempt a cooperating active writer
196 writer_gate_->valid.store(false);
197 cv_.notify_all();
198 // Short-circuit !writer_ FIRST: a non-looping writer may release (writer_gate_ ->
199 // nullptr) during the wait, so never deref the gate once the writer is gone.
200 cv_.wait(lk, [&] { return !writer_ || writer_gate_->acked.load(); });
201 if (writer_) { writer_suspended_ = true; writer_ = false; } // it paused → suspend it
202 // else it finished on its own; no resume owed.
203 }
204 read_gate_->valid.store(false); // drain any readers
205 cv_.notify_all();
206 cv_.wait(lk, [&] { return readers_ == 0 && !immediate_; });
207 immediate_ = true;
208 read_gate_ = make_gate();
209 auto ig = make_gate();
210 p.set_value(Lease<T, write>(&value_, std::move(ig), [this] { release_immediate(); }));
211 }
212 return Request<Lease<T, write>>(std::move(fut));
213 }
215 void release_read() {
216 std::lock_guard<std::mutex> lk(mtx_);
217 --readers_;
218 cv_.notify_all();
219 }
220 void release_write() {
221 std::lock_guard<std::mutex> lk(mtx_);
222 writer_ = false;
223 writer_gate_ = nullptr;
224 cv_.notify_all();
225 }
226 void release_immediate() {
227 std::lock_guard<std::mutex> lk(mtx_);
228 immediate_ = false;
229 if (writer_suspended_) { // resume the writer we preempted (before the queue)
230 writer_suspended_ = false;
231 writer_ = true;
232 writer_gate_->acked.store(false);
233 writer_gate_->valid.store(true); // its valid() flips back to true → it continues
234 }
235 cv_.notify_all();
236 }
237};
239/// Take sole ownership of @p value and hand back its `Owner`. @tparam T the owned type.
240/// @param value the object to own (moved in). @param pol the scheduling policy.
241/// @return an `Owner<T>` that has consumed @p value.
242/// @complexity O(1) plus moving @p value. @alloc one read-generation gate (`std::make_shared`, in
243/// the `Owner` constructor); the value's own storage just moves.
244/// @test Memory.ObjectDiesWithOwner
245template <Ownable T>
246Owner<T> own(T value, policy pol = policy::interleave) { return Owner<T>(std::move(value), pol); }
248} // namespace cheatah::memory