@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
c 233 lines 7.02 KB
Raw
1 #include "../libnetdata.h"
2
3 // Initialize a pre-allocated circular buffer
4 void cbuffer_init(struct circular_buffer *buf, size_t initial, size_t max, size_t *statistics) {
5 if (unlikely(!buf))
6 return;
7
8 buf->size = initial;
9 buf->data = mallocz(initial);
10 buf->write = 0;
11 buf->read = 0;
12 buf->max_size = max;
13 buf->statistics = statistics;
14
15 if(buf->statistics)
16 __atomic_add_fetch(buf->statistics, buf->size, __ATOMIC_RELAXED);
17 }
18
19 // Cleanup resources for a pre-allocated circular buffer
20 void cbuffer_cleanup(struct circular_buffer *buf) {
21 if (unlikely(!buf))
22 return;
23
24 if(buf->statistics)
25 __atomic_sub_fetch(buf->statistics, buf->size, __ATOMIC_RELAXED);
26
27 freez(buf->data);
28 buf->data = NULL;
29 buf->size = 0;
30 buf->write = 0;
31 buf->read = 0;
32 }
33
34 // Allocate and initialize a new circular buffer
35 struct circular_buffer *cbuffer_new(size_t initial, size_t max, size_t *statistics) {
36 struct circular_buffer *buf = mallocz(sizeof(struct circular_buffer));
37 cbuffer_init(buf, initial, max, statistics);
38
39 if(buf->statistics)
40 __atomic_add_fetch(buf->statistics, sizeof(struct circular_buffer), __ATOMIC_RELAXED);
41
42 return buf;
43 }
44
45 // Free a circular buffer allocated with cbuffer_new
46 void cbuffer_free(struct circular_buffer *buf) {
47 if (unlikely(!buf))
48 return;
49
50 if(buf->statistics)
51 __atomic_sub_fetch(buf->statistics, sizeof(struct circular_buffer), __ATOMIC_RELAXED);
52
53 cbuffer_cleanup(buf);
54 freez(buf);
55 }
56
57 static int cbuffer_realloc_unsafe(struct circular_buffer *buf) {
58 // Check that we can grow
59 if (buf->size >= buf->max_size)
60 return 1;
61
62 size_t old_size = buf->size;
63 size_t new_size = buf->size * 2;
64 if (new_size > buf->max_size)
65 new_size = buf->max_size;
66
67 // We know that: size < new_size <= max_size
68 // For simplicity align the current data at the bottom of the new buffer
69 char *new_data = mallocz(new_size);
70 if (buf->read == buf->write)
71 buf->write = 0; // buffer is empty
72 else if (buf->read < buf->write) {
73 memcpy(new_data, buf->data + buf->read, buf->write - buf->read);
74 buf->write -= buf->read;
75 } else {
76 size_t top_part = buf->size - buf->read;
77 memcpy(new_data, buf->data + buf->read, top_part);
78 memcpy(new_data + top_part, buf->data, buf->write);
79 buf->write = top_part + buf->write;
80 }
81 buf->read = 0;
82
83 // Switch buffers
84 freez(buf->data);
85 buf->data = new_data;
86 buf->size = new_size;
87
88 if(buf->statistics)
89 __atomic_add_fetch(buf->statistics, new_size - old_size, __ATOMIC_RELAXED);
90
91 return 0;
92 }
93
94 ALWAYS_INLINE
95 size_t cbuffer_used_size_unsafe(struct circular_buffer *buf) {
96 return (buf->write >= buf->read) ? (buf->write - buf->read) : (buf->size - buf->read + buf->write);
97 }
98
99 ALWAYS_INLINE
100 size_t cbuffer_available_size_unsafe(struct circular_buffer *buf) {
101 return buf->max_size - cbuffer_used_size_unsafe(buf);
102 }
103
104 int cbuffer_add_unsafe(struct circular_buffer *buf, const char *d, size_t d_len) {
105 size_t len = cbuffer_used_size_unsafe(buf);
106 while (d_len + len >= buf->size) {
107 if (cbuffer_realloc_unsafe(buf)) {
108 return 1;
109 }
110 }
111 // Guarantee: write + d_len cannot hit read
112 if (buf->write + d_len < buf->size) {
113 memcpy(buf->data + buf->write, d, d_len);
114 buf->write += d_len;
115 }
116 else {
117 size_t top_part = buf->size - buf->write;
118 memcpy(buf->data + buf->write, d, top_part);
119 memcpy(buf->data, d + top_part, d_len - top_part);
120 buf->write = d_len - top_part;
121 }
122 return 0;
123 }
124
125 // Assume caller does not remove too many bytes (i.e. read will jump over write)
126 ALWAYS_INLINE
127 void cbuffer_remove_unsafe(struct circular_buffer *buf, size_t num) {
128 buf->read += num;
129 // Assume num < size (i.e. caller cannot remove more bytes than are in the buffer)
130 if (buf->read >= buf->size)
131 buf->read -= buf->size;
132 }
133
134 ALWAYS_INLINE
135 size_t cbuffer_next_unsafe(struct circular_buffer *buf, char **start) {
136 if (start != NULL)
137 *start = buf->data + buf->read;
138
139 if (buf->read <= buf->write) {
140 return buf->write - buf->read; // Includes empty case
141 }
142 return buf->size - buf->read;
143 }
144
145 ALWAYS_INLINE
146 void cbuffer_flush(struct circular_buffer*buf) {
147 buf->write = 0;
148 buf->read = 0;
149 }
150
151 // Ensures that the requested size is available as a contiguous block in the buffer
152 // Returns true if there's enough data and it's now contiguous, false otherwise
153 bool cbuffer_ensure_unwrapped_size(struct circular_buffer *buf, size_t size) {
154 if (unlikely(!buf || !buf->data))
155 return false;
156
157 size_t used = cbuffer_used_size_unsafe(buf);
158 if(used < size)
159 return false;
160
161 char *unwrapped;
162 size_t unwrapped_size = cbuffer_next_unsafe(buf, &unwrapped);
163 if(unwrapped_size >= size)
164 return true;
165
166 size_t wrapped_size = used - unwrapped_size;
167
168 char *tmp = mallocz(unwrapped_size);
169 memcpy(tmp, unwrapped, unwrapped_size);
170 cbuffer_remove_unsafe(buf, unwrapped_size);
171
172 memmove(buf->data + unwrapped_size, buf->data, wrapped_size);
173 memcpy(buf->data, tmp, unwrapped_size);
174 freez(tmp);
175
176 buf->read = 0;
177 buf->write = unwrapped_size + wrapped_size;
178
179 return true;
180 }
181
182 // Reserve space in the circular buffer for direct writing
183 // Returns a pointer to the reserved space, or NULL if reservation fails
184 char *cbuffer_reserve_unsafe(struct circular_buffer *buf, size_t size) {
185 if (unlikely(!buf || !buf->data || size == 0))
186 return NULL;
187
188 // First, make sure we have enough space in the buffer
189 size_t len = cbuffer_used_size_unsafe(buf);
190 while (size + len >= buf->size) {
191 if (cbuffer_realloc_unsafe(buf)) {
192 // Can't grow buffer anymore
193 return NULL;
194 }
195 }
196
197 if(buf->write + size > buf->size) {
198 if (!cbuffer_ensure_unwrapped_size(buf, len))
199 return NULL;
200
201 if(buf->read != 0 && buf->write + size > buf->size) {
202 // It is a contiguous buffer, but we need to move the data
203 // Move the data to the beginning of the buffer
204 memmove(buf->data, buf->data + buf->read, buf->write - buf->read);
205 buf->write -= buf->read;
206 buf->read = 0;
207 }
208 }
209
210 // Check if we can write contiguously from the current write position
211 if (buf->write + size <= buf->size) {
212 // Simple case - we have enough space at the current write position
213 return buf->data + buf->write;
214 }
215 else {
216 // impossible case since cbuffer_ensure_unwrapped_size() returned true
217 return NULL;
218 }
219 }
220
221 // Commit the reserved space after writing to it
222 // Size should be less than or equal to the size passed to cbuffer_reserve_unsafe
223 void cbuffer_commit_reserved_unsafe(struct circular_buffer *buf, size_t size) {
224 if (unlikely(!buf || !buf->data || size == 0))
225 return;
226
227 // Update the write pointer
228 buf->write += size;
229
230 // Handle wrap-around if we've gone past the buffer boundary
231 if (buf->write >= buf->size)
232 buf->write -= buf->size;
233 }