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 <atomic>
20 : : #include <chrono>
21 : : #include <cstddef>
22 : : #include <cstdint>
23 : :
24 : : #include "wirestead/diagnostics/logger.hpp"
25 : : #include "wirestead/wrapper/runtime_stats.hpp"
26 : :
27 : : namespace wirestead {
28 : : namespace diagnostics {
29 : :
30 : : struct RuntimeStatsCounters {
31 : : std::atomic<uint64_t> bytes_accepted{0};
32 : : std::atomic<uint64_t> messages_accepted{0};
33 : :
34 : : std::atomic<uint64_t> bytes_sent{0};
35 : : std::atomic<uint64_t> messages_sent{0};
36 : :
37 : : std::atomic<uint64_t> bytes_received{0};
38 : : std::atomic<uint64_t> messages_received{0};
39 : : // Steady-clock nanoseconds at the last receive; 0 means nothing has arrived.
40 : : std::atomic<int64_t> last_receive_ns{0};
41 : :
42 : : std::atomic<uint64_t> failed_sends{0};
43 : :
44 : : std::atomic<uint64_t> dropped_messages{0};
45 : : std::atomic<uint64_t> dropped_bytes{0};
46 : :
47 : : std::atomic<uint64_t> backpressure_events{0};
48 : :
49 : : std::atomic<size_t> max_queued_bytes{0};
50 : :
51 : : // Latched, deliberately not cleared by reset() - the warning below is meant
52 : : // to fire once in a process's life, not once per measurement window.
53 : : std::atomic<bool> drop_warned{false};
54 : :
55 : 429360 : void record_accepted(size_t bytes) {
56 : 429360 : messages_accepted.fetch_add(1, std::memory_order_relaxed);
57 : 429360 : bytes_accepted.fetch_add(bytes, std::memory_order_relaxed);
58 : 429360 : }
59 : :
60 : 1553 : void record_sent(size_t bytes) {
61 : 1553 : messages_sent.fetch_add(1, std::memory_order_relaxed);
62 : 1553 : bytes_sent.fetch_add(bytes, std::memory_order_relaxed);
63 : 1553 : }
64 : :
65 : 2944 : void record_received(size_t bytes) {
66 : 2944 : messages_received.fetch_add(1, std::memory_order_relaxed);
67 : 2944 : bytes_received.fetch_add(bytes, std::memory_order_relaxed);
68 : : // Steady rather than system clock: this is only ever read as an age, and
69 : : // an age computed across a clock step is worse than no age at all.
70 : 2944 : last_receive_ns.store(steady_now_ns(), std::memory_order_relaxed);
71 : 2944 : }
72 : :
73 : 88573 : void record_failed_send() { failed_sends.fetch_add(1, std::memory_order_relaxed); }
74 : :
75 : 3028 : static int64_t steady_now_ns() {
76 : 6056 : return std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::steady_clock::now().time_since_epoch())
77 : 3028 : .count();
78 : : }
79 : :
80 : : // Every drop in the library routes through here. BestEffort discards by
81 : : // design and can do it hundreds of thousands of times a second, so the
82 : : // running total is the counter to watch - but a caller who never looks at it
83 : : // would otherwise lose data in complete silence, which is what makes the
84 : : // first one worth saying out loud exactly once.
85 : 24 : void record_dropped(size_t messages, size_t bytes) {
86 : 24 : dropped_messages.fetch_add(static_cast<uint64_t>(messages), std::memory_order_relaxed);
87 : 24 : dropped_bytes.fetch_add(static_cast<uint64_t>(bytes), std::memory_order_relaxed);
88 [ + + ]: 24 : if (!drop_warned.exchange(true, std::memory_order_relaxed)) {
89 : 13 : WIRESTEAD_LOG_WARNING("runtime_stats", "record_dropped",
90 : : "Dropping queued data (BestEffort). Watch RuntimeStats::dropped_bytes - "
91 : : "on_backpressure() is not a reliable loss signal under this strategy.");
92 : : }
93 : 24 : }
94 : :
95 : 139 : void record_backpressure_event() { backpressure_events.fetch_add(1, std::memory_order_relaxed); }
96 : :
97 : 701588 : void observe_queue(size_t queued) {
98 : 701588 : size_t current = max_queued_bytes.load(std::memory_order_relaxed);
99 [ + + - + : 1069364 : while (queued > current && !max_queued_bytes.compare_exchange_weak(current, queued, std::memory_order_relaxed,
- + ]
100 : : std::memory_order_relaxed)) {
101 : : }
102 : 701588 : }
103 : :
104 : : // Folds a finished contributor's totals in, so a server's counters outlive
105 : : // the session that produced them. The cumulative fields add; max_queued_bytes
106 : : // takes the deeper of the two peaks, matching how a server aggregates it
107 : : // across live sessions.
108 : : //
109 : : // queued_bytes, pending_bytes and backpressure_active are deliberately left
110 : : // out: they describe a live queue, and a contributor being absorbed no longer
111 : : // has one.
112 : 132 : void absorb(const wrapper::RuntimeStats& other) {
113 : 132 : bytes_accepted.fetch_add(other.bytes_accepted, std::memory_order_relaxed);
114 : 132 : messages_accepted.fetch_add(other.messages_accepted, std::memory_order_relaxed);
115 : 132 : bytes_sent.fetch_add(other.bytes_sent, std::memory_order_relaxed);
116 : 132 : messages_sent.fetch_add(other.messages_sent, std::memory_order_relaxed);
117 : 132 : bytes_received.fetch_add(other.bytes_received, std::memory_order_relaxed);
118 : 132 : messages_received.fetch_add(other.messages_received, std::memory_order_relaxed);
119 : 132 : failed_sends.fetch_add(other.failed_sends, std::memory_order_relaxed);
120 : 132 : dropped_messages.fetch_add(other.dropped_messages, std::memory_order_relaxed);
121 : 132 : dropped_bytes.fetch_add(other.dropped_bytes, std::memory_order_relaxed);
122 : 132 : backpressure_events.fetch_add(other.backpressure_events, std::memory_order_relaxed);
123 : 132 : observe_queue(other.max_queued_bytes);
124 : 132 : }
125 : :
126 : 4032 : wrapper::RuntimeStats snapshot(size_t queued_bytes, size_t pending_bytes, bool backpressure_active) const {
127 : 4032 : wrapper::RuntimeStats stats;
128 : 4032 : stats.bytes_accepted = bytes_accepted.load(std::memory_order_relaxed);
129 : 4032 : stats.messages_accepted = messages_accepted.load(std::memory_order_relaxed);
130 : 4032 : stats.bytes_sent = bytes_sent.load(std::memory_order_relaxed);
131 : 4032 : stats.messages_sent = messages_sent.load(std::memory_order_relaxed);
132 : 4032 : stats.bytes_received = bytes_received.load(std::memory_order_relaxed);
133 : 4032 : stats.messages_received = messages_received.load(std::memory_order_relaxed);
134 : 4032 : stats.failed_sends = failed_sends.load(std::memory_order_relaxed);
135 : 4032 : stats.dropped_messages = dropped_messages.load(std::memory_order_relaxed);
136 : 4032 : stats.dropped_bytes = dropped_bytes.load(std::memory_order_relaxed);
137 : 4032 : stats.backpressure_events = backpressure_events.load(std::memory_order_relaxed);
138 : 4032 : stats.queued_bytes = queued_bytes;
139 : 4032 : stats.pending_bytes = pending_bytes;
140 : 4032 : stats.max_queued_bytes = max_queued_bytes.load(std::memory_order_relaxed);
141 : 4032 : stats.backpressure_active = backpressure_active;
142 : 4032 : const int64_t last = last_receive_ns.load(std::memory_order_relaxed);
143 [ + + ]: 4032 : if (last != 0) {
144 : 84 : const int64_t age = steady_now_ns() - last;
145 [ + - ]: 84 : stats.last_receive_age_ms = static_cast<uint64_t>(age > 0 ? age / 1'000'000 : 0);
146 : : }
147 : 4032 : return stats;
148 : : }
149 : :
150 : 177 : void reset(size_t current_queued_bytes) {
151 : 177 : bytes_accepted.store(0, std::memory_order_relaxed);
152 : 177 : messages_accepted.store(0, std::memory_order_relaxed);
153 : 177 : bytes_sent.store(0, std::memory_order_relaxed);
154 : 177 : messages_sent.store(0, std::memory_order_relaxed);
155 : 177 : bytes_received.store(0, std::memory_order_relaxed);
156 : 177 : messages_received.store(0, std::memory_order_relaxed);
157 : 177 : last_receive_ns.store(0, std::memory_order_relaxed);
158 : 177 : failed_sends.store(0, std::memory_order_relaxed);
159 : 177 : dropped_messages.store(0, std::memory_order_relaxed);
160 : 177 : dropped_bytes.store(0, std::memory_order_relaxed);
161 : 177 : backpressure_events.store(0, std::memory_order_relaxed);
162 : 177 : max_queued_bytes.store(current_queued_bytes, std::memory_order_relaxed);
163 : 177 : }
164 : : };
165 : :
166 : : } // namespace diagnostics
167 : : } // namespace wirestead
|
