@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
h 248 lines 9.62 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #ifndef NETDATA_STREAM_THREAD_H
4 #define NETDATA_STREAM_THREAD_H
5
6 #include "libnetdata/libnetdata.h"
7 #include "stream-circular-buffer.h"
8 #include "stream-handshake.h"
9
10 struct stream_thread;
11 struct pollfd_slotted {
12 struct stream_thread *sth;
13 int32_t slot;
14 int fd;
15 };
16
17 #define PFD_EMPTY (struct pollfd_slotted){ .sth = NULL, .fd = -1, .slot = -1, }
18
19 typedef enum __attribute__((packed)) {
20 STREAM_OPCODE_NONE = 0,
21 STREAM_OPCODE_SENDER_POLLOUT = (1 << 0), // move traffic around as soon as possible
22 STREAM_OPCODE_RECEIVER_POLLOUT = (1 << 1), // disconnect the node, it has buffer overflow
23 STREAM_OPCODE_SENDER_BUFFER_OVERFLOW = (1 << 2), // reconnect the node, it has buffer overflow
24 STREAM_OPCODE_RECEIVER_BUFFER_OVERFLOW = (1 << 3), // reconnect the node, it has buffer overflow
25 STREAM_OPCODE_SENDER_RECONNECT_WITHOUT_COMPRESSION = (1 << 4), // reconnect the node, but disable compression
26 STREAM_OPCODE_SENDER_STOP_RECEIVER_LEFT = (1 << 5), // disconnect the node, the receiver left
27 STREAM_OPCODE_SENDER_STOP_HOST_CLEANUP = (1 << 6), // disconnect the node, it is being de-allocated
28 } STREAM_OPCODE;
29
30 struct stream_opcode {
31 int32_t thread_slot; // the dispatcher id this message refers to
32 uint32_t session; // random number used to verify that the message the dispatcher receives is for this sender
33 STREAM_OPCODE opcode; // the actual message to be delivered
34 STREAM_HANDSHAKE reason;
35 struct pollfd_meta *meta;
36 };
37
38 // IMPORTANT: to add workers, you have to edit WORKER_PARSER_FIRST_JOB accordingly
39
40 // stream thread events
41 #define WORKER_STREAM_JOB_LIST 0
42 #define WORKER_STREAM_JOB_DEQUEUE 1
43 #define WORKER_STREAM_JOB_PREP 2
44 #define WORKER_STREAM_JOB_POLL_ERROR 3
45 #define WORKER_SENDER_JOB_PIPE_READ 4
46
47 // socket operations
48 #define WORKER_STREAM_JOB_SOCKET_RECEIVE 5
49 #define WORKER_STREAM_JOB_SOCKET_SEND 6
50
51 // compression
52 #define WORKER_STREAM_JOB_COMPRESS 7
53 #define WORKER_STREAM_JOB_DECOMPRESS 8
54
55 // receiver events
56 #define WORKER_RECEIVER_JOB_BYTES_READ 9
57 #define WORKER_RECEIVER_JOB_BYTES_UNCOMPRESSED 10
58
59 // sender received commands
60 #define WORKER_SENDER_JOB_EXECUTE 11
61 #define WORKER_SENDER_JOB_EXECUTE_REPLAY 12
62 #define WORKER_SENDER_JOB_EXECUTE_FUNCTION 13
63 #define WORKER_SENDER_JOB_EXECUTE_META 14
64
65 // disconnect reasons
66 #define WORKER_STREAM_JOB_DISCONNECT_REMOTE_CLOSED 15
67 #define WORKER_STREAM_JOB_DISCONNECT_RECEIVE_ERROR 16
68 #define WORKER_STREAM_JOB_DISCONNECT_SEND_ERROR 17
69 #define WORKER_STREAM_JOB_DISCONNECT_TIMEOUT 18
70 #define WORKER_STREAM_JOB_DISCONNECT_SOCKET_ERROR 19
71
72 // sender-only disconnect reasons
73 #define WORKER_SENDER_JOB_DISCONNECT_OVERFLOW 20
74 #define WORKER_SENDER_JOB_DISCONNECT_COMPRESSION_ERROR 21
75 #define WORKER_SENDER_JOB_DISCONNECT_RECEIVER_LEFT 22
76 #define WORKER_SENDER_JOB_DISCONNECT_HOST_CLEANUP 23
77
78 // dispatcher metrics
79 // this has to be the same at pluginsd_parser.h
80 #define WORKER_RECEIVER_JOB_REPLICATION_COMPLETION 24
81 #define WORKER_STREAM_METRIC_NODES 25
82 #define WORKER_SENDER_JOB_BUFFER_RATIO 26
83 #define WORKER_SENDER_JOB_BYTES_RECEIVED 27
84 #define WORKER_SENDER_JOB_BYTES_SENT 28
85 #define WORKER_SENDER_JOB_BYTES_COMPRESSED 29
86 #define WORKER_SENDER_JOB_BYTES_UNCOMPRESSED 30
87 #define WORKER_SENDER_JOB_BYTES_COMPRESSION_RATIO 31
88 #define WORKER_SENDER_JOB_REPLAY_DICT_SIZE 32
89 #define WORKER_SENDER_JOB_MESSAGES 33
90 #define WORKER_STREAM_JOB_RECEIVERS_WAITING_LIST_SIZE 34
91 #define WORKER_STREAM_JOB_SEND_MISSES 35
92
93 // IMPORTANT: to add workers, you have to edit WORKER_PARSER_FIRST_JOB accordingly
94
95 #if WORKER_UTILIZATION_MAX_JOB_TYPES < 36
96 #error WORKER_UTILIZATION_MAX_JOB_TYPES has to be at least 34
97 #endif
98
99 #define STREAM_MAX_THREADS 2048
100 #define THREAD_TAG_STREAM "STREAM"
101
102 typedef enum {
103 EVLOOP_STATUS_CONTINUE,
104 EVLOOP_STATUS_SOCKET_FULL,
105 EVLOOP_STATUS_SOCKET_CLOSED,
106 EVLOOP_STATUS_SOCKET_ERROR,
107 EVLOOP_STATUS_NO_MORE_DATA,
108 EVLOOP_STATUS_OPCODE_ON_ME,
109 EVLOOP_STATUS_CANT_GET_LOCK,
110 EVLOOP_STATUS_PARSER_FAILED,
111 } EVLOOP_STATUS;
112
113 #define EVLOOP_STATUS_STILL_ALIVE(status) (status == EVLOOP_STATUS_CONTINUE || status == EVLOOP_STATUS_NO_MORE_DATA || status == EVLOOP_STATUS_SOCKET_FULL || status == EVLOOP_STATUS_CANT_GET_LOCK)
114
115 typedef enum {
116 POLLFD_TYPE_EMPTY,
117 POLLFD_TYPE_SENDER,
118 POLLFD_TYPE_RECEIVER,
119 POLLFD_TYPE_PIPE,
120 } POLLFD_TYPE;
121
122 struct pollfd_meta {
123 POLLFD_TYPE type;
124 union {
125 struct receiver_state *rpt;
126 struct sender_state *s;
127 };
128 };
129
130 DEFINE_JUDYL_TYPED(SENDERS, struct sender_state *);
131 DEFINE_JUDYL_TYPED(RECEIVERS, struct receiver_state *);
132 DEFINE_JUDYL_TYPED(META, struct pollfd_meta *);
133
134 struct stream_thread {
135 ND_THREAD *thread;
136
137 pid_t tid;
138 size_t id;
139 size_t nodes_count;
140
141 struct {
142 size_t bytes_received;
143 size_t bytes_sent;
144 size_t send_misses;
145 } snd;
146
147 struct {
148 size_t bytes_received;
149 size_t bytes_received_uncompressed;
150 NETDATA_DOUBLE replication_completion;
151 } rcv;
152
153 struct {
154 SPINLOCK spinlock; // ensure a single writer at a time
155 int fds[2];
156 size_t size;
157 char *buffer;
158 } pipe;
159
160 struct {
161 // the incoming queue of the dispatcher thread
162 // the connector thread leaves the connected senders in this list, for the dispatcher to pick them up
163 SPINLOCK spinlock;
164 Word_t id;
165 SENDERS_JudyLSet senders;
166 RECEIVERS_JudyLSet receivers;
167
168 size_t receivers_waiting;
169 } queue;
170
171 struct {
172 usec_t last_accepted_ut;
173 size_t metadata;
174 size_t replication;
175 } waiting_list;
176
177 struct {
178 SPINLOCK spinlock;
179 size_t added;
180 size_t processed;
181 size_t bypassed;
182 size_t size;
183 size_t used;
184 struct stream_opcode *array; // the array of messages from the senders
185 struct stream_opcode *copy; // a copy of the array of messages from the senders, to work on
186 } messages;
187
188 struct {
189 nd_poll_t *ndpl;
190 struct pollfd_meta pipe;
191 META_JudyLSet meta;
192 } run;
193 };
194
195 struct stream_thread_globals {
196 struct {
197 SPINLOCK spinlock;
198 size_t id;
199 size_t cores;
200 } assign;
201
202 struct stream_thread threads[STREAM_MAX_THREADS];
203 };
204
205 struct rrdhost;
206 extern struct stream_thread_globals stream_thread_globals;
207
208 void stream_sender_move_queue_to_running_unsafe(struct stream_thread *sth);
209 void stream_receiver_move_entire_queue_to_running_unsafe(struct stream_thread *sth);
210 void stream_sender_check_all_nodes_from_poll(struct stream_thread *sth, usec_t now_ut);
211 void stream_sender_replication_check_from_poll(struct stream_thread *sth, usec_t now_ut);
212
213 void stream_receiver_add_to_queue(struct receiver_state *rpt);
214 void stream_sender_add_to_connector_queue(struct rrdhost *host);
215
216 bool stream_sender_process_poll_events(struct stream_thread *sth, struct sender_state *s, nd_poll_event_t events, usec_t now_ut);
217 bool stream_receive_process_poll_events(struct stream_thread *sth, struct receiver_state *rpt, nd_poll_event_t events, usec_t now_ut);
218
219 void stream_sender_cleanup(struct stream_thread *sth);
220 void stream_receiver_cleanup(struct stream_thread *sth);
221 void stream_sender_handle_op(struct stream_thread *sth, struct sender_state *s, struct stream_opcode *msg);
222
223 struct stream_thread *stream_thread_by_slot_id(size_t thread_slot);
224
225 void stream_thread_node_queued(struct rrdhost *host);
226 void stream_thread_node_removed(struct rrdhost *host);
227
228 // returns true if my_meta has received a message
229 bool stream_thread_process_opcodes(struct stream_thread *sth, struct pollfd_meta *my_meta);
230
231 void stream_receiver_move_to_running_unsafe(struct stream_thread *sth, struct receiver_state *rpt);
232
233 bool stream_sender_receive_data(struct stream_thread *sth, struct sender_state *s, usec_t now_ut, bool process_opcodes);
234 bool stream_sender_send_data(struct stream_thread *sth, struct sender_state *s, usec_t now_ut, bool process_opcodes_and_enable_removal);
235
236 bool stream_receiver_receive_data(struct stream_thread *sth, struct receiver_state *rpt, usec_t now_ut, bool process_opcodes);
237 bool stream_receiver_send_data(struct stream_thread *sth, struct receiver_state *rpt, usec_t now_ut, bool process_opcodes_and_enable_removal);
238
239 #include "stream-sender-internals.h"
240 #include "stream-receiver-internals.h"
241 #include "plugins.d/pluginsd_parser.h"
242
243 static inline bool rrdhost_is_this_a_stream_thread(RRDHOST *host) {
244 pid_t tid = gettid_cached();
245 return host->stream.rcv.status.tid == tid || host->stream.snd.status.tid == tid;
246 }
247
248 #endif //NETDATA_STREAM_THREAD_H