1#include <enforce.h>
2#include <mem/alloc.h>
3#include <stat_series.h>
4#include <sync/lock_chk_assert.h>
5#include <time/time.h>
6#include <tsa.h>
7
8void stat_series_init(struct stat_series *s, struct stat_bucket *buckets,
9 uint32_t nbuckets, time_us_t bucket_us,
10 stat_series_callback bucket_reset, void *private) {
11 memset(buckets, 0, nbuckets * sizeof(struct stat_bucket));
12 s->buckets = buckets;
13 s->bucket_reset = bucket_reset;
14 s->nbuckets = nbuckets;
15 s->bucket_us = bucket_us;
16 atomic_init(&s->current, 0);
17 atomic_init(&s->last_update_us, time_get_us());
18 s->private = private;
19 spinlock_init(&s->lock);
20
21 for (size_t i = 0; i < nbuckets; i++)
22 s->buckets[i].parent = s;
23}
24
25struct stat_series *stat_series_create(uint32_t nbuckets, time_us_t bucket_us,
26 stat_series_callback bucket_reset,
27 void *private) {
28 struct stat_bucket *buckets = vnews(buckets, nbuckets, ALLOC_ZERO);
29 if (!buckets)
30 return NULL;
31
32 struct stat_series *series = vnew(series, ALLOC_ZERO);
33 if (!series) {
34 vfree(buckets);
35 return NULL;
36 }
37
38 stat_series_init(s: series, buckets, nbuckets, bucket_us, bucket_reset,
39 private);
40 return series;
41}
42
43void stat_series_reset(struct stat_series *s) {
44 enum irql irql = spin_lock(&s->lock);
45
46 struct stat_bucket *bucket;
47
48 stat_series_for_each(bucket, s) {
49 atomic_store(&bucket->count, 0);
50 atomic_store(&bucket->sum, 0);
51 s->bucket_reset(bucket);
52 }
53
54 spin_unlock(&s->lock, irql);
55}
56
57static void stat_series_advance_locked(struct stat_series *s, time_us_t now_us)
58 TSA_CALLER_HOLDS(&s->lock) {
59 /* re-evaluate based on locked state (canonical update) */
60 size_t delta = now_us - atomic_load_rlx(&s->last_update_us);
61 uint32_t steps = delta / s->bucket_us;
62 if (steps == 0)
63 return;
64
65 if (steps > s->nbuckets)
66 steps = s->nbuckets;
67
68 size_t series_current = atomic_load_rlx(&s->current);
69
70 for (uint32_t i = 0; i < steps; i++) {
71 /* update canonical non-atomic s->current while holding lock */
72 size_t current = (series_current + 1) % s->nbuckets;
73 struct stat_bucket *bucket = &s->buckets[current];
74
75 atomic_store(&bucket->count, 0);
76 atomic_store(&bucket->sum, 0);
77 s->bucket_reset(bucket);
78 atomic_store_release(&s->current, current);
79 }
80
81 /* publish last_update_us atomically for lockless readers/writers */
82 atomic_store(&s->last_update_us,
83 atomic_load_rlx(&s->last_update_us) + steps * s->bucket_us);
84}
85
86void stat_series_advance(struct stat_series *s, time_us_t now_us) {
87 /* Fast-path: check without taking lock */
88 time_us_t last = atomic_load(&s->last_update_us);
89 size_t delta = now_us - last;
90 uint32_t steps = delta / s->bucket_us;
91 if (steps == 0)
92 return; /* no rotation required, avoid locking */
93
94 /* Take lock and re-check/recompute using the protected fields */
95 enum irql irql = spin_lock(&s->lock);
96 stat_series_advance_locked(s, now_us);
97 spin_unlock(&s->lock, irql);
98}
99
100void stat_series_record(struct stat_series *s, size_t value,
101 stat_series_callback callback) {
102 time_us_t now_us = time_get_us();
103
104 /* attempt to advance if needed */
105 stat_series_advance(s, now_us);
106
107 /* read canonical current published by advance (atomic load) */
108 uint32_t cur = atomic_load(&s->current); /* acquire ordering */
109 struct stat_bucket *b = &s->buckets[cur];
110
111 /* hot path: only atomic RMWs on bucket */
112 atomic_inc(&b->count);
113 atomic_fetch_add(&b->sum, value);
114 if (callback)
115 callback(b);
116}
117