1use std::sync::Arc;
2use std::sync::atomic::AtomicU64;
3use std::sync::atomic::Ordering;
4
5use ursula_shard::BucketStreamId;
6use ursula_shard::CoreId;
7use ursula_shard::RaftGroupId;
8use ursula_shard::ShardPlacement;
9use ursula_stream::StreamErrorCode;
10use ursula_stream::StreamErrorContext;
11
12use crate::engine::GroupEngine;
13use crate::engine::GroupEngineError;
14use crate::error::RuntimeError;
15use crate::request::AppendBatchRequest;
16use crate::request::ColdWriteAdmission;
17use crate::rt::time::Instant;
18
19pub(crate) const GROUP_ACTOR_MAX_WRITE_BATCH: usize = 64;
20pub(crate) const COLD_FLUSH_GROUP_BATCH_MAX_CHUNKS: usize = 64;
21
22#[derive(Debug, Clone)]
23pub struct RuntimeMetrics {
24 pub(crate) inner: Arc<RuntimeMetricsInner>,
25}
26
27impl RuntimeMetrics {
28 pub fn snapshot(&self) -> RuntimeMetricsSnapshot {
29 let per_core_appends = self
30 .inner
31 .per_core_appends
32 .iter()
33 .map(PaddedAtomicU64::load_relaxed)
34 .collect::<Vec<_>>();
35 let accepted_appends = per_core_appends.iter().sum();
36 let per_group_appends = self
37 .inner
38 .per_group_appends
39 .iter()
40 .map(PaddedAtomicU64::load_relaxed)
41 .collect();
42 let per_core_applied_mutations = self
43 .inner
44 .per_core_applied_mutations
45 .iter()
46 .map(PaddedAtomicU64::load_relaxed)
47 .collect::<Vec<_>>();
48 let applied_mutations = per_core_applied_mutations.iter().sum();
49 let per_group_applied_mutations = self
50 .inner
51 .per_group_applied_mutations
52 .iter()
53 .map(PaddedAtomicU64::load_relaxed)
54 .collect();
55 let per_core_mutation_apply_ns = self
56 .inner
57 .per_core_mutation_apply_ns
58 .iter()
59 .map(PaddedAtomicU64::load_relaxed)
60 .collect::<Vec<_>>();
61 let mutation_apply_ns = per_core_mutation_apply_ns.iter().sum();
62 let per_group_mutation_apply_ns = self
63 .inner
64 .per_group_mutation_apply_ns
65 .iter()
66 .map(PaddedAtomicU64::load_relaxed)
67 .collect();
68 let per_core_group_lock_wait_ns = self
69 .inner
70 .per_core_group_lock_wait_ns
71 .iter()
72 .map(PaddedAtomicU64::load_relaxed)
73 .collect::<Vec<_>>();
74 let group_lock_wait_ns = per_core_group_lock_wait_ns.iter().sum();
75 let per_group_group_lock_wait_ns = self
76 .inner
77 .per_group_group_lock_wait_ns
78 .iter()
79 .map(PaddedAtomicU64::load_relaxed)
80 .collect();
81 let per_core_group_engine_exec_ns = self
82 .inner
83 .per_core_group_engine_exec_ns
84 .iter()
85 .map(PaddedAtomicU64::load_relaxed)
86 .collect::<Vec<_>>();
87 let group_engine_exec_ns = per_core_group_engine_exec_ns.iter().sum();
88 let per_group_group_engine_exec_ns = self
89 .inner
90 .per_group_group_engine_exec_ns
91 .iter()
92 .map(PaddedAtomicU64::load_relaxed)
93 .collect();
94 let per_group_group_mailbox_depth = self
95 .inner
96 .per_group_group_mailbox_depth
97 .iter()
98 .map(PaddedAtomicU64::load_relaxed)
99 .collect::<Vec<_>>();
100 let group_mailbox_depth = per_group_group_mailbox_depth.iter().sum();
101 let per_group_group_mailbox_max_depth = self
102 .inner
103 .per_group_group_mailbox_max_depth
104 .iter()
105 .map(PaddedAtomicU64::load_relaxed)
106 .collect::<Vec<_>>();
107 let group_mailbox_max_depth = per_group_group_mailbox_max_depth
108 .iter()
109 .copied()
110 .max()
111 .unwrap_or(0);
112 let per_group_group_mailbox_full_events = self
113 .inner
114 .per_group_group_mailbox_full_events
115 .iter()
116 .map(PaddedAtomicU64::load_relaxed)
117 .collect::<Vec<_>>();
118 let group_mailbox_full_events = per_group_group_mailbox_full_events.iter().sum();
119 let per_core_raft_write_many_batches = self
120 .inner
121 .per_core_raft_write_many_batches
122 .iter()
123 .map(PaddedAtomicU64::load_relaxed)
124 .collect::<Vec<_>>();
125 let raft_write_many_batches = per_core_raft_write_many_batches.iter().sum();
126 let per_group_raft_write_many_batches = self
127 .inner
128 .per_group_raft_write_many_batches
129 .iter()
130 .map(PaddedAtomicU64::load_relaxed)
131 .collect();
132 let per_core_raft_write_many_commands = self
133 .inner
134 .per_core_raft_write_many_commands
135 .iter()
136 .map(PaddedAtomicU64::load_relaxed)
137 .collect::<Vec<_>>();
138 let raft_write_many_commands = per_core_raft_write_many_commands.iter().sum();
139 let per_group_raft_write_many_commands = self
140 .inner
141 .per_group_raft_write_many_commands
142 .iter()
143 .map(PaddedAtomicU64::load_relaxed)
144 .collect();
145 let per_core_raft_write_many_logical_commands = self
146 .inner
147 .per_core_raft_write_many_logical_commands
148 .iter()
149 .map(PaddedAtomicU64::load_relaxed)
150 .collect::<Vec<_>>();
151 let raft_write_many_logical_commands =
152 per_core_raft_write_many_logical_commands.iter().sum();
153 let per_group_raft_write_many_logical_commands = self
154 .inner
155 .per_group_raft_write_many_logical_commands
156 .iter()
157 .map(PaddedAtomicU64::load_relaxed)
158 .collect();
159 let per_core_raft_write_many_responses = self
160 .inner
161 .per_core_raft_write_many_responses
162 .iter()
163 .map(PaddedAtomicU64::load_relaxed)
164 .collect::<Vec<_>>();
165 let raft_write_many_responses = per_core_raft_write_many_responses.iter().sum();
166 let per_group_raft_write_many_responses = self
167 .inner
168 .per_group_raft_write_many_responses
169 .iter()
170 .map(PaddedAtomicU64::load_relaxed)
171 .collect();
172 let per_core_raft_write_many_submit_ns = self
173 .inner
174 .per_core_raft_write_many_submit_ns
175 .iter()
176 .map(PaddedAtomicU64::load_relaxed)
177 .collect::<Vec<_>>();
178 let raft_write_many_submit_ns = per_core_raft_write_many_submit_ns.iter().sum();
179 let per_group_raft_write_many_submit_ns = self
180 .inner
181 .per_group_raft_write_many_submit_ns
182 .iter()
183 .map(PaddedAtomicU64::load_relaxed)
184 .collect();
185 let per_core_raft_write_many_response_ns = self
186 .inner
187 .per_core_raft_write_many_response_ns
188 .iter()
189 .map(PaddedAtomicU64::load_relaxed)
190 .collect::<Vec<_>>();
191 let raft_write_many_response_ns = per_core_raft_write_many_response_ns.iter().sum();
192 let per_group_raft_write_many_response_ns = self
193 .inner
194 .per_group_raft_write_many_response_ns
195 .iter()
196 .map(PaddedAtomicU64::load_relaxed)
197 .collect();
198 let per_core_raft_apply_entries = self
199 .inner
200 .per_core_raft_apply_entries
201 .iter()
202 .map(PaddedAtomicU64::load_relaxed)
203 .collect::<Vec<_>>();
204 let raft_apply_entries = per_core_raft_apply_entries.iter().sum();
205 let per_group_raft_apply_entries = self
206 .inner
207 .per_group_raft_apply_entries
208 .iter()
209 .map(PaddedAtomicU64::load_relaxed)
210 .collect();
211 let per_core_raft_apply_ns = self
212 .inner
213 .per_core_raft_apply_ns
214 .iter()
215 .map(PaddedAtomicU64::load_relaxed)
216 .collect::<Vec<_>>();
217 let raft_apply_ns = per_core_raft_apply_ns.iter().sum();
218 let per_group_raft_apply_ns = self
219 .inner
220 .per_group_raft_apply_ns
221 .iter()
222 .map(PaddedAtomicU64::load_relaxed)
223 .collect();
224 let per_core_live_read_waiters = self
225 .inner
226 .per_core_live_read_waiters
227 .iter()
228 .map(PaddedAtomicU64::load_relaxed)
229 .collect::<Vec<_>>();
230 let live_read_waiters = per_core_live_read_waiters.iter().sum();
231 let per_core_live_read_backpressure_events = self
232 .inner
233 .per_core_live_read_backpressure_events
234 .iter()
235 .map(PaddedAtomicU64::load_relaxed)
236 .collect::<Vec<_>>();
237 let live_read_backpressure_events = per_core_live_read_backpressure_events.iter().sum();
238 let per_core_routed_requests = self
239 .inner
240 .per_core_routed_requests
241 .iter()
242 .map(PaddedAtomicU64::load_relaxed)
243 .collect::<Vec<_>>();
244 let routed_requests = per_core_routed_requests.iter().sum();
245 let per_core_mailbox_send_wait_ns = self
246 .inner
247 .per_core_mailbox_send_wait_ns
248 .iter()
249 .map(PaddedAtomicU64::load_relaxed)
250 .collect::<Vec<_>>();
251 let mailbox_send_wait_ns = per_core_mailbox_send_wait_ns.iter().sum();
252 let per_core_mailbox_full_events = self
253 .inner
254 .per_core_mailbox_full_events
255 .iter()
256 .map(PaddedAtomicU64::load_relaxed)
257 .collect::<Vec<_>>();
258 let mailbox_full_events = per_core_mailbox_full_events.iter().sum();
259 let per_core_wal_batches = self
260 .inner
261 .per_core_wal_batches
262 .iter()
263 .map(PaddedAtomicU64::load_relaxed)
264 .collect::<Vec<_>>();
265 let wal_batches = per_core_wal_batches.iter().sum();
266 let per_group_wal_batches = self
267 .inner
268 .per_group_wal_batches
269 .iter()
270 .map(PaddedAtomicU64::load_relaxed)
271 .collect();
272 let per_core_wal_records = self
273 .inner
274 .per_core_wal_records
275 .iter()
276 .map(PaddedAtomicU64::load_relaxed)
277 .collect::<Vec<_>>();
278 let wal_records = per_core_wal_records.iter().sum();
279 let per_group_wal_records = self
280 .inner
281 .per_group_wal_records
282 .iter()
283 .map(PaddedAtomicU64::load_relaxed)
284 .collect();
285 let per_core_wal_write_ns = self
286 .inner
287 .per_core_wal_write_ns
288 .iter()
289 .map(PaddedAtomicU64::load_relaxed)
290 .collect::<Vec<_>>();
291 let wal_write_ns = per_core_wal_write_ns.iter().sum();
292 let per_group_wal_write_ns = self
293 .inner
294 .per_group_wal_write_ns
295 .iter()
296 .map(PaddedAtomicU64::load_relaxed)
297 .collect();
298 let per_core_wal_sync_ns = self
299 .inner
300 .per_core_wal_sync_ns
301 .iter()
302 .map(PaddedAtomicU64::load_relaxed)
303 .collect::<Vec<_>>();
304 let wal_sync_ns = per_core_wal_sync_ns.iter().sum();
305 let per_group_wal_sync_ns = self
306 .inner
307 .per_group_wal_sync_ns
308 .iter()
309 .map(PaddedAtomicU64::load_relaxed)
310 .collect();
311 let cold_flush_uploads = self.inner.cold_flush_uploads.load_relaxed();
312 let cold_flush_upload_bytes = self.inner.cold_flush_upload_bytes.load_relaxed();
313 let cold_flush_upload_ns = self.inner.cold_flush_upload_ns.load_relaxed();
314 let cold_flush_publishes = self.inner.cold_flush_publishes.load_relaxed();
315 let cold_flush_publish_bytes = self.inner.cold_flush_publish_bytes.load_relaxed();
316 let cold_flush_publish_ns = self.inner.cold_flush_publish_ns.load_relaxed();
317 let cold_orphan_cleanup_attempts = self.inner.cold_orphan_cleanup_attempts.load_relaxed();
318 let cold_orphan_cleanup_errors = self.inner.cold_orphan_cleanup_errors.load_relaxed();
319 let cold_orphan_bytes = self.inner.cold_orphan_bytes.load_relaxed();
320 let cold_gc_reclaimed = self.inner.cold_gc_reclaimed.load_relaxed();
321 let cold_gc_errors = self.inner.cold_gc_errors.load_relaxed();
322 let cold_flush_write_errors = self.inner.cold_flush_write_errors.load_relaxed();
323 let per_group_cold_hot_bytes = self
324 .inner
325 .per_group_cold_hot_bytes
326 .iter()
327 .map(PaddedAtomicU64::load_relaxed)
328 .collect::<Vec<_>>();
329 let cold_hot_bytes = per_group_cold_hot_bytes.iter().sum();
330 let per_group_cold_hot_bytes_max = self
331 .inner
332 .per_group_cold_hot_bytes_max
333 .iter()
334 .map(PaddedAtomicU64::load_relaxed)
335 .collect::<Vec<_>>();
336 let cold_hot_group_bytes_max = per_group_cold_hot_bytes_max
337 .iter()
338 .copied()
339 .max()
340 .unwrap_or(0);
341 let cold_hot_stream_bytes_max = self.inner.cold_hot_stream_bytes_max.load_relaxed();
342 let per_core_cold_backpressure_events = self
343 .inner
344 .per_core_cold_backpressure_events
345 .iter()
346 .map(PaddedAtomicU64::load_relaxed)
347 .collect::<Vec<_>>();
348 let cold_backpressure_events = per_core_cold_backpressure_events.iter().sum();
349 let per_group_cold_backpressure_events = self
350 .inner
351 .per_group_cold_backpressure_events
352 .iter()
353 .map(PaddedAtomicU64::load_relaxed)
354 .collect();
355 let cold_backpressure_bytes = self.inner.cold_backpressure_bytes.load_relaxed();
356
357 RuntimeMetricsSnapshot {
358 accepted_appends,
359 per_core_appends,
360 per_group_appends,
361 applied_mutations,
362 per_core_applied_mutations,
363 per_group_applied_mutations,
364 mutation_apply_ns,
365 per_core_mutation_apply_ns,
366 per_group_mutation_apply_ns,
367 group_lock_wait_ns,
368 per_core_group_lock_wait_ns,
369 per_group_group_lock_wait_ns,
370 group_engine_exec_ns,
371 per_core_group_engine_exec_ns,
372 per_group_group_engine_exec_ns,
373 group_mailbox_depth,
374 per_group_group_mailbox_depth,
375 group_mailbox_max_depth,
376 per_group_group_mailbox_max_depth,
377 group_mailbox_full_events,
378 per_group_group_mailbox_full_events,
379 raft_write_many_batches,
380 per_core_raft_write_many_batches,
381 per_group_raft_write_many_batches,
382 raft_write_many_commands,
383 per_core_raft_write_many_commands,
384 per_group_raft_write_many_commands,
385 raft_write_many_logical_commands,
386 per_core_raft_write_many_logical_commands,
387 per_group_raft_write_many_logical_commands,
388 raft_write_many_responses,
389 per_core_raft_write_many_responses,
390 per_group_raft_write_many_responses,
391 raft_write_many_submit_ns,
392 per_core_raft_write_many_submit_ns,
393 per_group_raft_write_many_submit_ns,
394 raft_write_many_response_ns,
395 per_core_raft_write_many_response_ns,
396 per_group_raft_write_many_response_ns,
397 raft_apply_entries,
398 per_core_raft_apply_entries,
399 per_group_raft_apply_entries,
400 raft_apply_ns,
401 per_core_raft_apply_ns,
402 per_group_raft_apply_ns,
403 live_read_waiters,
404 per_core_live_read_waiters,
405 live_read_backpressure_events,
406 per_core_live_read_backpressure_events,
407 routed_requests,
408 per_core_routed_requests,
409 mailbox_send_wait_ns,
410 per_core_mailbox_send_wait_ns,
411 mailbox_full_events,
412 per_core_mailbox_full_events,
413 wal_batches,
414 per_core_wal_batches,
415 per_group_wal_batches,
416 wal_records,
417 per_core_wal_records,
418 per_group_wal_records,
419 wal_write_ns,
420 per_core_wal_write_ns,
421 per_group_wal_write_ns,
422 wal_sync_ns,
423 per_core_wal_sync_ns,
424 per_group_wal_sync_ns,
425 cold_flush_uploads,
426 cold_flush_upload_bytes,
427 cold_flush_upload_ns,
428 cold_flush_publishes,
429 cold_flush_publish_bytes,
430 cold_flush_publish_ns,
431 cold_orphan_cleanup_attempts,
432 cold_orphan_cleanup_errors,
433 cold_gc_reclaimed,
434 cold_gc_errors,
435 cold_flush_write_errors,
436 cold_orphan_bytes,
437 cold_hot_bytes,
438 per_group_cold_hot_bytes,
439 cold_hot_group_bytes_max,
440 per_group_cold_hot_bytes_max,
441 cold_hot_stream_bytes_max,
442 cold_backpressure_events,
443 per_core_cold_backpressure_events,
444 per_group_cold_backpressure_events,
445 cold_backpressure_bytes,
446 }
447 }
448}
449
450#[derive(Debug, Clone, PartialEq, Eq)]
451pub struct RuntimeMetricsSnapshot {
452 pub accepted_appends: u64,
453 pub per_core_appends: Vec<u64>,
454 pub per_group_appends: Vec<u64>,
455 pub applied_mutations: u64,
456 pub per_core_applied_mutations: Vec<u64>,
457 pub per_group_applied_mutations: Vec<u64>,
458 pub mutation_apply_ns: u64,
459 pub per_core_mutation_apply_ns: Vec<u64>,
460 pub per_group_mutation_apply_ns: Vec<u64>,
461 pub group_lock_wait_ns: u64,
462 pub per_core_group_lock_wait_ns: Vec<u64>,
463 pub per_group_group_lock_wait_ns: Vec<u64>,
464 pub group_engine_exec_ns: u64,
465 pub per_core_group_engine_exec_ns: Vec<u64>,
466 pub per_group_group_engine_exec_ns: Vec<u64>,
467 pub group_mailbox_depth: u64,
468 pub per_group_group_mailbox_depth: Vec<u64>,
469 pub group_mailbox_max_depth: u64,
470 pub per_group_group_mailbox_max_depth: Vec<u64>,
471 pub group_mailbox_full_events: u64,
472 pub per_group_group_mailbox_full_events: Vec<u64>,
473 pub raft_write_many_batches: u64,
474 pub per_core_raft_write_many_batches: Vec<u64>,
475 pub per_group_raft_write_many_batches: Vec<u64>,
476 pub raft_write_many_commands: u64,
477 pub per_core_raft_write_many_commands: Vec<u64>,
478 pub per_group_raft_write_many_commands: Vec<u64>,
479 pub raft_write_many_logical_commands: u64,
480 pub per_core_raft_write_many_logical_commands: Vec<u64>,
481 pub per_group_raft_write_many_logical_commands: Vec<u64>,
482 pub raft_write_many_responses: u64,
483 pub per_core_raft_write_many_responses: Vec<u64>,
484 pub per_group_raft_write_many_responses: Vec<u64>,
485 pub raft_write_many_submit_ns: u64,
486 pub per_core_raft_write_many_submit_ns: Vec<u64>,
487 pub per_group_raft_write_many_submit_ns: Vec<u64>,
488 pub raft_write_many_response_ns: u64,
489 pub per_core_raft_write_many_response_ns: Vec<u64>,
490 pub per_group_raft_write_many_response_ns: Vec<u64>,
491 pub raft_apply_entries: u64,
492 pub per_core_raft_apply_entries: Vec<u64>,
493 pub per_group_raft_apply_entries: Vec<u64>,
494 pub raft_apply_ns: u64,
495 pub per_core_raft_apply_ns: Vec<u64>,
496 pub per_group_raft_apply_ns: Vec<u64>,
497 pub live_read_waiters: u64,
498 pub per_core_live_read_waiters: Vec<u64>,
499 pub live_read_backpressure_events: u64,
500 pub per_core_live_read_backpressure_events: Vec<u64>,
501 pub routed_requests: u64,
502 pub per_core_routed_requests: Vec<u64>,
503 pub mailbox_send_wait_ns: u64,
504 pub per_core_mailbox_send_wait_ns: Vec<u64>,
505 pub mailbox_full_events: u64,
506 pub per_core_mailbox_full_events: Vec<u64>,
507 pub wal_batches: u64,
508 pub per_core_wal_batches: Vec<u64>,
509 pub per_group_wal_batches: Vec<u64>,
510 pub wal_records: u64,
511 pub per_core_wal_records: Vec<u64>,
512 pub per_group_wal_records: Vec<u64>,
513 pub wal_write_ns: u64,
514 pub per_core_wal_write_ns: Vec<u64>,
515 pub per_group_wal_write_ns: Vec<u64>,
516 pub wal_sync_ns: u64,
517 pub per_core_wal_sync_ns: Vec<u64>,
518 pub per_group_wal_sync_ns: Vec<u64>,
519 pub cold_flush_uploads: u64,
520 pub cold_flush_upload_bytes: u64,
521 pub cold_flush_upload_ns: u64,
522 pub cold_flush_publishes: u64,
523 pub cold_flush_publish_bytes: u64,
524 pub cold_flush_publish_ns: u64,
525 pub cold_orphan_cleanup_attempts: u64,
526 pub cold_orphan_cleanup_errors: u64,
527 pub cold_orphan_bytes: u64,
528 pub cold_gc_reclaimed: u64,
529 pub cold_gc_errors: u64,
530 pub cold_flush_write_errors: u64,
531 pub cold_hot_bytes: u64,
532 pub per_group_cold_hot_bytes: Vec<u64>,
533 pub cold_hot_group_bytes_max: u64,
534 pub per_group_cold_hot_bytes_max: Vec<u64>,
535 pub cold_hot_stream_bytes_max: u64,
536 pub cold_backpressure_events: u64,
537 pub per_core_cold_backpressure_events: Vec<u64>,
538 pub per_group_cold_backpressure_events: Vec<u64>,
539 pub cold_backpressure_bytes: u64,
540}
541
542#[derive(Debug, Clone, PartialEq, Eq)]
543pub struct RuntimeMailboxSnapshot {
544 pub depths: Vec<usize>,
545 pub capacities: Vec<usize>,
546}
547
548#[derive(Debug)]
549pub(crate) struct RuntimeMetricsInner {
550 pub(crate) per_core_appends: Vec<PaddedAtomicU64>,
551 pub(crate) per_group_appends: Vec<PaddedAtomicU64>,
552 pub(crate) per_core_applied_mutations: Vec<PaddedAtomicU64>,
553 pub(crate) per_group_applied_mutations: Vec<PaddedAtomicU64>,
554 pub(crate) per_core_mutation_apply_ns: Vec<PaddedAtomicU64>,
555 pub(crate) per_group_mutation_apply_ns: Vec<PaddedAtomicU64>,
556 pub(crate) per_core_group_lock_wait_ns: Vec<PaddedAtomicU64>,
557 pub(crate) per_group_group_lock_wait_ns: Vec<PaddedAtomicU64>,
558 pub(crate) per_core_group_engine_exec_ns: Vec<PaddedAtomicU64>,
559 pub(crate) per_group_group_engine_exec_ns: Vec<PaddedAtomicU64>,
560 pub(crate) per_group_group_mailbox_depth: Vec<PaddedAtomicU64>,
561 pub(crate) per_group_group_mailbox_max_depth: Vec<PaddedAtomicU64>,
562 pub(crate) per_group_group_mailbox_full_events: Vec<PaddedAtomicU64>,
563 pub(crate) per_core_raft_write_many_batches: Vec<PaddedAtomicU64>,
564 pub(crate) per_group_raft_write_many_batches: Vec<PaddedAtomicU64>,
565 pub(crate) per_core_raft_write_many_commands: Vec<PaddedAtomicU64>,
566 pub(crate) per_group_raft_write_many_commands: Vec<PaddedAtomicU64>,
567 pub(crate) per_core_raft_write_many_logical_commands: Vec<PaddedAtomicU64>,
568 pub(crate) per_group_raft_write_many_logical_commands: Vec<PaddedAtomicU64>,
569 pub(crate) per_core_raft_write_many_responses: Vec<PaddedAtomicU64>,
570 pub(crate) per_group_raft_write_many_responses: Vec<PaddedAtomicU64>,
571 pub(crate) per_core_raft_write_many_submit_ns: Vec<PaddedAtomicU64>,
572 pub(crate) per_group_raft_write_many_submit_ns: Vec<PaddedAtomicU64>,
573 pub(crate) per_core_raft_write_many_response_ns: Vec<PaddedAtomicU64>,
574 pub(crate) per_group_raft_write_many_response_ns: Vec<PaddedAtomicU64>,
575 pub(crate) per_core_raft_apply_entries: Vec<PaddedAtomicU64>,
576 pub(crate) per_group_raft_apply_entries: Vec<PaddedAtomicU64>,
577 pub(crate) per_core_raft_apply_ns: Vec<PaddedAtomicU64>,
578 pub(crate) per_group_raft_apply_ns: Vec<PaddedAtomicU64>,
579 pub(crate) per_core_live_read_waiters: Vec<PaddedAtomicU64>,
580 pub(crate) per_core_live_read_backpressure_events: Vec<PaddedAtomicU64>,
581 pub(crate) per_core_routed_requests: Vec<PaddedAtomicU64>,
582 pub(crate) per_core_mailbox_send_wait_ns: Vec<PaddedAtomicU64>,
583 pub(crate) per_core_mailbox_full_events: Vec<PaddedAtomicU64>,
584 pub(crate) per_core_wal_batches: Vec<PaddedAtomicU64>,
585 pub(crate) per_group_wal_batches: Vec<PaddedAtomicU64>,
586 pub(crate) per_core_wal_records: Vec<PaddedAtomicU64>,
587 pub(crate) per_group_wal_records: Vec<PaddedAtomicU64>,
588 pub(crate) per_core_wal_write_ns: Vec<PaddedAtomicU64>,
589 pub(crate) per_group_wal_write_ns: Vec<PaddedAtomicU64>,
590 pub(crate) per_core_wal_sync_ns: Vec<PaddedAtomicU64>,
591 pub(crate) per_group_wal_sync_ns: Vec<PaddedAtomicU64>,
592 pub(crate) cold_flush_uploads: PaddedAtomicU64,
593 pub(crate) cold_flush_upload_bytes: PaddedAtomicU64,
594 pub(crate) cold_flush_upload_ns: PaddedAtomicU64,
595 pub(crate) cold_flush_publishes: PaddedAtomicU64,
596 pub(crate) cold_flush_publish_bytes: PaddedAtomicU64,
597 pub(crate) cold_flush_publish_ns: PaddedAtomicU64,
598 pub(crate) cold_orphan_cleanup_attempts: PaddedAtomicU64,
599 pub(crate) cold_orphan_cleanup_errors: PaddedAtomicU64,
600 pub(crate) cold_gc_reclaimed: PaddedAtomicU64,
601 pub(crate) cold_gc_errors: PaddedAtomicU64,
602 pub(crate) cold_flush_write_errors: PaddedAtomicU64,
603 pub(crate) cold_orphan_bytes: PaddedAtomicU64,
604 pub(crate) per_group_cold_hot_bytes: Vec<PaddedAtomicU64>,
605 pub(crate) per_group_cold_hot_bytes_max: Vec<PaddedAtomicU64>,
606 pub(crate) cold_hot_stream_bytes_max: PaddedAtomicU64,
607 pub(crate) per_core_cold_backpressure_events: Vec<PaddedAtomicU64>,
608 pub(crate) per_group_cold_backpressure_events: Vec<PaddedAtomicU64>,
609 pub(crate) cold_backpressure_bytes: PaddedAtomicU64,
610}
611
612#[derive(Debug, Clone, Copy)]
613pub(crate) struct RaftWriteManySample {
614 pub(crate) command_count: u64,
615 pub(crate) logical_command_count: u64,
616 pub(crate) response_count: u64,
617 pub(crate) submit_ns: u64,
618 pub(crate) response_ns: u64,
619}
620
621impl RuntimeMetricsInner {
622 pub(crate) fn new(core_count: usize, raft_group_count: usize) -> Self {
623 Self {
624 per_core_appends: (0..core_count).map(|_| PaddedAtomicU64::new(0)).collect(),
625 per_group_appends: (0..raft_group_count)
626 .map(|_| PaddedAtomicU64::new(0))
627 .collect(),
628 per_core_applied_mutations: (0..core_count).map(|_| PaddedAtomicU64::new(0)).collect(),
629 per_group_applied_mutations: (0..raft_group_count)
630 .map(|_| PaddedAtomicU64::new(0))
631 .collect(),
632 per_core_mutation_apply_ns: (0..core_count).map(|_| PaddedAtomicU64::new(0)).collect(),
633 per_group_mutation_apply_ns: (0..raft_group_count)
634 .map(|_| PaddedAtomicU64::new(0))
635 .collect(),
636 per_core_group_lock_wait_ns: (0..core_count).map(|_| PaddedAtomicU64::new(0)).collect(),
637 per_group_group_lock_wait_ns: (0..raft_group_count)
638 .map(|_| PaddedAtomicU64::new(0))
639 .collect(),
640 per_core_group_engine_exec_ns: (0..core_count)
641 .map(|_| PaddedAtomicU64::new(0))
642 .collect(),
643 per_group_group_engine_exec_ns: (0..raft_group_count)
644 .map(|_| PaddedAtomicU64::new(0))
645 .collect(),
646 per_group_group_mailbox_depth: (0..raft_group_count)
647 .map(|_| PaddedAtomicU64::new(0))
648 .collect(),
649 per_group_group_mailbox_max_depth: (0..raft_group_count)
650 .map(|_| PaddedAtomicU64::new(0))
651 .collect(),
652 per_group_group_mailbox_full_events: (0..raft_group_count)
653 .map(|_| PaddedAtomicU64::new(0))
654 .collect(),
655 per_core_raft_write_many_batches: (0..core_count)
656 .map(|_| PaddedAtomicU64::new(0))
657 .collect(),
658 per_group_raft_write_many_batches: (0..raft_group_count)
659 .map(|_| PaddedAtomicU64::new(0))
660 .collect(),
661 per_core_raft_write_many_commands: (0..core_count)
662 .map(|_| PaddedAtomicU64::new(0))
663 .collect(),
664 per_group_raft_write_many_commands: (0..raft_group_count)
665 .map(|_| PaddedAtomicU64::new(0))
666 .collect(),
667 per_core_raft_write_many_logical_commands: (0..core_count)
668 .map(|_| PaddedAtomicU64::new(0))
669 .collect(),
670 per_group_raft_write_many_logical_commands: (0..raft_group_count)
671 .map(|_| PaddedAtomicU64::new(0))
672 .collect(),
673 per_core_raft_write_many_responses: (0..core_count)
674 .map(|_| PaddedAtomicU64::new(0))
675 .collect(),
676 per_group_raft_write_many_responses: (0..raft_group_count)
677 .map(|_| PaddedAtomicU64::new(0))
678 .collect(),
679 per_core_raft_write_many_submit_ns: (0..core_count)
680 .map(|_| PaddedAtomicU64::new(0))
681 .collect(),
682 per_group_raft_write_many_submit_ns: (0..raft_group_count)
683 .map(|_| PaddedAtomicU64::new(0))
684 .collect(),
685 per_core_raft_write_many_response_ns: (0..core_count)
686 .map(|_| PaddedAtomicU64::new(0))
687 .collect(),
688 per_group_raft_write_many_response_ns: (0..raft_group_count)
689 .map(|_| PaddedAtomicU64::new(0))
690 .collect(),
691 per_core_raft_apply_entries: (0..core_count).map(|_| PaddedAtomicU64::new(0)).collect(),
692 per_group_raft_apply_entries: (0..raft_group_count)
693 .map(|_| PaddedAtomicU64::new(0))
694 .collect(),
695 per_core_raft_apply_ns: (0..core_count).map(|_| PaddedAtomicU64::new(0)).collect(),
696 per_group_raft_apply_ns: (0..raft_group_count)
697 .map(|_| PaddedAtomicU64::new(0))
698 .collect(),
699 per_core_live_read_waiters: (0..core_count).map(|_| PaddedAtomicU64::new(0)).collect(),
700 per_core_live_read_backpressure_events: (0..core_count)
701 .map(|_| PaddedAtomicU64::new(0))
702 .collect(),
703 per_core_routed_requests: (0..core_count).map(|_| PaddedAtomicU64::new(0)).collect(),
704 per_core_mailbox_send_wait_ns: (0..core_count)
705 .map(|_| PaddedAtomicU64::new(0))
706 .collect(),
707 per_core_mailbox_full_events: (0..core_count)
708 .map(|_| PaddedAtomicU64::new(0))
709 .collect(),
710 per_core_wal_batches: (0..core_count).map(|_| PaddedAtomicU64::new(0)).collect(),
711 per_group_wal_batches: (0..raft_group_count)
712 .map(|_| PaddedAtomicU64::new(0))
713 .collect(),
714 per_core_wal_records: (0..core_count).map(|_| PaddedAtomicU64::new(0)).collect(),
715 per_group_wal_records: (0..raft_group_count)
716 .map(|_| PaddedAtomicU64::new(0))
717 .collect(),
718 per_core_wal_write_ns: (0..core_count).map(|_| PaddedAtomicU64::new(0)).collect(),
719 per_group_wal_write_ns: (0..raft_group_count)
720 .map(|_| PaddedAtomicU64::new(0))
721 .collect(),
722 per_core_wal_sync_ns: (0..core_count).map(|_| PaddedAtomicU64::new(0)).collect(),
723 per_group_wal_sync_ns: (0..raft_group_count)
724 .map(|_| PaddedAtomicU64::new(0))
725 .collect(),
726 cold_flush_uploads: PaddedAtomicU64::new(0),
727 cold_flush_upload_bytes: PaddedAtomicU64::new(0),
728 cold_flush_upload_ns: PaddedAtomicU64::new(0),
729 cold_flush_publishes: PaddedAtomicU64::new(0),
730 cold_flush_publish_bytes: PaddedAtomicU64::new(0),
731 cold_flush_publish_ns: PaddedAtomicU64::new(0),
732 cold_orphan_cleanup_attempts: PaddedAtomicU64::new(0),
733 cold_orphan_cleanup_errors: PaddedAtomicU64::new(0),
734 cold_gc_reclaimed: PaddedAtomicU64::new(0),
735 cold_gc_errors: PaddedAtomicU64::new(0),
736 cold_flush_write_errors: PaddedAtomicU64::new(0),
737 cold_orphan_bytes: PaddedAtomicU64::new(0),
738 per_group_cold_hot_bytes: (0..raft_group_count)
739 .map(|_| PaddedAtomicU64::new(0))
740 .collect(),
741 per_group_cold_hot_bytes_max: (0..raft_group_count)
742 .map(|_| PaddedAtomicU64::new(0))
743 .collect(),
744 cold_hot_stream_bytes_max: PaddedAtomicU64::new(0),
745 per_core_cold_backpressure_events: (0..core_count)
746 .map(|_| PaddedAtomicU64::new(0))
747 .collect(),
748 per_group_cold_backpressure_events: (0..raft_group_count)
749 .map(|_| PaddedAtomicU64::new(0))
750 .collect(),
751 cold_backpressure_bytes: PaddedAtomicU64::new(0),
752 }
753 }
754
755 pub(crate) fn record_routed_request(&self, core_id: CoreId, mailbox_send_wait_ns: u64) {
756 let index = usize::from(core_id.0);
757 self.per_core_routed_requests[index].fetch_add_relaxed(1);
758 self.per_core_mailbox_send_wait_ns[index].fetch_add_relaxed(mailbox_send_wait_ns);
759 }
760
761 pub(crate) fn record_mailbox_full(&self, core_id: CoreId) {
762 self.per_core_mailbox_full_events[usize::from(core_id.0)].fetch_add_relaxed(1);
763 }
764
765 pub(crate) fn record_append(&self, core_id: CoreId, group_id: RaftGroupId) {
766 self.record_append_batch(core_id, group_id, 1);
767 }
768
769 pub(crate) fn record_append_batch(&self, core_id: CoreId, group_id: RaftGroupId, count: u64) {
770 self.per_core_appends[usize::from(core_id.0)].fetch_add_relaxed(count);
771 self.per_group_appends[usize::try_from(group_id.0).expect("u32 fits usize")]
772 .fetch_add_relaxed(count);
773 }
774
775 pub(crate) fn record_applied_mutation(
776 &self,
777 core_id: CoreId,
778 group_id: RaftGroupId,
779 apply_ns: u64,
780 ) {
781 self.record_applied_mutation_batch(core_id, group_id, 1, apply_ns);
782 }
783
784 pub(crate) fn record_applied_mutation_batch(
785 &self,
786 core_id: CoreId,
787 group_id: RaftGroupId,
788 count: u64,
789 apply_ns: u64,
790 ) {
791 let core_index = usize::from(core_id.0);
792 let group_index = usize::try_from(group_id.0).expect("u32 fits usize");
793 self.per_core_applied_mutations[core_index].fetch_add_relaxed(count);
794 self.per_group_applied_mutations[group_index].fetch_add_relaxed(count);
795 self.per_core_mutation_apply_ns[core_index].fetch_add_relaxed(apply_ns);
796 self.per_group_mutation_apply_ns[group_index].fetch_add_relaxed(apply_ns);
797 }
798
799 pub(crate) fn record_group_engine_exec(
800 &self,
801 core_id: CoreId,
802 group_id: RaftGroupId,
803 exec_ns: u64,
804 ) {
805 let core_index = usize::from(core_id.0);
806 let group_index = usize::try_from(group_id.0).expect("u32 fits usize");
807 self.per_core_group_engine_exec_ns[core_index].fetch_add_relaxed(exec_ns);
808 self.per_group_group_engine_exec_ns[group_index].fetch_add_relaxed(exec_ns);
809 }
810
811 pub(crate) fn record_group_mailbox_enqueued(&self, group_id: RaftGroupId) {
812 let group_index = usize::try_from(group_id.0).expect("u32 fits usize");
813 let depth = self.per_group_group_mailbox_depth[group_index]
814 .fetch_add_relaxed(1)
815 .saturating_add(1);
816 self.per_group_group_mailbox_max_depth[group_index].fetch_max_relaxed(depth);
817 }
818
819 pub(crate) fn record_group_mailbox_dequeued(&self, group_id: RaftGroupId) {
820 let group_index = usize::try_from(group_id.0).expect("u32 fits usize");
821 self.per_group_group_mailbox_depth[group_index].fetch_sub_saturating_relaxed(1);
822 }
823
824 pub(crate) fn record_group_mailbox_full(&self, group_id: RaftGroupId) {
825 let group_index = usize::try_from(group_id.0).expect("u32 fits usize");
826 self.per_group_group_mailbox_full_events[group_index].fetch_add_relaxed(1);
827 }
828
829 pub(crate) fn record_raft_write_many(
830 &self,
831 core_id: CoreId,
832 group_id: RaftGroupId,
833 sample: RaftWriteManySample,
834 ) {
835 let core_index = usize::from(core_id.0);
836 let group_index = usize::try_from(group_id.0).expect("u32 fits usize");
837 self.per_core_raft_write_many_batches[core_index].fetch_add_relaxed(1);
838 self.per_group_raft_write_many_batches[group_index].fetch_add_relaxed(1);
839 self.per_core_raft_write_many_commands[core_index].fetch_add_relaxed(sample.command_count);
840 self.per_group_raft_write_many_commands[group_index]
841 .fetch_add_relaxed(sample.command_count);
842 self.per_core_raft_write_many_logical_commands[core_index]
843 .fetch_add_relaxed(sample.logical_command_count);
844 self.per_group_raft_write_many_logical_commands[group_index]
845 .fetch_add_relaxed(sample.logical_command_count);
846 self.per_core_raft_write_many_responses[core_index]
847 .fetch_add_relaxed(sample.response_count);
848 self.per_group_raft_write_many_responses[group_index]
849 .fetch_add_relaxed(sample.response_count);
850 self.per_core_raft_write_many_submit_ns[core_index].fetch_add_relaxed(sample.submit_ns);
851 self.per_group_raft_write_many_submit_ns[group_index].fetch_add_relaxed(sample.submit_ns);
852 self.per_core_raft_write_many_response_ns[core_index].fetch_add_relaxed(sample.response_ns);
853 self.per_group_raft_write_many_response_ns[group_index]
854 .fetch_add_relaxed(sample.response_ns);
855 }
856
857 pub(crate) fn record_raft_apply_batch(
858 &self,
859 core_id: CoreId,
860 group_id: RaftGroupId,
861 entry_count: u64,
862 apply_ns: u64,
863 ) {
864 let core_index = usize::from(core_id.0);
865 let group_index = usize::try_from(group_id.0).expect("u32 fits usize");
866 self.per_core_raft_apply_entries[core_index].fetch_add_relaxed(entry_count);
867 self.per_group_raft_apply_entries[group_index].fetch_add_relaxed(entry_count);
868 self.per_core_raft_apply_ns[core_index].fetch_add_relaxed(apply_ns);
869 self.per_group_raft_apply_ns[group_index].fetch_add_relaxed(apply_ns);
870 }
871
872 pub(crate) fn record_wal_batch(
873 &self,
874 core_id: CoreId,
875 group_id: RaftGroupId,
876 record_count: u64,
877 write_ns: u64,
878 sync_ns: u64,
879 ) {
880 let core_index = usize::from(core_id.0);
881 let group_index = usize::try_from(group_id.0).expect("u32 fits usize");
882 self.per_core_wal_batches[core_index].fetch_add_relaxed(1);
883 self.per_group_wal_batches[group_index].fetch_add_relaxed(1);
884 self.per_core_wal_records[core_index].fetch_add_relaxed(record_count);
885 self.per_group_wal_records[group_index].fetch_add_relaxed(record_count);
886 self.per_core_wal_write_ns[core_index].fetch_add_relaxed(write_ns);
887 self.per_group_wal_write_ns[group_index].fetch_add_relaxed(write_ns);
888 self.per_core_wal_sync_ns[core_index].fetch_add_relaxed(sync_ns);
889 self.per_group_wal_sync_ns[group_index].fetch_add_relaxed(sync_ns);
890 }
891
892 pub(crate) fn record_cold_upload(&self, bytes: u64, upload_ns: u64) {
893 self.cold_flush_uploads.fetch_add_relaxed(1);
894 self.cold_flush_upload_bytes.fetch_add_relaxed(bytes);
895 self.cold_flush_upload_ns.fetch_add_relaxed(upload_ns);
896 }
897
898 pub(crate) fn record_cold_publish(&self, bytes: u64, publish_ns: u64) {
899 self.cold_flush_publishes.fetch_add_relaxed(1);
900 self.cold_flush_publish_bytes.fetch_add_relaxed(bytes);
901 self.cold_flush_publish_ns.fetch_add_relaxed(publish_ns);
902 }
903
904 pub(crate) fn record_cold_gc_reclaimed(&self, entries: u64) {
905 self.cold_gc_reclaimed.fetch_add_relaxed(entries);
906 }
907
908 pub(crate) fn record_cold_flush_write_error(&self) {
909 self.cold_flush_write_errors.fetch_add_relaxed(1);
910 }
911
912 pub(crate) fn record_cold_gc_error(&self) {
913 self.cold_gc_errors.fetch_add_relaxed(1);
914 }
915
916 pub(crate) fn record_cold_hot_backlog(
917 &self,
918 group_id: RaftGroupId,
919 stream_hot_bytes: u64,
920 group_hot_bytes: u64,
921 ) {
922 let group_index = usize::try_from(group_id.0).expect("u32 fits usize");
923 self.per_group_cold_hot_bytes[group_index].store_relaxed(group_hot_bytes);
924 self.per_group_cold_hot_bytes_max[group_index].fetch_max_relaxed(group_hot_bytes);
925 self.cold_hot_stream_bytes_max
926 .fetch_max_relaxed(stream_hot_bytes);
927 }
928
929 pub(crate) fn record_cold_backpressure(
930 &self,
931 core_id: CoreId,
932 group_id: RaftGroupId,
933 incoming_bytes: u64,
934 _limit: u64,
935 ) {
936 let core_index = usize::from(core_id.0);
937 let group_index = usize::try_from(group_id.0).expect("u32 fits usize");
938 self.per_core_cold_backpressure_events[core_index].fetch_add_relaxed(1);
939 self.per_group_cold_backpressure_events[group_index].fetch_add_relaxed(1);
940 self.cold_backpressure_bytes
941 .fetch_add_relaxed(incoming_bytes);
942 }
943
944 pub(crate) fn record_read_watcher_added(&self, core_id: CoreId) {
945 self.record_read_watchers_added(core_id, 1);
946 }
947
948 pub(crate) fn record_read_watchers_added(&self, core_id: CoreId, count: usize) {
949 self.per_core_live_read_waiters[usize::from(core_id.0)]
950 .fetch_add_relaxed(u64::try_from(count).expect("watcher count fits u64"));
951 }
952
953 pub(crate) fn record_read_watchers_removed(&self, core_id: CoreId, count: usize) {
954 self.per_core_live_read_waiters[usize::from(core_id.0)]
955 .fetch_sub_relaxed(u64::try_from(count).expect("watcher count fits u64"));
956 }
957
958 pub(crate) fn record_live_read_backpressure(&self, core_id: CoreId) {
959 self.per_core_live_read_backpressure_events[usize::from(core_id.0)].fetch_add_relaxed(1);
960 }
961}
962
963pub(crate) fn elapsed_ns(started_at: Instant) -> u64 {
964 u64::try_from(started_at.elapsed().as_nanos()).unwrap_or(u64::MAX)
965}
966
967pub(crate) fn append_batch_payload_bytes(request: &AppendBatchRequest) -> u64 {
968 request
969 .payloads
970 .iter()
971 .map(|payload| u64::try_from(payload.len()).expect("payload len fits u64"))
972 .sum()
973}
974
975pub(crate) fn record_cold_backpressure_error(
976 metrics: &RuntimeMetricsInner,
977 placement: ShardPlacement,
978 incoming_bytes: u64,
979 admission: ColdWriteAdmission,
980 err: &GroupEngineError,
981) {
982 if !err.is_cold_backpressure() {
983 return;
984 }
985 metrics.record_cold_backpressure(
986 placement.core_id,
987 placement.raft_group_id,
988 incoming_bytes,
989 admission.max_hot_bytes_per_group.unwrap_or(0),
990 );
991}
992
993pub(crate) fn is_stale_cold_flush_candidate_error(err: &RuntimeError) -> bool {
994 match err.stream_error_code() {
995 Some(StreamErrorCode::StreamGone | StreamErrorCode::StreamNotFound) => true,
996 Some(StreamErrorCode::InvalidColdFlush) => err
997 .stream_error_context()
998 .iter()
999 .any(|context| matches!(context, StreamErrorContext::StaleColdFlushCandidate)),
1000 _ => false,
1001 }
1002}
1003
1004pub(crate) async fn record_cold_hot_backlog(
1005 group: &mut Box<dyn GroupEngine>,
1006 metrics: &RuntimeMetricsInner,
1007 stream_id: BucketStreamId,
1008 placement: ShardPlacement,
1009) {
1010 if let Ok(backlog) = group.cold_hot_backlog(stream_id, placement).await {
1011 metrics.record_cold_hot_backlog(
1012 placement.raft_group_id,
1013 backlog.stream_hot_bytes,
1014 backlog.group_hot_bytes,
1015 );
1016 }
1017}
1018
1019#[derive(Debug)]
1020#[repr(align(128))]
1021pub(crate) struct PaddedAtomicU64 {
1022 value: AtomicU64,
1023}
1024
1025impl PaddedAtomicU64 {
1026 pub(crate) fn new(value: u64) -> Self {
1027 Self {
1028 value: AtomicU64::new(value),
1029 }
1030 }
1031
1032 pub(crate) fn load_relaxed(&self) -> u64 {
1033 self.value.load(Ordering::Relaxed)
1034 }
1035
1036 pub(crate) fn fetch_add_relaxed(&self, value: u64) -> u64 {
1037 self.value.fetch_add(value, Ordering::Relaxed)
1038 }
1039
1040 pub(crate) fn fetch_sub_relaxed(&self, value: u64) {
1041 self.value.fetch_sub(value, Ordering::Relaxed);
1042 }
1043
1044 pub(crate) fn fetch_sub_saturating_relaxed(&self, value: u64) {
1045 let mut current = self.value.load(Ordering::Relaxed);
1046 loop {
1047 let next = current.saturating_sub(value);
1048 match self.value.compare_exchange_weak(
1049 current,
1050 next,
1051 Ordering::Relaxed,
1052 Ordering::Relaxed,
1053 ) {
1054 Ok(_) => return,
1055 Err(observed) => current = observed,
1056 }
1057 }
1058 }
1059
1060 pub(crate) fn fetch_max_relaxed(&self, value: u64) {
1061 self.value.fetch_max(value, Ordering::Relaxed);
1062 }
1063
1064 pub(crate) fn store_relaxed(&self, value: u64) {
1065 self.value.store(value, Ordering::Relaxed);
1066 }
1067}