@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
c 143 lines 5.57 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "stream.h"
4 #include "stream-sender-internals.h"
5
6 struct stream_circular_buffer {
7 struct circular_buffer *cb;
8 STREAM_CIRCULAR_BUFFER_STATS stats;
9
10 usec_t last_recreate_ut; // recreates are only used to shrink the buffer, they are normal during operation
11 usec_t last_sent_ut; // the last time we removed or flushed data from the buffer
12
13 struct {
14 // the current max size of the buffer
15 size_t max_size;
16
17 // the current utilization of the buffer
18 size_t buffer_ratio;
19
20 // the last time we flushed the buffer
21 // by monitoring this we can know if the system was reconnected
22 usec_t last_flush_ut;
23 } atomic;
24 };
25
26 static inline void stream_circular_buffer_stats_update_unsafe(STREAM_CIRCULAR_BUFFER *scb) {
27 scb->stats.bytes_size = scb->cb->size;
28 scb->stats.bytes_max_size = scb->cb->max_size;
29 scb->stats.bytes_outstanding = cbuffer_next_unsafe(scb->cb, NULL);
30 scb->stats.bytes_available = cbuffer_available_size_unsafe(scb->cb);
31 scb->stats.buffer_ratio = (double)(scb->cb->max_size - scb->stats.bytes_available) * 100.0 / (double)scb->cb->max_size;
32
33 __atomic_store_n(&((scb)->atomic.buffer_ratio), (size_t)round(scb->stats.buffer_ratio), __ATOMIC_RELAXED);
34 }
35
36 STREAM_CIRCULAR_BUFFER *stream_circular_buffer_create(void) {
37 STREAM_CIRCULAR_BUFFER *scb = callocz(1, sizeof(*scb));
38 scb->cb = cbuffer_new(CBUFFER_INITIAL_SIZE, CBUFFER_INITIAL_MAX_SIZE, &netdata_buffers_statistics.cbuffers_streaming);
39 stream_circular_buffer_stats_update_unsafe(scb);
40 return scb;
41 }
42
43 // returns true if it increased the buffer size
44 bool stream_circular_buffer_set_max_size_unsafe(STREAM_CIRCULAR_BUFFER *scb, size_t max_size, bool force) {
45 if(force || scb->cb->max_size < max_size) {
46 scb->cb->max_size = max_size;
47 scb->stats.bytes_max_size = scb->cb->max_size;
48 __atomic_store_n(&scb->atomic.max_size, scb->cb->max_size, __ATOMIC_RELAXED);
49 stream_circular_buffer_stats_update_unsafe(scb);
50 return true;
51 }
52
53 return false;
54 }
55
56 void stream_circular_buffer_flush_unsafe(STREAM_CIRCULAR_BUFFER *scb, size_t buffer_max_size) {
57 usec_t now_ut = now_monotonic_usec();
58 __atomic_store_n(&scb->atomic.last_flush_ut, now_ut, __ATOMIC_RELAXED);
59
60 // flush the output buffer from any data it may have
61 scb->last_sent_ut = now_ut;
62 cbuffer_flush(scb->cb);
63 memset(&scb->stats, 0, sizeof(scb->stats));
64 stream_circular_buffer_set_max_size_unsafe(scb, buffer_max_size, true);
65 stream_circular_buffer_recreate_timed_unsafe(scb, now_monotonic_usec(), true);
66 }
67
68 inline size_t stream_sender_get_buffer_used_percent(STREAM_CIRCULAR_BUFFER *scb) {
69 return __atomic_load_n(&scb->atomic.buffer_ratio, __ATOMIC_RELAXED);
70 }
71
72 size_t stream_circular_buffer_get_max_size(STREAM_CIRCULAR_BUFFER *scb) {
73 return __atomic_load_n(&scb->atomic.max_size, __ATOMIC_RELAXED);
74 }
75
76 void stream_circular_buffer_recreate_timed_unsafe(STREAM_CIRCULAR_BUFFER *scb, usec_t now_ut, bool force) {
77 if(!force && (scb->stats.bytes_outstanding || now_ut - scb->last_recreate_ut < 300 * USEC_PER_SEC))
78 return;
79
80 scb->last_recreate_ut = now_ut;
81
82 scb->stats.recreates++; // we increase even if we don't do it, to have sender_start() recreate its buffers
83
84 if(scb->cb && scb->cb->size > CBUFFER_INITIAL_SIZE) {
85 size_t max_size = scb->cb->max_size;
86 cbuffer_free(scb->cb);
87 scb->cb = cbuffer_new(CBUFFER_INITIAL_SIZE, max_size, &netdata_buffers_statistics.cbuffers_streaming);
88 }
89 }
90
91 inline usec_t stream_circular_buffer_last_flush_ut(STREAM_CIRCULAR_BUFFER *scb) {
92 return __atomic_load_n(&((scb)->atomic.last_flush_ut), __ATOMIC_RELAXED);
93 }
94
95 inline usec_t stream_circular_buffer_last_sent_ut(STREAM_CIRCULAR_BUFFER *scb) {
96 // this is ok without locks and atomics, since only the stream threads
97 // can actually remove data and call this
98 return scb->last_sent_ut;
99 }
100
101 void stream_circular_buffer_destroy(STREAM_CIRCULAR_BUFFER *scb) {
102 if(!scb) return;
103 cbuffer_free(scb->cb);
104 freez(scb);
105 }
106
107 // adds data to the circular buffer, returns false when it can't (buffer is full)
108 bool stream_circular_buffer_add_unsafe(
109 STREAM_CIRCULAR_BUFFER *scb, const char *data,
110 size_t bytes_actual, size_t bytes_uncompressed, STREAM_TRAFFIC_TYPE type, bool autoscale) {
111 scb->stats.adds++;
112 scb->stats.bytes_added += bytes_actual;
113 scb->stats.bytes_uncompressed += bytes_uncompressed;
114 scb->stats.bytes_sent_by_type[type] += bytes_actual;
115
116 if(unlikely(autoscale && cbuffer_available_size_unsafe(scb->cb) < bytes_actual))
117 stream_circular_buffer_set_max_size_unsafe(scb, scb->cb->max_size * 2, true);
118
119 if(unlikely(cbuffer_add_unsafe(scb->cb, data, bytes_actual) != 0))
120 return false;
121
122 stream_circular_buffer_stats_update_unsafe(scb);
123 return true;
124 }
125
126 // return the first available chunk at the beginning of the buffer
127 size_t stream_circular_buffer_get_unsafe(STREAM_CIRCULAR_BUFFER *scb, char **chunk) {
128 return cbuffer_next_unsafe(scb->cb, chunk);
129 }
130
131 // removes data from the beginning of the circular buffer
132 void stream_circular_buffer_del_unsafe(STREAM_CIRCULAR_BUFFER *scb, size_t bytes, usec_t now_ut) {
133 scb->last_sent_ut = now_ut ? now_ut : now_monotonic_usec();
134 scb->stats.sends++;
135 scb->stats.bytes_sent += bytes;
136 cbuffer_remove_unsafe(scb->cb, bytes);
137 stream_circular_buffer_stats_update_unsafe(scb);
138 }
139
140 // returns a copy of the current circular buffer statistics
141 STREAM_CIRCULAR_BUFFER_STATS *stream_circular_buffer_stats_unsafe(STREAM_CIRCULAR_BUFFER *scb) {
142 return &scb->stats;
143 }