Branch data Line data Source code
1 : : /*
2 : : * Copyright 2025 Jinwoo Sung
3 : : *
4 : : * Licensed under the Apache License, Version 2.0 (the "License");
5 : : * you may not use this file except in compliance with the License.
6 : : * You may obtain a copy of the License at
7 : : *
8 : : * http://www.apache.org/licenses/LICENSE-2.0
9 : : *
10 : : * Unless required by applicable law or agreed to in writing, software
11 : : * distributed under the License is distributed on an "AS IS" BASIS,
12 : : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 : : * See the License for the specific language governing permissions and
14 : : * limitations under the License.
15 : : */
16 : :
17 : : #pragma once
18 : :
19 : : #include <algorithm>
20 : : #include <cstddef>
21 : : #include <cstdint>
22 : : #include <memory>
23 : : #include <mutex>
24 : : #include <unordered_map>
25 : : #include <vector>
26 : :
27 : : #include "wirestead/wrapper/send_accounting.hpp"
28 : :
29 : : namespace wirestead::diagnostics {
30 : :
31 : : // Server callers hold their session-map mutex while transferring or reading
32 : : // contributors. Each source is a coherent ledger snapshot.
33 : 972 : inline void accumulate_send_accounting(wrapper::SendAccounting& total, const wrapper::SendAccounting& source) {
34 : 10692 : const auto add = [](wrapper::SendRequestTotals& to, const wrapper::SendRequestTotals& from) {
35 : 10692 : to.requests += from.requests;
36 : 10692 : to.bytes += from.bytes;
37 : 10692 : };
38 : 3888 : const auto loss = [&](wrapper::SendLossTotals& to, const wrapper::SendLossTotals& from) {
39 : 3888 : add(to.discarded_before_write, from.discarded_before_write);
40 : 3888 : add(to.aborted_during_write, from.aborted_during_write);
41 : 4860 : };
42 : 972 : add(total.accepted, source.accepted);
43 : 972 : add(total.written, source.written);
44 : 972 : add(total.outstanding, source.outstanding);
45 : 972 : loss(total.explicit_stop, source.explicit_stop);
46 : 972 : loss(total.connection_loss, source.connection_loss);
47 : 972 : loss(total.queue_pressure, source.queue_pressure);
48 : 972 : loss(total.session_expiry, source.session_expiry);
49 : 972 : total.confirmed_written_bytes += source.confirmed_written_bytes;
50 : 972 : }
51 : :
52 : : // The admitting thread and the executor each take the ledger lock per message
53 : : // or batch, for a few dozen nanoseconds. Sleeping on contention costs a futex
54 : : // round trip on both sides, far longer than the wait itself.
55 : : class BriefMutex {
56 : : public:
57 : 753501 : void lock() {
58 [ + + ]: 773654 : for (int i = 0; i < 100; ++i)
59 [ + + ]: 773554 : if (mutex_.try_lock()) return;
60 : 100 : mutex_.lock();
61 : : }
62 : 753501 : void unlock() { mutex_.unlock(); }
63 : : bool try_lock() { return mutex_.try_lock(); }
64 : :
65 : : private:
66 : : std::mutex mutex_;
67 : : };
68 : :
69 : : // Shared by admission threads and the transport executor. IDs never repeat
70 : : // across reset; old completions cannot change a replacement measurement epoch.
71 : : class SendAccountingLedger {
72 : : public:
73 : : using Request = uint64_t;
74 : : enum class Cause { ExplicitStop, ConnectionLoss, QueuePressure, SessionExpiry };
75 : :
76 : : // A contributor belongs to this ledger's measurement epoch. Entries retain
77 : : // it while I/O is outstanding, even after a virtual session leaves its map.
78 : : struct Group {
79 : : wrapper::SendAccounting totals;
80 : : uint64_t epoch = 0;
81 : : };
82 : : using GroupHandle = std::shared_ptr<Group>;
83 : :
84 : 501825 : Request admit(size_t bytes, GroupHandle group = {}) {
85 : 501825 : std::lock_guard<BriefMutex> lock(mutex_);
86 : : // Commit the ID only once the entry is stored: a failed insert changes nothing.
87 : 501825 : const auto id = next_id_ + 1;
88 [ + + + + : 501825 : if (group && group->epoch != epoch_) {
+ + ]
89 : 129 : group->totals = {};
90 : 129 : group->epoch = epoch_;
91 : : }
92 [ + + + + ]: 501825 : if (count_ == slots_.size()) make_room();
93 : 501823 : auto& entry = slot(count_);
94 : 501823 : entry = Entry{bytes, epoch_, false, true, std::move(group)};
95 : 501823 : ++count_;
96 : 501823 : next_id_ = id;
97 : 501823 : update(entry, [&](auto& totals) {
98 : 511416 : add(totals.accepted, bytes);
99 : 511416 : add(totals.outstanding, bytes);
100 : 511416 : });
101 : 501823 : return id;
102 : 501825 : }
103 : :
104 : : // Roll back tracking if posting throws before the API returns acceptance.
105 : : class Admission {
106 : : public:
107 : 491503 : Admission(SendAccountingLedger& ledger, size_t bytes, GroupHandle group = {})
108 : 491503 : : ledger_(ledger), request_(ledger.admit(bytes, std::move(group))) {}
109 : : Admission(const Admission&) = delete;
110 : : Admission& operator=(const Admission&) = delete;
111 : 491503 : ~Admission() {
112 [ + + ]: 491503 : if (!committed_) ledger_.rollback(request_);
113 : 491503 : }
114 : 291357 : Request request() const { return request_; }
115 : 291358 : void commit() { committed_ = true; }
116 : :
117 : : private:
118 : : SendAccountingLedger& ledger_;
119 : : Request request_;
120 : : bool committed_ = false;
121 : : };
122 : :
123 : : // Called immediately before handing a request to local I/O.
124 : 4457 : bool begin(Request id) {
125 : 4457 : std::lock_guard<BriefMutex> lock(mutex_);
126 : 8914 : return begin_locked(id);
127 : 4457 : }
128 : :
129 : : // Projections only read the caller-owned batch. They must not re-enter this
130 : : // ledger. Returns false, activating nothing, if end() already terminated any
131 : : // request: the caller must not write the batch. This makes the handoff
132 : : // atomic with stop/loss without the caller's admission lock.
133 : : template <typename Range, typename RequestOf>
134 : 3728 : bool begin_batch(const Range& batch, RequestOf request_of) {
135 : 3728 : std::lock_guard<BriefMutex> lock(mutex_);
136 [ + + ]: 16367 : for (const auto& item : batch)
137 [ + - + + ]: 12745 : if (!find(request_of(item))) return false;
138 [ + - + + ]: 16259 : for (const auto& item : batch) begin_locked(request_of(item));
139 : 3622 : return true;
140 : 3728 : }
141 : :
142 : : // One composed-operation completion supplies the prefix length for each
143 : : // request. A fully covered request is written even if a later request fails.
144 : 5953 : void complete(Request id, size_t confirmed_bytes) {
145 : 5953 : std::lock_guard<BriefMutex> lock(mutex_);
146 : 5953 : complete_locked(id, confirmed_bytes);
147 : 5953 : }
148 : :
149 : : // Consume the wire prefix even for retired IDs or older measurement epochs.
150 : : // Release the ledger mutex before the caller handles connection loss/stop.
151 : : template <typename Range, typename RequestOf, typename SizeOf>
152 : 3476 : void complete_batch(const Range& batch, size_t confirmed_bytes, RequestOf request_of, SizeOf size_of) {
153 : 3476 : std::lock_guard<BriefMutex> lock(mutex_);
154 [ + + ]: 16198 : for (const auto& item : batch) {
155 : 12722 : const auto confirmed = std::min(confirmed_bytes, size_of(item));
156 : 12722 : complete_locked(request_of(item), confirmed);
157 : 12722 : confirmed_bytes -= confirmed;
158 : : }
159 : 3476 : }
160 : :
161 : 9 : void discard(Request id, Cause cause) {
162 : 9 : std::lock_guard<BriefMutex> lock(mutex_);
163 : 9 : auto* entry = find(id);
164 [ + + ]: 9 : if (!entry) return;
165 : 7 : terminate(*entry, cause);
166 : 7 : erase(id, *entry);
167 : 9 : }
168 : :
169 : 12543 : bool contains(Request id) const {
170 : 12543 : std::lock_guard<BriefMutex> lock(mutex_);
171 [ + + + - ]: 12543 : if (id < base_) return sparse_.contains(id);
172 [ + + + + ]: 12274 : return id - base_ < count_ && slot(id - base_).live;
173 : 12543 : }
174 : :
175 : : // Expiry races with handoff under the caller's admission mutex. Active
176 : : // operations keep their actual outcome; unrelated contributors are untouched.
177 : 8285 : void discard_waiting(const GroupHandle& group, Cause cause) {
178 [ - + ]: 8285 : if (!group) return;
179 : 8285 : std::lock_guard<BriefMutex> lock(mutex_);
180 [ + + ]: 16529 : for (auto it = sparse_.begin(); it != sparse_.end();) {
181 [ + + + + : 8244 : if (it->second.group == group && !it->second.active) {
+ + ]
182 : 1 : terminate(it->second, cause);
183 : 1 : it = sparse_.erase(it);
184 : : } else {
185 : 8243 : ++it;
186 : : }
187 : : }
188 [ + + ]: 16698 : for (size_t i = 0; i < count_; ++i) {
189 : 8413 : auto& entry = slot(i);
190 [ + + + + : 8413 : if (entry.live && entry.group == group && !entry.active) {
+ + + + ]
191 : 8274 : terminate(entry, cause);
192 : 8274 : entry.live = false;
193 : 8274 : entry.group.reset();
194 : : }
195 : : }
196 : 8285 : compact();
197 : 8285 : }
198 : :
199 : : // The caller serializes this boundary with admission. Clearing every entry
200 : : // includes requests accepted before their enqueue handler has executed.
201 : 5626 : void end(Cause cause) {
202 : 5626 : std::lock_guard<BriefMutex> lock(mutex_);
203 [ + - + + ]: 5628 : for (const auto& [id, entry] : sparse_) terminate(entry, cause);
204 : 5626 : sparse_.clear();
205 [ + + ]: 282131 : for (size_t i = 0; i < count_; ++i) {
206 [ + + + - ]: 276505 : if (slot(i).live) terminate(slot(i), cause);
207 : 276505 : slot(i) = Entry{};
208 : : }
209 : 5626 : head_ = 0;
210 : 5626 : count_ = 0;
211 : 5626 : base_ = next_id_ + 1;
212 : 5626 : }
213 : :
214 : 6828 : wrapper::SendAccounting snapshot() const {
215 : 6828 : std::lock_guard<BriefMutex> lock(mutex_);
216 : 6828 : return totals_;
217 : 6828 : }
218 : :
219 : 295 : wrapper::SendAccounting snapshot(const GroupHandle& group) const {
220 : 295 : std::lock_guard<BriefMutex> lock(mutex_);
221 [ + - + + ]: 590 : return group && group->epoch == epoch_ ? group->totals : wrapper::SendAccounting{};
222 : 295 : }
223 : :
224 : : // A measurement epoch includes only requests accepted since reset. Retained
225 : : // old requests still transmit but their later completions/cleanup are ignored.
226 : 331 : void reset() {
227 : 331 : std::lock_guard<BriefMutex> lock(mutex_);
228 : 331 : ++epoch_;
229 : 331 : totals_ = {};
230 : 331 : }
231 : :
232 : : private:
233 : : struct Entry {
234 : : size_t bytes = 0;
235 : : uint64_t epoch = 0;
236 : : bool active = false;
237 : : bool live = false;
238 : : GroupHandle group;
239 : : };
240 : :
241 : 2922356 : Entry& slot(size_t index) { return slots_[(head_ + index) & (slots_.size() - 1)]; }
242 : 12272 : const Entry& slot(size_t index) const { return slots_[(head_ + index) & (slots_.size() - 1)]; }
243 : :
244 : : // Grow only when more than half the slots are live. Otherwise detach the
245 : : // surviving older IDs before reusing the ring, so repeated rollback/expiry
246 : : // cannot grow storage behind a stalled request. Stage all allocations first:
247 : : // a failed promotion leaves both stores and every request unchanged.
248 : 2184 : void make_room() {
249 [ + + ]: 2184 : if (!slots_.empty()) {
250 : 117 : size_t live = 0;
251 [ + + ]: 547701 : for (size_t i = 0; i < count_; ++i) live += slot(i).live;
252 [ + + ]: 117 : if (live <= count_ / 2) {
253 : 7 : auto staged = sparse_;
254 : 6 : staged.reserve(staged.size() + live);
255 [ + + ]: 102 : for (size_t i = 0; i < count_; ++i)
256 [ + + + - ]: 96 : if (slot(i).live) staged.emplace(base_ + i, slot(i));
257 : 6 : sparse_.swap(staged);
258 [ + + ]: 102 : for (size_t i = 0; i < count_; ++i) slot(i) = Entry{};
259 : 6 : head_ = 0;
260 : 6 : count_ = 0;
261 : 6 : base_ = next_id_ + 1;
262 : 6 : return;
263 : 6 : }
264 : : }
265 [ + + + + ]: 2178 : std::vector<Entry> larger(slots_.empty() ? 16 : slots_.size() * 2);
266 [ + + ]: 549632 : for (size_t i = 0; i < count_; ++i) larger[i] = std::move(slot(i));
267 : 2176 : slots_ = std::move(larger);
268 : 2176 : head_ = 0;
269 : 2176 : }
270 : :
271 : 248668 : Entry* find(Request id) {
272 [ + + ]: 248668 : if (id < base_) {
273 : 900 : const auto it = sparse_.find(id);
274 [ + + ]: 900 : return it == sparse_.end() ? nullptr : &it->second;
275 : : }
276 [ - + ]: 247768 : if (id - base_ >= count_) return nullptr;
277 : 247768 : auto& entry = slot(id - base_);
278 [ + + ]: 247768 : return entry.live ? &entry : nullptr;
279 : : }
280 : :
281 : 218045 : void erase(Request id, Entry& entry) {
282 [ + + ]: 218045 : if (id < base_) {
283 : 6 : sparse_.erase(id);
284 : 6 : return;
285 : : }
286 : 218039 : entry.live = false;
287 : 218039 : entry.group.reset();
288 : 218039 : compact();
289 : : }
290 : :
291 : 226324 : void compact() {
292 [ + + + + : 451546 : while (count_ && !slot(0).live) {
+ + ]
293 : 225222 : head_ = (head_ + 1) & (slots_.size() - 1);
294 : 225222 : --count_;
295 : 225222 : ++base_;
296 : : }
297 : 226324 : }
298 : :
299 : 17094 : bool begin_locked(Request id) {
300 : 17094 : auto* entry = find(id);
301 [ + + ]: 17094 : if (!entry) return false;
302 : 17093 : entry->active = true;
303 : 17093 : return true;
304 : : }
305 : :
306 : 18675 : void complete_locked(Request id, size_t confirmed_bytes) {
307 : 18675 : auto* found = find(id);
308 [ + + ]: 18675 : if (!found) return;
309 : 17893 : const auto& entry = *found;
310 : 17893 : update(entry, [&](auto& totals) {
311 : 17903 : const auto confirmed = std::min(confirmed_bytes, entry.bytes);
312 : 17903 : totals.confirmed_written_bytes += confirmed;
313 : 17903 : remove(totals.outstanding, entry.bytes);
314 [ + + ]: 17903 : if (confirmed == entry.bytes) {
315 : 17606 : add(totals.written, entry.bytes);
316 : : } else {
317 : 297 : add(totals.connection_loss.aborted_during_write, entry.bytes);
318 : : }
319 : 17903 : });
320 : 17893 : erase(id, *found);
321 : : }
322 : :
323 : 200145 : void rollback(Request id) {
324 : 200145 : std::lock_guard<BriefMutex> lock(mutex_);
325 : 200145 : auto* entry = find(id);
326 [ - + ]: 200145 : if (!entry) return;
327 : 200145 : update(*entry, [&](auto& totals) {
328 : 200146 : remove(totals.accepted, entry->bytes);
329 : 200146 : remove(totals.outstanding, entry->bytes);
330 : 200146 : });
331 : 200145 : erase(id, *entry);
332 : 200145 : }
333 : : template <typename F>
334 : 1003646 : void update(const Entry& entry, F&& apply) {
335 [ + + ]: 1003646 : if (entry.epoch != epoch_) return;
336 : 1003460 : apply(totals_);
337 [ + + ]: 1003460 : if (entry.group) apply(entry.group->totals);
338 : : }
339 : 1333909 : static void add(wrapper::SendRequestTotals& totals, size_t bytes) {
340 : 1333909 : ++totals.requests;
341 : 1333909 : totals.bytes += bytes;
342 : 1333909 : }
343 : 711369 : static void remove(wrapper::SendRequestTotals& totals, size_t bytes) {
344 : 711369 : --totals.requests;
345 : 711369 : totals.bytes -= bytes;
346 : 711369 : }
347 : 283785 : void terminate(const Entry& entry, Cause cause) {
348 : 283785 : update(entry, [&](auto& totals) {
349 [ + + ]: 310402 : auto& loss = cause == Cause::ExplicitStop ? totals.explicit_stop
350 [ + + ]: 17228 : : cause == Cause::ConnectionLoss ? totals.connection_loss
351 [ + + ]: 16556 : : cause == Cause::SessionExpiry ? totals.session_expiry
352 : : : totals.queue_pressure;
353 [ + + ]: 293174 : add(entry.active ? loss.aborted_during_write : loss.discarded_before_write, entry.bytes);
354 : 293174 : remove(totals.outstanding, entry.bytes);
355 : 293174 : });
356 : 283785 : }
357 : :
358 : : mutable BriefMutex mutex_;
359 : : // slot(i) is request base_ + i in a power-of-two ring. IDs are assigned in
360 : : // order and transports retire them nearly in order, so lookup needs no
361 : : // hashing and steady traffic reuses slots without allocating. Retired holes
362 : : // trigger sparse promotion before growing a mostly empty ring. Ring capacity
363 : : // is bounded by max(16, 4 * peak live requests), independent of retired IDs.
364 : : std::vector<Entry> slots_;
365 : : // Only detached live IDs precede base_; FIFO traffic never inserts here.
366 : : std::unordered_map<Request, Entry> sparse_;
367 : : size_t head_ = 0;
368 : : size_t count_ = 0;
369 : : Request base_ = 1;
370 : : Request next_id_ = 0;
371 : : uint64_t epoch_ = 0;
372 : : wrapper::SendAccounting totals_;
373 : : };
374 : :
375 : : } // namespace wirestead::diagnostics
|
