master
c 580 lines 20 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "daemon/static_threads.h"
4 #include "libnetdata/libnetdata.h"
5
6 static inline void poll_process_updated_events(POLLINFO *pi) {
7 if(pi->events != pi->events_we_wait_for && !(pi->flags & POLLINFO_FLAG_REMOVED_FROM_POLL)) {
8 if(!nd_poll_upd(pi->p->ndpl, pi->fd, pi->events))
9 nd_log(NDLS_DAEMON, NDLP_ERR, "Failed to update socket %d to nd_poll", pi->fd);
10 pi->events_we_wait_for = pi->events;
11 }
12 }
13
14 // poll() based listener
15 // this should be the fastest possible listener for up to 100 sockets
16 // above 100, an epoll() interface is needed on Linux
17
18 POLLINFO *poll_add_fd(POLLJOB *p
19 , int fd
20 , int socktype
21 , HTTP_ACL port_acl
22 , uint32_t flags
23 , const char *client_ip
24 , const char *client_port
25 , const char *client_host
26 , poll_events_add_callback_t add_callback
27 , poll_events_del_callback_t del_callback
28 , poll_events_rcv_callback_t rcv_callback
29 , poll_events_snd_callback_t snd_callback
30 , void *data
31 ) {
32 if(unlikely(fd < 0)) return NULL;
33
34 //if(p->limit && p->used >= p->limit) {
35 // nd_log(NDLS_DAEMON, NDLP_WARNING, "Max sockets limit reached (%zu sockets), dropping connection", p->used);
36 // close(fd);
37 // return NULL;
38 //}
39
40 POLLINFO *pi = callocz(1, sizeof(*pi));
41
42 pi->fd = fd;
43 pi->events = ND_POLL_READ;
44 pi->p = p;
45 pi->socktype = socktype;
46 pi->port_acl = port_acl;
47 pi->flags = flags;
48 pi->client_ip = strdupz(client_ip);
49 pi->client_port = strdupz(client_port);
50 pi->client_host = strdupz(client_host);
51
52 pi->del_callback = del_callback;
53 pi->rcv_callback = rcv_callback;
54 pi->snd_callback = snd_callback;
55
56 pi->connected_t = now_boottime_sec();
57 pi->last_received_t = 0;
58 pi->last_sent_t = 0;
59 pi->last_sent_t = 0;
60 pi->recv_count = 0;
61 pi->send_count = 0;
62
63 p->used++;
64
65 if(pi->flags & POLLINFO_FLAG_CLIENT_SOCKET)
66 pi->data = add_callback(pi, &pi->events, data);
67
68 DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(p->ll, pi, prev, next);
69 pi->events_we_wait_for = pi->events;
70 if(!nd_poll_add(pi->p->ndpl, pi->fd, pi->events, pi))
71 nd_log(NDLS_DAEMON, NDLP_ERR, "Failed to add socket %d to nd_poll", pi->fd);
72
73 return pi;
74 }
75
76 void poll_process_remove_from_poll(POLLINFO *pi) {
77 POLLJOB *p = pi->p;
78
79 if(!nd_poll_del(p->ndpl, pi->fd))
80 nd_log(NDLS_DAEMON, NDLP_ERR,
81 "Failed to delete socket %d from nd_poll() - is the socket already closed?", pi->fd);
82 else
83 pi->flags |= POLLINFO_FLAG_REMOVED_FROM_POLL;
84 }
85
86 static inline void poll_close_fd(POLLINFO *pi, const char *func) {
87 POLLJOB *p = pi->p;
88
89 DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(p->ll, pi, prev, next);
90 if(!(pi->flags & POLLINFO_FLAG_REMOVED_FROM_POLL) && !nd_poll_del(p->ndpl, pi->fd))
91 nd_log(NDLS_DAEMON, NDLP_ERR,
92 "Failed to delete socket %d from nd_poll() - called from %s() - is the socket already closed?",
93 pi->fd, func);
94 else
95 pi->flags |= POLLINFO_FLAG_REMOVED_FROM_POLL;
96
97 if(pi->flags & POLLINFO_FLAG_CLIENT_SOCKET) {
98 pi->del_callback(pi);
99
100 if(likely(!(pi->flags & POLLINFO_FLAG_DONT_CLOSE))) {
101 if(close(pi->fd) == -1)
102 nd_log(NDLS_DAEMON, NDLP_ERR,
103 "Failed to close() poll_events() socket %d",
104 pi->fd);
105 }
106 }
107
108 freez(pi->client_ip);
109 freez(pi->client_port);
110 freez(pi->client_host);
111 freez(pi);
112
113 p->used--;
114 }
115
116 void *poll_default_add_callback(POLLINFO *pi __maybe_unused, nd_poll_event_t *events __maybe_unused, void *data __maybe_unused) {
117 return NULL;
118 }
119
120 void poll_default_del_callback(POLLINFO *pi) {
121 if(pi->data)
122 nd_log(NDLS_DAEMON, NDLP_ERR,
123 "POLLFD: internal error: del_callback_default() called with data pointer - possible memory leak");
124 }
125
126 int poll_default_rcv_callback(POLLINFO *pi, nd_poll_event_t *events) {
127 *events |= ND_POLL_READ;
128
129 char buffer[1024 + 1];
130
131 ssize_t rc;
132 do {
133 rc = recv(pi->fd, buffer, 1024, MSG_DONTWAIT);
134 if (rc < 0) {
135 // read failed
136 if (errno != EWOULDBLOCK && errno != EAGAIN) {
137 nd_log(NDLS_DAEMON, NDLP_ERR,
138 "POLLFD: poll_default_rcv_callback(): recv() failed with %zd.",
139 rc);
140
141 return -1;
142 }
143 } else if (rc) {
144 // data received
145 nd_log(NDLS_DAEMON, NDLP_WARNING,
146 "POLLFD: internal error: poll_default_rcv_callback() is discarding %zd bytes received on socket %d",
147 rc, pi->fd);
148 }
149 } while (rc != -1);
150
151 return 0;
152 }
153
154 int poll_default_snd_callback(POLLINFO *pi, nd_poll_event_t *events) {
155 *events &= ~ND_POLL_WRITE;
156
157 nd_log(NDLS_DAEMON, NDLP_WARNING,
158 "POLLFD: internal error: poll_default_snd_callback(): nothing to send on socket %d",
159 pi->fd);
160
161 return 0;
162 }
163
164 void poll_default_tmr_callback(void *timer_data) {
165 (void)timer_data;
166 }
167
168 static void poll_events_cleanup(void *pptr) {
169 POLLJOB *p = CLEANUP_FUNCTION_GET_PTR(pptr);
170 if(!p) return;
171
172 while(p->ll) {
173 POLLINFO *pi = p->ll;
174 pi->flags &= ~(POLLINFO_FLAG_DONT_CLOSE);
175 poll_close_fd(pi, __FUNCTION__ );
176 }
177
178 nd_poll_destroy(p->ndpl);
179 p->ndpl = NULL;
180 }
181
182 static int poll_process_error(POLLINFO *pi, nd_poll_event_t revents) {
183 ND_LOG_STACK lgs[] = {
184 ND_LOG_FIELD_TXT(NDF_SRC_IP, pi->client_ip),
185 ND_LOG_FIELD_TXT(NDF_SRC_PORT, pi->client_port),
186 ND_LOG_FIELD_END(),
187 };
188 ND_LOG_STACK_PUSH(lgs);
189
190 nd_log(NDLS_DAEMON, NDLP_DEBUG,
191 "POLLFD: LISTENER: received %s %s %s on socket %d client '%s' port '%s' expecting %s %s, having %s %s"
192 , revents & ND_POLL_ERROR ? "ERROR" : ""
193 , revents & ND_POLL_HUP ? "HUP" : ""
194 , revents & ND_POLL_INVALID ? "INVALID" : ""
195 , pi->fd
196 , pi->client_ip ? pi->client_ip : "<undefined-ip>"
197 , pi->client_port ? pi->client_port : "<undefined-port>"
198 , pi->events & ND_POLL_READ ? "READ" : "", pi->events & ND_POLL_WRITE ? "WRITE" : ""
199 , revents & ND_POLL_READ ? "READ" : "", revents & ND_POLL_WRITE ? "WRITE" : ""
200 );
201
202 poll_close_fd(pi, __FUNCTION__ );
203 return 1;
204 }
205
206 static inline int poll_process_send(POLLINFO *pi, time_t now) {
207 pi->last_sent_t = now;
208 pi->send_count++;
209
210 pi->events = 0;
211
212 if (unlikely(pi->snd_callback(pi, &pi->events) == -1))
213 poll_close_fd(pi, __FUNCTION__ );
214 else
215 poll_process_updated_events(pi);
216
217 return 1;
218 }
219
220 static inline int poll_process_tcp_read(POLLINFO *pi, time_t now) {
221 pi->last_received_t = now;
222 pi->recv_count++;
223
224 pi->events = 0;
225
226 if (pi->rcv_callback(pi, &pi->events) == -1)
227 poll_close_fd(pi, __FUNCTION__ );
228 else
229 poll_process_updated_events(pi);
230
231 return 1;
232 }
233
234 static inline int poll_process_udp_read(POLLINFO *pi, time_t now __maybe_unused) {
235 pi->last_received_t = now;
236 pi->recv_count++;
237
238 // TODO: access_list is not applied to UDP
239 // but checking the access list on every UDP packet will destroy
240 // performance, especially for statsd.
241
242 pi->events = 0;
243
244 if(pi->rcv_callback(pi, &pi->events) == -1)
245 return 0;
246 else {
247 poll_process_updated_events(pi);
248 return 1;
249 }
250 }
251
252 static int poll_process_new_tcp_connection(POLLINFO *pi, time_t now) {
253 POLLJOB *p = pi->p;
254
255 pi->last_received_t = now;
256 pi->recv_count++;
257
258 char client_ip[INET6_ADDRSTRLEN] = "";
259 char client_port[NI_MAXSERV] = "";
260 char client_host[NI_MAXHOST] = "";
261
262 #ifdef SOCK_NONBLOCK
263 int flags = SOCK_NONBLOCK;
264 #else
265 int flags = 0;
266 #endif
267
268 int nfd = accept_socket(
269 pi->fd, flags,
270 client_ip, INET6_ADDRSTRLEN, client_port,NI_MAXSERV, client_host, NI_MAXHOST,
271 p->access_list, p->allow_dns
272 );
273
274 #ifndef SOCK_NONBLOCK
275 if (nfd > 0) {
276 int flags = fcntl(nfd, F_GETFL);
277 (void)fcntl(nfd, F_SETFL, flags| O_NONBLOCK);
278 }
279 #endif
280
281 if (unlikely(nfd < 0)) {
282 // accept failed
283
284 if(unlikely(errno == EMFILE)) {
285 nd_log_limit_static_global_var(erl, 10, 1000);
286 nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR,
287 "POLLFD: LISTENER: too many open files - used by this thread %zu, max for this thread %zu",
288 p->used, p->limit);
289 }
290 else if(unlikely(errno != EWOULDBLOCK && errno != EAGAIN))
291 nd_log(NDLS_DAEMON, NDLP_ERR,
292 "POLLFD: LISTENER: accept() failed.");
293
294 }
295 else if(is_socket_closed(nfd))
296 close(nfd);
297
298 else {
299 // accept ok
300
301 poll_add_fd(p
302 , nfd
303 , SOCK_STREAM
304 , pi->port_acl
305 , POLLINFO_FLAG_CLIENT_SOCKET
306 , client_ip
307 , client_port
308 , client_host
309 , p->add_callback
310 , p->del_callback
311 , p->rcv_callback
312 , p->snd_callback
313 , NULL
314 );
315
316 return 1;
317 }
318
319 return 0;
320 }
321
322 void poll_events(LISTEN_SOCKETS *sockets
323 , poll_events_add_callback_t add_callback
324 , poll_events_del_callback_t del_callback
325 , poll_events_rcv_callback_t rcv_callback
326 , poll_events_snd_callback_t snd_callback
327 , poll_events_tmr_callback_t tmr_callback
328 , bool (*check_to_stop_callback)(void)
329 , SIMPLE_PATTERN *access_list
330 , int allow_dns
331 , void *data
332 , time_t tcp_request_timeout_seconds
333 , time_t tcp_idle_timeout_seconds
334 , time_t timer_milliseconds
335 , void *timer_data
336 , size_t max_tcp_sockets
337 ) {
338 if(!sockets || !sockets->opened) {
339 nd_log(NDLS_DAEMON, NDLP_ERR,
340 "POLLFD: internal error: no listening sockets are opened");
341 return;
342 }
343
344 if(timer_milliseconds <= 0) timer_milliseconds = 0;
345
346 int retval;
347
348 POLLJOB p = {
349 .ndpl = nd_poll_create(),
350 .used = 0,
351 .limit = max_tcp_sockets,
352
353 .complete_request_timeout = tcp_request_timeout_seconds,
354 .idle_timeout = tcp_idle_timeout_seconds,
355 .checks_every = (tcp_idle_timeout_seconds / 3) + 1,
356
357 .access_list = access_list,
358 .allow_dns = allow_dns,
359
360 .timer_milliseconds = timer_milliseconds,
361 .timer_data = timer_data,
362
363 .add_callback = add_callback?add_callback:poll_default_add_callback,
364 .del_callback = del_callback?del_callback:poll_default_del_callback,
365 .rcv_callback = rcv_callback?rcv_callback:poll_default_rcv_callback,
366 .snd_callback = snd_callback?snd_callback:poll_default_snd_callback,
367 .tmr_callback = tmr_callback?tmr_callback:poll_default_tmr_callback
368 };
369
370 size_t i;
371 for(i = 0; i < sockets->opened ;i++) {
372
373 POLLINFO *pi = poll_add_fd(&p
374 , sockets->fds[i]
375 , sockets->fds_types[i]
376 , sockets->fds_acl_flags[i]
377 , POLLINFO_FLAG_SERVER_SOCKET
378 , (sockets->fds_names[i])?sockets->fds_names[i]:"UNKNOWN"
379 , ""
380 , ""
381 , p.add_callback
382 , p.del_callback
383 , p.rcv_callback
384 , p.snd_callback
385 , NULL
386 );
387
388 pi->data = data;
389 nd_log(NDLS_DAEMON, NDLP_DEBUG,
390 "POLLFD: LISTENER: listening on '%s'",
391 (sockets->fds_names[i])?sockets->fds_names[i]:"UNKNOWN");
392 }
393
394 int listen_sockets_active = 1;
395
396 time_t last_check = now_boottime_sec();
397
398 usec_t timer_usec = timer_milliseconds * USEC_PER_MS;
399 usec_t now_usec = 0, next_timer_usec = 0, last_timer_usec = 0;
400 (void)last_timer_usec;
401
402 if(unlikely(timer_usec)) {
403 now_usec = now_boottime_usec();
404 next_timer_usec = now_usec - (now_usec % timer_usec) + timer_usec;
405 }
406
407 CLEANUP_FUNCTION_REGISTER(poll_events_cleanup) cleanup_ptr = &p;
408
409 size_t iteration_counter = 0,
410 timeout_counter = 0,
411 errors_counter = 0,
412 read_counter = 0,
413 writes_counter = 0,
414 unhandled_counter = 0,
415 cleanup_counter = 0;
416
417 while(!check_to_stop_callback() && !nd_thread_signaled_to_cancel()) {
418 if(unlikely(timer_usec)) {
419 now_usec = now_boottime_usec();
420
421 if(unlikely(timer_usec && now_usec >= next_timer_usec)) {
422 last_timer_usec = now_usec;
423 p.tmr_callback(p.timer_data);
424 now_usec = now_boottime_usec();
425 next_timer_usec = now_usec - (now_usec % timer_usec) + timer_usec;
426 }
427 }
428
429 // enable or disable the TCP listening sockets, based on the current number of sockets used and the limit set
430 if((listen_sockets_active && (p.limit && p.used >= p.limit)) || (!listen_sockets_active && (!p.limit || p.used < p.limit))) {
431 listen_sockets_active = !listen_sockets_active;
432
433 nd_log(NDLS_DAEMON, NDLP_DEBUG,
434 "%s listening sockets (used TCP sockets %zu, max allowed for this worker %zu)",
435 (listen_sockets_active)?"ENABLING":"DISABLING", p.used, p.limit);
436
437 for(POLLINFO *pi = p.ll; pi ; pi = pi->next) {
438 if((pi->flags & POLLINFO_FLAG_SERVER_SOCKET) && pi->socktype == SOCK_STREAM) {
439 pi->events = (short int) ((listen_sockets_active) ? ND_POLL_READ : 0);
440 poll_process_updated_events(pi);
441 }
442 }
443 }
444
445 nd_poll_result_t result;
446 retval = nd_poll_wait(p.ndpl, ND_CHECK_CANCELLABILITY_WHILE_WAITING_EVERY_MS, &result);
447 iteration_counter++;
448 time_t now = now_boottime_sec();
449
450 if(unlikely(retval == -1)) {
451 nd_log(NDLS_DAEMON, NDLP_ERR, "POLLFD: LISTENER: nd_poll_wait() failed.");
452 break;
453 }
454 else if(unlikely(!retval)) {
455 timeout_counter++;
456 // timeout
457 ;
458 }
459 else {
460 POLLINFO *pi = (POLLINFO *)result.data;
461
462 if(result.events & (ND_POLL_HUP | ND_POLL_INVALID | ND_POLL_ERROR)) {
463 errors_counter++;
464 poll_process_error(pi, result.events);
465 }
466 else if(result.events & ND_POLL_WRITE) {
467 writes_counter++;
468 poll_process_send(pi, now);
469 }
470 else if(result.events & ND_POLL_READ) {
471 read_counter++;
472 if (pi->flags & POLLINFO_FLAG_CLIENT_SOCKET) {
473 if (pi->socktype == SOCK_DGRAM)
474 poll_process_udp_read(pi, now);
475 else if (pi->socktype == SOCK_STREAM)
476 poll_process_tcp_read(pi, now);
477 else {
478 nd_log(NDLS_DAEMON, NDLP_ERR,
479 "POLLFD: LISTENER: server slot %zu (fd %d) connection from %s port %s using unhandled socket type %d.",
480 i,
481 pi->fd,
482 pi->client_ip ? pi->client_ip : "<undefined-ip>",
483 pi->client_port ? pi->client_port : "<undefined-port>",
484 pi->socktype);
485
486 poll_close_fd(pi, "poll_events1");
487 }
488 }
489 else if (pi->flags & POLLINFO_FLAG_SERVER_SOCKET) {
490 if(pi->socktype == SOCK_DGRAM)
491 poll_process_udp_read(pi, now);
492
493 else if(pi->socktype == SOCK_STREAM) {
494 if (!p.limit || p.used < p.limit)
495 poll_process_new_tcp_connection(pi, now);
496 }
497 else {
498 nd_log(NDLS_DAEMON, NDLP_ERR,
499 "POLLFD: LISTENER: server slot %zu (fd %d) connection from %s port %s using unhandled socket type %d.",
500 i,
501 pi->fd,
502 pi->client_ip ? pi->client_ip : "<undefined-ip>",
503 pi->client_port ? pi->client_port : "<undefined-port>",
504 pi->socktype);
505
506 poll_close_fd(pi, "poll_events2");
507 }
508 }
509 else {
510 nd_log(NDLS_DAEMON, NDLP_ERR,
511 "POLLFD: LISTENER: client slot %zu (fd %d) data from %s port %s using flags %08X is neither client nor server."
512 , i
513 , pi->fd
514 , pi->client_ip ? pi->client_ip : "<undefined-ip>"
515 , pi->client_port ? pi->client_port : "<undefined-port>"
516 , pi->flags
517 );
518
519 poll_close_fd(pi, "poll_events3");
520 }
521 }
522 else {
523 unhandled_counter++;
524
525 nd_log(NDLS_DAEMON, NDLP_ERR,
526 "POLLFD: LISTENER: socket slot %zu (fd %d) client %s port %s unhandled event id %d."
527 , i
528 , pi->fd
529 , pi->client_ip ? pi->client_ip : "<undefined-ip>"
530 , pi->client_port ? pi->client_port : "<undefined-port>"
531 , (int)result.events
532 );
533
534 poll_close_fd(pi, "poll_events4");
535 }
536 }
537
538 if(unlikely(p.checks_every > 0 && now - last_check > p.checks_every)) {
539 cleanup_counter++;
540
541 last_check = now;
542
543 // cleanup old sockets
544 POLLINFO *pi, *next = NULL;
545 for(pi = p.ll; pi ; pi = next) {
546 next = pi->next;
547
548 if(likely(pi->flags & POLLINFO_FLAG_CLIENT_SOCKET)) {
549 if (unlikely(
550 !(pi->flags & POLLINFO_FLAG_FIRST_REQUEST_RECEIVED) &&
551 pi->send_count == 0 &&
552 p.complete_request_timeout > 0 &&
553 (now - pi->connected_t) >= p.complete_request_timeout
554 )) {
555 nd_log(NDLS_DAEMON, NDLP_DEBUG,
556 "POLLFD: LISTENER: client slot %zu (fd %d) from %s port %s has not completed its first request in %zu seconds - closing it. "
557 , i
558 , pi->fd
559 , pi->client_ip ? pi->client_ip : "<undefined-ip>"
560 , pi->client_port ? pi->client_port : "<undefined-port>"
561 , (size_t) p.complete_request_timeout
562 );
563 poll_close_fd(pi, "poll_events4");
564 }
565 else if(unlikely(pi->recv_count && p.idle_timeout > 0 && now - ((pi->last_received_t > pi->last_sent_t) ? pi->last_received_t : pi->last_sent_t) >= p.idle_timeout )) {
566 nd_log(NDLS_DAEMON, NDLP_DEBUG,
567 "POLLFD: LISTENER: client slot %zu (fd %d) from %s port %s is idle for more than %zu seconds - closing it. "
568 , i
569 , pi->fd
570 , pi->client_ip ? pi->client_ip : "<undefined-ip>"
571 , pi->client_port ? pi->client_port : "<undefined-port>"
572 , (size_t) p.idle_timeout
573 );
574 poll_close_fd(pi, "poll_events5");
575 }
576 }
577 }
578 }
579 }
580 }