The FreeRADIUS server $Id: 15bac2a4c627c01d1aa2047687b3418955ac7f00 $
Loading...
Searching...
No Matches
network.c
Go to the documentation of this file.
1/*
2 * This program is free software; you can redistribute it and/or modify
3 * it under the terms of the GNU General Public License as published by
4 * the Free Software Foundation; either version 2 of the License, or
5 * (at your option) any later version.
6 *
7 * This program is distributed in the hope that it will be useful,
8 * but WITHOUT ANY WARRANTY; without even the implied warranty of
9 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
10 * GNU General Public License for more details.
11 *
12 * You should have received a copy of the GNU General Public License
13 * along with this program; if not, write to the Free Software
14 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
15 */
16
17/**
18 * $Id: f7828609b9aac5a722c9ffb34561b1184d5986de $
19 *
20 * @brief Receiver of socket data, which sends messages to the workers.
21 * @file io/network.c
22 *
23 * @copyright 2016 Alan DeKok (aland@freeradius.org)
24 */
25RCSID("$Id: f7828609b9aac5a722c9ffb34561b1184d5986de $")
26
27#define LOG_PREFIX nr->name
28
29#define LOG_DST nr->log
30
31#include <freeradius-devel/util/event.h>
32#include <freeradius-devel/util/misc.h>
33#include <freeradius-devel/util/rand.h>
34#include <freeradius-devel/util/rb.h>
35#include <freeradius-devel/util/syserror.h>
36#include <freeradius-devel/util/atexit.h>
37#include <freeradius-devel/util/talloc.h>
38
39#include <freeradius-devel/io/channel.h>
40#include <freeradius-devel/io/control.h>
41#include <freeradius-devel/io/listen.h>
42#include <freeradius-devel/io/network.h>
43#include <freeradius-devel/io/queue.h>
44#include <freeradius-devel/io/ring_buffer.h>
45#include <freeradius-devel/io/worker.h>
46
47#define MAX_WORKERS 64
48
49static _Thread_local fr_ring_buffer_t *fr_network_rb;
50
57
58/** Associate a worker thread with a network thread
59 *
60 */
61typedef struct {
62 fr_heap_index_t heap_id; //!< workers are in a heap
63 fr_time_delta_t cpu_time; //!< how much CPU time this worker has spent
64 fr_time_delta_t predicted; //!< predicted processing time for one packet
65
66 bool blocked; //!< is this worker blocked?
67
68 fr_channel_t *channel; //!< channel to the worker
69 fr_worker_t *worker; //!< worker pointer
72
73typedef struct {
74 fr_rb_node_t listen_node; //!< rbtree node for looking up by listener.
75 fr_rb_node_t num_node; //!< rbtree node for looking up by number.
76
77 fr_network_t *nr; //!< O(N) issues in talloc
78 int number; //!< unique ID
79 fr_heap_index_t heap_id; //!< for the sockets_by_num heap
80
81 fr_event_filter_t filter; //!< what type of filter it is
82
83 bool dead; //!< is it dead?
84 bool blocked; //!< is it blocked?
85
86 unsigned int outstanding; //!< number of outstanding packets sent to the worker
87 fr_listen_t *listen; //!< I/O ctx and functions.
88
89 fr_message_set_t *ms; //!< message buffers for this socket.
90 fr_channel_data_t *cd; //!< cached in case of allocation & read error
91 size_t leftover; //!< leftover data from a previous read
92 size_t written; //!< however much we did in a partial write
93
94 fr_channel_data_t *pending; //!< the currently pending partial packet
95 fr_heap_t *waiting; //!< packets waiting to be written
98
99/*
100 * We have an array of workers, so we can index the workers in
101 * O(1) time. remove the heap of "workers ordered by CPU time"
102 * when we send a packet to a worker, just update the predicted
103 * CPU time in place. when we receive a reply from a worker,
104 * just update the predicted CPU time in place.
105 *
106 * when we need to choose a worker, pick 2 at random, and then
107 * choose the one with the lowe cpu time. For background, see
108 * "Power of Two-Choices" and
109 * https://www.eecs.harvard.edu/~michaelm/postscripts/mythesis.pdf
110 * https://www.eecs.harvard.edu/~michaelm/postscripts/tpds2001.pdf
111 */
113 char const *name; //!< Network ID for logging.
114
115 pthread_t thread_id; //!< for self
116
117 bool suspended; //!< whether or not we're suspended.
118
119 fr_log_t const *log; //!< log destination
120 fr_log_lvl_t lvl; //!< debug log level
121
122 fr_atomic_queue_t *aq_control; //!< atomic queue for control messages sent to me
123
124 fr_control_t *control; //!< the control plane
125
126 fr_ring_buffer_t *rb; //!< ring buffer for my control-plane messages
127
128 fr_event_list_t *el; //!< our event list
129
130 fr_heap_t *replies; //!< replies from the worker, ordered by priority / origin time
131
133
134 fr_rb_tree_t *sockets; //!< list of sockets we're managing, ordered by the listener
135 fr_rb_tree_t *sockets_by_num; //!< ordered by number;
136
137 int num_workers; //!< number of active workers
138 int num_blocked; //!< number of blocked workers
139 int num_pending_workers; //!< number of workers we're waiting to start.
140 int max_workers; //!< maximum number of allowed workers
141 int num_sockets; //!< actually a counter...
142
143 int signal_pipe[2]; //!< Pipe for signalling the worker in an orderly way.
144 ///< This is more deterministic than using async signals.
145
146 bool exiting; //!< are we exiting?
147
148 fr_network_config_t config; //!< configuration
150};
151
152static void fr_network_post_event(fr_event_list_t *el, fr_time_t now, void *uctx);
153static int fr_network_pre_event(fr_time_t now, fr_time_delta_t wake, void *uctx);
155static void fr_network_read(UNUSED fr_event_list_t *el, int sockfd, UNUSED int flags, void *ctx);
156
157static int8_t reply_cmp(void const *one, void const *two)
158{
159 fr_channel_data_t const *a = one, *b = two;
160 int ret;
161
162 ret = CMP(a->priority, b->priority);
163 if (ret != 0) return ret;
164
165 return fr_time_cmp(a->m.when, b->m.when);
166}
167
168static int8_t waiting_cmp(void const *one, void const *two)
169{
170 fr_channel_data_t const *a = one, *b = two;
171 int ret;
172
173 ret = CMP(a->priority, b->priority);
174 if (ret != 0) return ret;
175
176 return fr_time_cmp(a->reply.request_time, b->reply.request_time);
177}
178
179static int8_t socket_listen_cmp(void const *one, void const *two)
180{
181 fr_network_socket_t const *a = one, *b = two;
182
183 return CMP(a->listen, b->listen);
184}
185
186static int8_t socket_num_cmp(void const *one, void const *two)
187{
188 fr_network_socket_t const *a = one, *b = two;
189
190 return CMP(a->number, b->number);
191}
192
193/*
194 * Explicitly cleanup the memory allocated to the ring buffer,
195 * just in case valgrind complains about it.
196 */
197static int _fr_network_rb_free(void *arg)
198{
199 return talloc_free(arg);
200}
201
202/** Initialise thread local storage
203 *
204 * @return fr_ring_buffer_t for messages
205 */
207{
209
211 if (rb) return rb;
212
214 if (!rb) {
215 fr_perror("Failed allocating memory for network ring buffer");
216 return NULL;
217 }
218
220
221 return rb;
222}
223
224static inline bool is_network_thread(fr_network_t const *nr)
225{
226 return (pthread_equal(pthread_self(), nr->thread_id) != 0);
227}
228
230
231/** Add a fr_listen_t to a network
232 *
233 * @param nr the network
234 * @param li the listener
235 */
237{
239
240 /*
241 * Associate the protocol dictionary with the listener, so that the decode functions can check /
242 * use it.
243 *
244 * A virtual server may start off with a "dictionary" block, and therefore define a local
245 * dictionary. So the "root" dictionary of a virtual server may not be a protocol dict.
246 */
247 fr_assert(li->server_cs != NULL);
249
250 fr_assert(li->dict != NULL);
251
252 /*
253 * Skip a bunch of work if we're already in the network thread.
254 */
255 if (is_network_thread(nr) && !li->needs_full_setup) {
256 return fr_network_listen_add_self(nr, li);
257 }
258
260 if (!rb) return -1;
261
262 return fr_control_message_send(nr->control, rb, FR_CONTROL_ID_LISTEN, &li, sizeof(li));
263}
264
265
266/** Delete a socket from a network. MUST be called only by the listener itself!.
267 *
268 * @param nr the network
269 * @param li the listener
270 */
272{
274
276
277 s = fr_rb_find(nr->sockets, &(fr_network_socket_t){ .listen = li });
278 if (!s) return -1;
279
281
282 return 0;
283}
284
285/** Add a "watch directory" call to a network
286 *
287 * @param nr the network
288 * @param li the listener
289 */
291{
293
295 if (!rb) return -1;
296
297 return fr_control_message_send(nr->control, rb, FR_CONTROL_ID_DIRECTORY, &li, sizeof(li));
298}
299
300/** Add a worker to a network in a different thread
301 *
302 * @param nr the network
303 * @param worker the worker
304 */
306{
308
310 if (!rb) return -1;
311
312 (void) talloc_get_type_abort(nr, fr_network_t);
313 (void) talloc_get_type_abort(worker, fr_worker_t);
314
315 return fr_control_message_send(nr->control, rb, FR_CONTROL_ID_WORKER, &worker, sizeof(worker));
316}
317
318static void fr_network_worker_started_callback(void *ctx, void const *data, size_t data_size, fr_time_t now);
319
320/** Add a worker to a network in the same thread
321 *
322 * @param nr the network
323 * @param worker the worker
324 */
326{
327 fr_network_worker_started_callback(nr, &worker, sizeof(worker), fr_time_wrap(0));
328}
329
330
331/** Signal the network to read from a listener
332 *
333 * @param nr the network
334 * @param li the listener to read from
335 */
337{
339
340 (void) talloc_get_type_abort(nr, fr_network_t);
342
343 s = fr_rb_find(nr->sockets, &(fr_network_socket_t){ .listen = li });
344 if (!s) return;
345
346 /*
347 * Go read the socket.
348 */
349 fr_network_read(nr->el, s->listen->fd, 0, s);
350}
351
352
353/** Inject a packet for a listener to write
354 *
355 * @param nr the network
356 * @param li the listener where the packet is being injected
357 * @param packet the packet to be written
358 * @param packet_len the length of the packet
359 * @param packet_ctx The packet context to write
360 * @param request_time when the packet was received.
361 */
362void fr_network_listen_write(fr_network_t *nr, fr_listen_t *li, uint8_t const *packet, size_t packet_len,
363 void *packet_ctx, fr_time_t request_time)
364{
365 fr_message_t *lm;
367
368 cd = (fr_channel_data_t) {
369 .m = (fr_message_t) {
371 .data_size = packet_len,
372 .when = request_time,
373 },
374
375 .channel = {
376 .heap_id = FR_HEAP_INDEX_INVALID,
377 },
378
379 .listen = li,
380 .priority = PRIORITY_NOW,
381 .reply.request_time = request_time,
382 };
383
384 memcpy(&cd.m.data, &packet, sizeof(packet)); /* const issues */
385 memcpy(&cd.packet_ctx, &packet_ctx, sizeof(packet_ctx)); /* const issues */
386
387 /*
388 * Localize the message and insert it into the heap of pending messages.
389 */
390 lm = fr_message_localize(nr, &cd.m, sizeof(cd));
391 if (!lm) return;
392
393 if (fr_heap_insert(&nr->replies, lm) < 0) {
394 fr_message_done(lm);
395 }
396}
397
398
399/** Inject a packet for a listener to read
400 *
401 * @param nr the network
402 * @param li the listener where the packet is being injected
403 * @param packet the packet to be injected
404 * @param packet_len the length of the packet
405 * @param recv_time when the packet was received.
406 * @return
407 * - <0 on error
408 * - 0 on success
409 */
410int fr_network_listen_inject(fr_network_t *nr, fr_listen_t *li, uint8_t const *packet, size_t packet_len, fr_time_t recv_time)
411{
413 fr_network_inject_t my_inject;
414
415 /*
416 * Can't inject to injection-less destinations.
417 */
418 if (!li->app_io->inject) {
419 fr_strerror_const("Listener cannot accept injected packet");
420 return -1;
421 }
422
423 /*
424 * Avoid a bounce through the event loop if we're being called from the network thread.
425 */
426 if (is_network_thread(nr)) {
428
429 s = fr_rb_find(nr->sockets, &(fr_network_socket_t){ .listen = li });
430 if (!s) {
431 fr_strerror_const("Listener was not found for injected packet");
432 return -1;
433 }
434
435 /*
436 * Inject the packet. The master.c mod_read() routine will then take care of avoiding
437 * IO, and instead return the packet to the network side.
438 */
439 if (li->app_io->inject(li, packet, packet_len, recv_time) == 0) {
440 (void) fr_network_read(nr->el, li->fd, 0, s);
441 }
442
443 return 0;
444 }
445
447 if (!rb) return -1;
448
449 my_inject.listen = li;
450 my_inject.packet = talloc_memdup(NULL, packet, packet_len);
451 my_inject.packet_len = packet_len;
452 my_inject.recv_time = recv_time;
453
454 return fr_control_message_send(nr->control, rb, FR_CONTROL_ID_INJECT, &my_inject, sizeof(my_inject));
455}
456
458{
459 static fr_event_update_t pause_read[] = {
461 { 0 }
462 };
465
466 if (nr->suspended) return;
467
468 for (s = fr_rb_iter_init_inorder(&iter, nr->sockets);
469 s != NULL;
470 s = fr_rb_iter_next_inorder(&iter)) {
472 }
473 nr->suspended = true;
474}
475
477{
478 static fr_event_update_t resume_read[] = {
480 { 0 }
481 };
484
485 if (!nr->suspended) return;
486
487 for (s = fr_rb_iter_init_inorder(&iter, nr->sockets);
488 s != NULL;
489 s = fr_rb_iter_next_inorder(&iter)) {
491 }
492 nr->suspended = false;
493}
494
495#define IALPHA (8)
496#define RTT(_old, _new) fr_time_delta_wrap((fr_time_delta_unwrap(_new) + (fr_time_delta_unwrap(_old) * (IALPHA - 1))) / IALPHA)
497
498/** Callback which handles a message being received on the network side.
499 *
500 * @param[in] ctx the network
501 * @param[in] ch the channel that the message is on.
502 * @param[in] cd the message (if any) to start with
503 */
505{
506 fr_network_t *nr = ctx;
507 fr_network_worker_t *worker;
508
509 cd->channel.ch = ch;
510
511 /*
512 * Update stats for the worker.
513 */
515 worker->stats.out++;
516 worker->cpu_time = cd->reply.cpu_time;
517 if (!fr_time_delta_ispos(worker->predicted)) {
518 worker->predicted = cd->reply.processing_time;
519 } else {
520 worker->predicted = RTT(worker->predicted, cd->reply.processing_time);
521 }
522
523 /*
524 * Unblock the worker.
525 */
526 if (worker->blocked) {
527 worker->blocked = false;
528 nr->num_blocked--;
530 }
531
532 /*
533 * Ensure that heap insert works.
534 */
535 cd->channel.heap_id = FR_HEAP_INDEX_INVALID;
536 if (fr_heap_insert(&nr->replies, cd) < 0) {
537 fr_message_done(&cd->m);
538 fr_assert(0 == 1);
539 }
540}
541
542/** Handle a network control message callback for a channel
543 *
544 * This is called from the event loop when we get a notification
545 * from the event signalling pipe.
546 *
547 * @param[in] ctx the network
548 * @param[in] data the message
549 * @param[in] data_size size of the data
550 * @param[in] now the current time
551 */
552static void fr_network_channel_callback(void *ctx, void const *data, size_t data_size, fr_time_t now)
553{
555 fr_channel_t *ch;
556 fr_network_t *nr = ctx;
557
558 ce = fr_channel_service_message(now, &ch, data, data_size);
559 DEBUG3("Channel %s",
560 fr_table_str_by_value(channel_signals, ce, "<INVALID>"));
561 switch (ce) {
562 case FR_CHANNEL_ERROR:
563 return;
564
565 case FR_CHANNEL_EMPTY:
566 return;
567
568 case FR_CHANNEL_NOOP:
569 break;
570
572 fr_assert(ch != NULL);
573 while (fr_channel_recv_reply(ch));
574 break;
575
577 fr_assert(0 == 1);
578 break;
579
580 case FR_CHANNEL_OPEN:
581 fr_assert(0 == 1);
582 break;
583
584 case FR_CHANNEL_CLOSE:
585 {
586 fr_network_worker_t *w = talloc_get_type_abort(fr_channel_requestor_uctx_get(ch),
588 int i;
589
590 /*
591 * Remove this worker from the array
592 */
593 for (i = 0; i < nr->num_workers; i++) {
594 DEBUG3("Worker acked our close request");
595 if (nr->workers[i] == w) {
596 nr->workers[i] = NULL;
597
598 if (i == (nr->num_workers - 1)) break;
599
600 /*
601 * Close the hole...
602 */
603 memcpy(&nr->workers[i], &nr->workers[i + 1], ((nr->num_workers - i) - 1));
604 break;
605 }
606 }
607 nr->num_workers--;
608 }
609 break;
610 }
611}
612
613#define OUTSTANDING(_x) ((_x)->stats.in - (_x)->stats.out)
614
615/** Send a message on the "best" channel.
616 *
617 * @param nr the network
618 * @param cd the message we've received
619 */
621{
622 fr_network_worker_t *worker;
623
624 (void) talloc_get_type_abort(nr, fr_network_t);
625
626retry:
627 if (nr->num_workers == 1) {
628 worker = nr->workers[0];
629 if (worker->blocked) {
630 RATE_LIMIT_GLOBAL(ERROR, "Failed sending packet to worker - "
631 "In single-threaded mode and worker is blocked");
632 drop:
633 worker->stats.dropped++;
634 return -1;
635 }
636
637 } else if (nr->num_blocked == 0) {
638 int64_t cmp;
639 uint32_t one, two;
640
641 one = fr_rand() % nr->num_workers;
642 do {
643 two = fr_rand() % nr->num_workers;
644 } while (two == one);
645
646 /*
647 * Choose a worker based on minimizing the amount
648 * of future work it's being asked to do.
649 *
650 * If both workers have the same number of
651 * outstanding requests, then choose the worker
652 * which has used the least total CPU time.
653 */
654 cmp = (OUTSTANDING(nr->workers[one]) - OUTSTANDING(nr->workers[two]));
655 if (cmp < 0) {
656 worker = nr->workers[one];
657
658 } else if (cmp > 0) {
659 worker = nr->workers[two];
660
661 } else if (fr_time_delta_lt(nr->workers[one]->cpu_time, nr->workers[two]->cpu_time)) {
662 worker = nr->workers[one];
663
664 } else {
665 worker = nr->workers[two];
666 }
667 } else {
668 int i;
669 uint64_t min_outstanding = UINT64_MAX;
670 fr_network_worker_t *found = NULL;
671
672 /*
673 * Some workers are blocked. Pick the worker
674 * with the least amount of future work to do.
675 */
676 for (i = 0; i < nr->num_workers; i++) {
677 uint64_t outstanding;
678
679 worker = nr->workers[i];
680 if (worker->blocked) continue;
681
682 outstanding = OUTSTANDING(worker);
683 if ((outstanding < min_outstanding) || !found) {
684 found = worker;
685 min_outstanding = outstanding;
686
687 } else if (outstanding == min_outstanding) {
688 /*
689 * Queue lengths are the same.
690 * Choose this worker if it's
691 * less busy than the previous one we found.
692 */
693 if (fr_time_delta_lt(worker->cpu_time, found->cpu_time)) {
694 found = worker;
695 }
696 }
697 }
698
699 if (!found) {
700 RATE_LIMIT_GLOBAL(PERROR, "Failed sending packet to worker - Couldn't find active worker, "
701 "%u/%u workers are blocked", nr->num_blocked, nr->num_workers);
702 return -1;
703 }
704
705 worker = found;
706 }
707
708 (void) talloc_get_type_abort(worker, fr_network_worker_t);
709
710 /*
711 * Too many outstanding packets for this worker. Drop
712 * the request.
713 *
714 * If the worker we've picked has too many outstanding
715 * packets, then we have either only one worker, in which
716 * cae we should drop the packet. Or, we were unable to
717 * find a worker with smaller than max_outstanding
718 * packets. In which case all of the workers are likely
719 * at max_outstanding.
720 *
721 * In both cases, we should just drop the new packet.
722 */
723 fr_assert(worker->stats.in >= worker->stats.out);
724 if (nr->config.max_outstanding &&
725 (OUTSTANDING(worker) >= nr->config.max_outstanding)) {
726 RATE_LIMIT_GLOBAL(PERROR, "max_outstanding reached - dropping packet");
727 goto drop;
728 }
729
730 /*
731 * Send the message to the channel. If we fail, drop the
732 * packet. The only reason for failure is that the
733 * worker isn't servicing it's input queue. When that
734 * happens, we have no idea what to do, and the whole
735 * thing falls over.
736 */
737 if (fr_channel_send_request(worker->channel, cd) < 0) {
738 worker->stats.dropped++;
739 worker->blocked = true;
740 nr->num_blocked++;
741
742 RATE_LIMIT_GLOBAL(PERROR, "Failed sending packet to worker - %u/%u workers are blocked",
743 nr->num_blocked, nr->num_workers);
744
745 if (nr->num_blocked == nr->num_workers) {
747 return -1;
748 }
749 goto retry;
750 }
751
752 worker->stats.in++;
753
754 /*
755 * We're projecting that the worker will use more CPU
756 * time to process this request. The CPU time will be
757 * updated with a more accurate number when we receive a
758 * reply from this channel.
759 */
760 worker->cpu_time = fr_time_delta_add(worker->cpu_time, worker->predicted);
761
762 return 0;
763}
764
765
766/** Send a packet to the worker.
767 *
768 * MUST only be called from the network thread.
769 *
770 * @param nr the network
771 * @param parent the parent listener
772 * @param li the listener that the packet was "read" from. Can be "parent"
773 * @param buffer the packet to send
774 * @param buflen size of the packet to send
775 * @param recv_time of the packet
776 * @param packet_ctx for the packet
777 * @return
778 * - <0 on error
779 * - 0 on success
780 */
782 const uint8_t *buffer, size_t buflen, fr_time_t recv_time, void *packet_ctx)
783{
786
787 (void) talloc_get_type_abort(nr, fr_network_t);
789
790 s = fr_rb_find(nr->sockets, &(fr_network_socket_t){ .listen = li });
791 if (!s) return -1;
792
793 cd = (fr_channel_data_t *) fr_message_alloc(s->ms, NULL, buflen);
794 if (!cd) return -1;
795
796 cd->listen = parent;
798 cd->packet_ctx = packet_ctx;
799 cd->request.recv_time = recv_time;
800 memcpy(cd->m.data, buffer, buflen);
801 cd->m.when = fr_time();
802
803 if (fr_network_send_request(nr, cd) < 0) {
805 fr_message_done(&cd->m);
806 nr->stats.dropped++;
807 s->stats.dropped++;
808 return -1;
809 }
810
811 s->outstanding++;
812 return 0;
813}
814
815/** Get the number of outstanding packets
816 *
817 * @param nr the network
818 * @param li the listener that the packet was "read" from
819 * @return
820 * - <0 on error
821 * - the number of outstanding packets
822*/
825
826 (void) talloc_get_type_abort(nr, fr_network_t);
828
829 s = fr_rb_find(nr->sockets, &(fr_network_socket_t){ .listen = li });
830 if (!s) return -1;
831
832 return s->outstanding;
833}
834
835/*
836 * Mark it as dead, but DON'T free it until all of the replies
837 * have come in.
838 */
840{
841 int i;
842
843 if (s->dead) return;
844
845 s->dead = true;
846
847 fr_event_fd_delete(nr->el, s->listen->fd, s->filter);
848
849
850 for (i = 0; i < nr->max_workers; i++) {
851 if (!nr->workers[i]) continue;
852
853 (void) fr_worker_listen_cancel(nr->workers[i]->worker, s->listen);
854 }
855
856 /*
857 * If there are no outstanding packets, then we can free
858 * it now.
859 */
860 if (!s->outstanding) {
861 talloc_free(s);
862 return;
863 }
864
865 /*
866 * There are still outstanding packets. Leave it in the
867 * socket tree, so that replies from the worker can find
868 * it. When we've received all of the replies, then
869 * fr_network_post_event() will clean up this socket.
870 */
871}
872
873/** Read a packet from the network.
874 *
875 * @param[in] el the event list.
876 * @param[in] sockfd the socket which is ready to read.
877 * @param[in] flags from kevent.
878 * @param[in] ctx the network socket context.
879 */
880static void fr_network_read(UNUSED fr_event_list_t *el, int sockfd, UNUSED int flags, void *ctx)
881{
882 int num_messages = 0;
883 fr_network_socket_t *s = ctx;
884 fr_network_t *nr = s->nr;
885 ssize_t data_size;
886 fr_channel_data_t *cd, *next;
887
888 if (!fr_cond_assert_msg(s->listen->fd == sockfd, "Expected listen->fd (%u) to be equal event fd (%u)",
889 s->listen->fd, sockfd)) return;
890
891 DEBUG3("Reading data from FD %u", sockfd);
892
893 if (!s->cd) {
895 if (!cd) {
896 ERROR("Failed allocating message size %zd! - Closing socket",
899 return;
900 }
901 } else {
902 cd = s->cd;
903 }
904
905 fr_assert(cd->m.data != NULL);
906
907next_message:
908 /*
909 * Poll this socket, but not too often. We have to go
910 * service other sockets, too.
911 */
912 if (num_messages > 16) {
913 s->cd = cd;
914 return;
915 }
916
918
919 /*
920 * Read data from the network.
921 *
922 * Return of 0 means "no data", which is fine for UDP.
923 * For TCP, if an underlying read() on the TCP socket
924 * returns 0, (which signals that the FD is no longer
925 * usable) this function should return -1, so that the
926 * network side knows that it needs to close the
927 * connection.
928 */
929 data_size = s->listen->app_io->read(s->listen, &cd->packet_ctx, &cd->request.recv_time,
930 cd->m.data, cd->m.rb_size, &s->leftover);
931 if (data_size == 0) {
932 /*
933 * Cache the message for later. This is
934 * important for stream sockets, which can do
935 * partial reads into the current buffer. We
936 * need to be able to give the same buffer back
937 * to the stream socket for subsequent reads.
938 *
939 * Since we have a message set for each
940 * fr_io_socket_t, no "head of line"
941 * blocking issues can happen for stream sockets.
942 */
943 s->cd = cd;
944 return;
945 }
946
947 /*
948 * Error: close the connection, and remove the fr_listen_t
949 */
950 if (data_size < 0) {
951// fr_log(nr->log, L_DBG_ERR, "error from transport read on socket %d", sockfd);
953 return;
954 }
955 s->cd = NULL;
956
957 DEBUG3("Read %zd byte(s) from FD %u", data_size, sockfd);
958 nr->stats.in++;
959 s->stats.in++;
960
961 /*
962 * Initialize the rest of the fields of the channel data.
963 *
964 * We always use "now" as the time of the message, as the
965 * packet MAY be a duplicate packet magically resurrected
966 * from the past. i.e. If the read routines are doing
967 * dedup, then they notice that the packet is a
968 * duplicate. In that case, they send over a copy of the
969 * packet, BUT with the original timestamp. This
970 * information tells the worker that the packet is a
971 * duplicate.
972 */
973 cd->m.when = fr_time();
974 cd->listen = s->listen;
975
976 /*
977 * Nothing in the buffer yet. Allocate room for one
978 * packet.
979 */
980 if ((cd->m.data_size == 0) && (!s->leftover)) {
981
982 (void) fr_message_alloc(s->ms, &cd->m, data_size);
983 next = NULL;
984
985 } else {
986 /*
987 * There are leftover bytes in the buffer, feed
988 * them to the next round of reading.
989 */
990 next = (fr_channel_data_t *) fr_message_alloc_reserve(s->ms, &cd->m, data_size, s->leftover,
992 if (!next) {
993 PERROR("Failed reserving partial packet.");
994 // @todo - probably close the socket...
995 fr_assert(0 == 1);
996 }
997 }
998
999 /*
1000 * Set the priority. Which incidentally also checks if
1001 * we're allowed to read this particular kind of packet.
1002 *
1003 * That check is because the app_io handlers just read
1004 * packets, and don't really have access to the parent
1005 * "list of allowed packet types". So we have to do the
1006 * work here in a callback.
1007 *
1008 * That should probably be fixed...
1009 */
1010 if (s->listen->app->priority) {
1011 int priority;
1012
1013 priority = s->listen->app->priority(s->listen->app_instance, cd->m.data, data_size);
1014 if (priority <= 0) goto discard;
1015
1016 cd->priority = priority;
1017 }
1018
1019 if (fr_network_send_request(nr, cd) < 0) {
1020 discard:
1021 talloc_free(cd->packet_ctx); /* not sure what else to do here */
1022 fr_message_done(&cd->m);
1023 nr->stats.dropped++;
1024 s->stats.dropped++;
1025
1026 } else {
1027 /*
1028 * One more packet sent to a worker.
1029 */
1030 s->outstanding++;
1031 }
1032
1033 /*
1034 * If there is a next message, go read it from the buffer.
1035 *
1036 * @todo - note that this calls read(), even if the
1037 * app_io has paused the reader. We likely want to be
1038 * able to check that, too. We might just remove this
1039 * "goto"...
1040 */
1041 if (next) {
1042 cd = next;
1043 num_messages++;
1044 goto next_message;
1045 }
1046}
1047
1048int fr_network_sendto_worker(fr_network_t *nr, fr_listen_t *li, void *packet_ctx, uint8_t const *data, size_t data_len, fr_time_t recv_time)
1049{
1052
1053 s = fr_rb_find(nr->sockets, &(fr_network_socket_t){ .listen = li });
1054 if (!s) return -1;
1055
1056 cd = (fr_channel_data_t *) fr_message_alloc(s->ms, NULL, data_len);
1057 if (!cd) return -1;
1058
1059 s->stats.in++;
1060
1062
1063 cd->m.when = recv_time;
1064 cd->listen = li;
1065 cd->packet_ctx = packet_ctx;
1066
1067 memcpy(cd->m.data, data, data_len);
1068
1069 if (fr_network_send_request(nr, cd) < 0) {
1070 talloc_free(packet_ctx);
1071 fr_message_done(&cd->m);
1072 nr->stats.dropped++;
1073 s->stats.dropped++;
1074
1075 } else {
1076 /*
1077 * One more packet sent to a worker.
1078 */
1079 s->outstanding++;
1080 }
1081
1082 return 0;
1083}
1084
1085
1086/** Get a notification that a vnode changed
1087 *
1088 * @param[in] el the event list.
1089 * @param[in] sockfd the socket which is ready to read.
1090 * @param[in] fflags from kevent.
1091 * @param[in] ctx the network socket context.
1092 */
1093static void fr_network_vnode_extend(UNUSED fr_event_list_t *el, int sockfd, int fflags, void *ctx)
1094{
1095 fr_network_socket_t *s = ctx;
1096 fr_network_t *nr = s->nr;
1097
1099
1100 DEBUG3("network vnode");
1101
1102 /*
1103 * Tell the IO handler that something has happened to the
1104 * file.
1105 */
1106 s->listen->app_io->vnode(s->listen, fflags);
1107}
1108
1109
1110/** Handle errors for a socket.
1111 *
1112 * @param[in] el the event list
1113 * @param[in] sockfd the socket which has a fatal error.
1114 * @param[in] flags returned by kevent.
1115 * @param[in] fd_errno returned by kevent.
1116 * @param[in] ctx the network socket context.
1117 */
1119 int fd_errno, void *ctx)
1120{
1121 fr_network_socket_t *s = ctx;
1122 fr_network_t *nr = s->nr;
1123
1124 if (s->listen->app_io->error) {
1125 s->listen->app_io->error(s->listen);
1126
1127 } else if (flags & EV_EOF) {
1128 DEBUG2("Socket %s closed by peer", s->listen->name);
1129
1130 } else {
1131 ERROR("Socket %s errored - %s", s->listen->name, fr_syserror(fd_errno));
1132 }
1133
1135}
1136
1137
1140 { 0 }
1141};
1142
1145 { 0 }
1146};
1147
1148
1149/** Write packets to the network.
1150 *
1151 * @param el the event list
1152 * @param sockfd the socket which is ready to write
1153 * @param flags returned by kevent.
1154 * @param ctx the network socket context.
1155 */
1156static void fr_network_write(UNUSED fr_event_list_t *el, UNUSED int sockfd, UNUSED int flags, void *ctx)
1157{
1158 fr_network_socket_t *s = ctx;
1159 fr_listen_t *li = s->listen;
1160 fr_network_t *nr = s->nr;
1162
1163 (void) talloc_get_type_abort(nr, fr_network_t);
1164
1165 /*
1166 * Start with the currently pending message, and then
1167 * work through the priority heap.
1168 */
1169 if (s->pending) {
1170 cd = s->pending;
1171 s->pending = NULL;
1172
1173 } else {
1174 cd = fr_heap_pop(&s->waiting);
1175 }
1176
1177 while (cd != NULL) {
1178 int rcode;
1179
1180 fr_assert(li == cd->listen);
1181 rcode = li->app_io->write(li, cd->packet_ctx,
1182 cd->reply.request_time,
1183 cd->m.data, cd->m.data_size, s->written);
1184
1185 /*
1186 * As a special case, allow write() to return
1187 * "0", which means "close the socket".
1188 */
1189 if (rcode == 0) goto dead;
1190
1191 /*
1192 * Or we have a write error.
1193 */
1194 if (rcode < 0) {
1195 /*
1196 * Stop processing the heap, and set the
1197 * pending message to the current one.
1198 */
1199 if (errno == EWOULDBLOCK) {
1200 save_pending:
1201 fr_assert(!s->pending);
1202
1203 if (cd->m.status != FR_MESSAGE_LOCALIZED) {
1204 fr_message_t *lm;
1205
1206 lm = fr_message_localize(s, &cd->m, sizeof(*cd));
1207 if (!lm) {
1208 ERROR("Failed saving pending packet");
1209 goto dead;
1210 }
1211
1212 cd = (fr_channel_data_t *) lm;
1213 }
1214
1215 if (!s->blocked) {
1217 PERROR("Failed adding write callback to event loop");
1218 goto dead;
1219 }
1220
1221 s->blocked = true;
1222 }
1223
1224 s->pending = cd;
1225 return;
1226 }
1227
1228 /*
1229 * As a special hack, check for something
1230 * that will never be returned from a
1231 * real write() routine. Which then
1232 * signals to us that we have to close
1233 * the socket, but NOT complain about it.
1234 */
1235 if (errno == ECONNREFUSED) goto dead;
1236
1237 PERROR("Failed writing to socket %s", s->listen->name);
1238 if (li->app_io->error) li->app_io->error(li);
1239
1240 dead:
1241 fr_message_done(&cd->m);
1243 return;
1244 }
1245
1246 /*
1247 * If we've done a partial write, localize the message and continue.
1248 */
1249 if ((size_t) rcode < cd->m.data_size) {
1250 s->written = rcode;
1251 goto save_pending;
1252 }
1253
1254 s->written = 0;
1255
1256 /*
1257 * Reset for the next message.
1258 */
1259 fr_message_done(&cd->m);
1260 nr->stats.out++;
1261 s->stats.out++;
1262
1263 /*
1264 * Grab the net entry.
1265 */
1266 cd = fr_heap_pop(&s->waiting);
1267 }
1268
1269 /*
1270 * We've successfully written all of the packets. Remove
1271 * the write callback.
1272 */
1274 PERROR("Failed removing write callback from event loop");
1276 }
1277
1278 s->blocked = false;
1279}
1280
1282{
1283 fr_network_t *nr = s->nr;
1285
1286 fr_assert(s->outstanding == 0);
1287
1288 fr_rb_delete(nr->sockets, s);
1290
1291 fr_event_fd_delete(nr->el, s->listen->fd, s->filter);
1292
1293 if (s->listen->app_io->close) {
1294 s->listen->app_io->close(s->listen);
1295 } else {
1296 close(s->listen->fd);
1297 }
1298
1299 if (s->pending) {
1301 s->pending = NULL;
1302 }
1303
1304 /*
1305 * Clean up any queued entries.
1306 */
1307 while ((cd = fr_heap_pop(&s->waiting)) != NULL) {
1308 fr_message_done(&cd->m);
1309 }
1310
1311 talloc_free(s->waiting);
1312 talloc_free(s->listen);
1313
1314 return 0;
1315}
1316
1317
1318/** Handle a network control message callback for a new listener
1319 *
1320 * @param[in] ctx the network
1321 * @param[in] data the message
1322 * @param[in] data_size size of the data
1323 * @param[in] now the current time
1324 */
1325static void fr_network_listen_callback(void *ctx, void const *data, size_t data_size, UNUSED fr_time_t now)
1326{
1327 fr_network_t *nr = talloc_get_type_abort(ctx, fr_network_t);
1328 fr_listen_t *li;
1329
1330 fr_assert(data_size == sizeof(li));
1331
1332 if (data_size != sizeof(li)) return;
1333
1334 li = talloc_get_type_abort(*((void * const *)data), fr_listen_t);
1335
1336 (void) fr_network_listen_add_self(nr, li);
1337}
1338
1340{
1342 fr_app_io_t const *app_io;
1343 size_t size;
1344 int num_messages;
1345
1346 fr_assert(li->app_io != NULL);
1347
1348 /*
1349 * Non-socket listeners just get told about the event
1350 * list, and nothing else.
1351 */
1352 if (li->non_socket_listener) {
1353 fr_assert(li->app_io->event_list_set != NULL);
1354 fr_assert(!li->app_io->read);
1355 fr_assert(!li->app_io->write);
1356
1357 li->app_io->event_list_set(li, nr->el, nr);
1358
1359 /*
1360 * We use fr_log() here to avoid the "Network - " prefix.
1361 */
1362 fr_log(nr->log, L_DBG, __FILE__, __LINE__, "Listener %s bound to virtual server %s",
1363 li->name, cf_section_name2(li->server_cs));
1364
1365 return 0;
1366 }
1367
1368 s = talloc_zero(nr, fr_network_socket_t);
1369 fr_assert(s != NULL);
1370 talloc_steal(s, li);
1371
1372 s->nr = nr;
1373 s->listen = li;
1374 s->number = nr->num_sockets++;
1375
1376 MEM(s->waiting = fr_heap_alloc(s, waiting_cmp, fr_channel_data_t, channel.heap_id, 0));
1377
1378 talloc_set_destructor(s, _network_socket_free);
1379
1380 /*
1381 * Put reasonable limits on the ring buffer size. Then
1382 * round it up to the nearest power of 2, which is
1383 * required by the ring buffer code.
1384 */
1385 num_messages = s->listen->num_messages;
1386 if (num_messages < 8) num_messages = 8;
1387
1388 size = s->listen->default_message_size * num_messages;
1389 if (size < (1 << 17)) size = (1 << 17);
1390 if (size > (100 * 1024 * 1024)) size = (100 * 1024 * 1024);
1391
1392 /*
1393 * Allocate the ring buffer for messages and packets.
1394 */
1395 s->ms = fr_message_set_create(s, num_messages,
1396 sizeof(fr_channel_data_t),
1397 size);
1398 if (!s->ms) {
1399 PERROR("Failed creating message buffers for network IO");
1400 talloc_free(s);
1401 return -1;
1402 }
1403
1404 app_io = s->listen->app_io;
1406
1407 if (fr_event_fd_insert(nr, NULL, nr->el, s->listen->fd,
1411 s) < 0) {
1412 PERROR("Failed adding new socket to network event loop");
1413 talloc_free(s);
1414 return -1;
1415 }
1416
1417 /*
1418 * Start of with write updates being paused. We don't
1419 * care about being able to write if there's nothing to
1420 * write.
1421 */
1423
1424 /*
1425 * Add the listener before calling the app_io, so that
1426 * the app_io can find the listener which we're adding
1427 * here.
1428 */
1429 (void) fr_rb_insert(nr->sockets, s);
1430 (void) fr_rb_insert(nr->sockets_by_num, s);
1431
1432 if (app_io->event_list_set) app_io->event_list_set(s->listen, nr->el, nr);
1433
1434 /*
1435 * We use fr_log() here to avoid the "Network - " prefix.
1436 */
1437 fr_log(nr->log, L_DBG, __FILE__, __LINE__, "Listening on %s bound to virtual server %s",
1439
1440 DEBUG3("Using new socket %s with FD %d", s->listen->name, s->listen->fd);
1441
1442 return 0;
1443}
1444
1445/** Handle a network control message callback for a new "watch directory"
1446 *
1447 * @param[in] ctx the network
1448 * @param[in] data the message
1449 * @param[in] data_size size of the data
1450 * @param[in] now the current time
1451 */
1452static void fr_network_directory_callback(void *ctx, void const *data, size_t data_size, UNUSED fr_time_t now)
1453{
1454 int num_messages;
1455 fr_network_t *nr = talloc_get_type_abort(ctx, fr_network_t);
1456 fr_listen_t *li = talloc_get_type_abort(*((void * const *)data), fr_listen_t);
1458 fr_app_io_t const *app_io;
1460
1461 if (fr_cond_assert(data_size == sizeof(li))) return;
1462
1463 memcpy(&li, data, sizeof(li));
1464
1465 s = talloc_zero(nr, fr_network_socket_t);
1466 fr_assert(s != NULL);
1467 talloc_steal(s, li);
1468
1469 s->nr = nr;
1470 s->listen = li;
1471 s->number = nr->num_sockets++;
1472
1473 MEM(s->waiting = fr_heap_alloc(s, waiting_cmp, fr_channel_data_t, channel.heap_id, 0));
1474
1475 talloc_set_destructor(s, _network_socket_free);
1476
1477 /*
1478 * Allocate the ring buffer for messages and packets.
1479 */
1480 num_messages = s->listen->num_messages;
1481 if (num_messages < 8) num_messages = 8;
1482
1483 s->ms = fr_message_set_create(s, num_messages,
1484 sizeof(fr_channel_data_t),
1486 if (!s->ms) {
1487 PERROR("Failed creating message buffers for directory IO");
1488 talloc_free(s);
1489 return;
1490 }
1491
1492 app_io = s->listen->app_io;
1493
1494 if (app_io->event_list_set) app_io->event_list_set(s->listen, nr->el, nr);
1495
1497
1498 if (fr_event_filter_insert(nr, NULL, nr->el, s->listen->fd, s->filter,
1499 &funcs,
1500 app_io->error ? fr_network_error : NULL,
1501 s) < 0) {
1502 PERROR("Failed adding directory monitor event loop");
1503 talloc_free(s);
1504 return;
1505 }
1506
1507 (void) fr_rb_insert(nr->sockets, s);
1508 (void) fr_rb_insert(nr->sockets_by_num, s);
1509
1510 DEBUG3("Using new socket with FD %d", s->listen->fd);
1511}
1512
1513/** Handle a network control message callback for a new worker
1514 *
1515 * @param[in] ctx the network
1516 * @param[in] data the message
1517 * @param[in] data_size size of the data
1518 * @param[in] now the current time
1519 */
1520static void fr_network_worker_started_callback(void *ctx, void const *data, size_t data_size, UNUSED fr_time_t now)
1521{
1522 int i;
1523 fr_network_t *nr = ctx;
1524 fr_worker_t *worker;
1526
1527 fr_assert(data_size == sizeof(worker));
1528
1529 memcpy(&worker, data, data_size);
1530 (void) talloc_get_type_abort(worker, fr_worker_t);
1531
1532 MEM(w = talloc_zero(nr, fr_network_worker_t));
1533
1534 w->worker = worker;
1535 w->channel = fr_worker_channel_create(worker, w, nr->control);
1537 fr_fatal_assert_msg(w->channel, "Failed creating new channel");
1538
1541
1542 /*
1543 * FIXME: This creates a race in the network loop
1544 * exit condition, because it can theoretically
1545 * be signalled to exit before the workers have
1546 * ACKd channel creation.
1547 */
1548 nr->num_workers++;
1549
1550 /*
1551 * Insert the worker into the array of workers.
1552 */
1553 for (i = 0; i < nr->max_workers; i++) {
1554 if (nr->workers[i]) continue;
1555
1556 nr->workers[i] = w;
1557 return;
1558 }
1559
1560 /*
1561 * Run out of room to put workers!
1562 */
1563 fr_assert(0 == 1);
1564}
1565
1566/** Handle a network control message callback for a packet sent to a socket
1567 *
1568 * @param[in] ctx the network
1569 * @param[in] data the message
1570 * @param[in] data_size size of the data
1571 * @param[in] now the current time
1572 */
1573static void fr_network_inject_callback(void *ctx, void const *data, size_t data_size, UNUSED fr_time_t now)
1574{
1575 fr_network_t *nr = ctx;
1576 fr_network_inject_t my_inject;
1578
1579 fr_assert(data_size == sizeof(my_inject));
1580
1581 memcpy(&my_inject, data, data_size);
1582 s = fr_rb_find(nr->sockets, &(fr_network_socket_t){ .listen = my_inject.listen });
1583 if (!s) {
1584 talloc_free(my_inject.packet); /* MUST be it's own TALLOC_CTX */
1585 return;
1586 }
1587
1588 /*
1589 * Inject the packet, and then read it back from the
1590 * network.
1591 */
1592 if (s->listen->app_io->inject(s->listen, my_inject.packet, my_inject.packet_len, my_inject.recv_time) == 0) {
1593 fr_network_read(nr->el, s->listen->fd, 0, s);
1594 }
1595
1596 talloc_free(my_inject.packet);
1597}
1598
1599/** Run the event loop 'pre' callback
1600 *
1601 * This function MUST DO NO WORK. All it does is check if there's
1602 * work, and tell the event code to return to the main loop if
1603 * there's work to do.
1604 *
1605 * @param[in] now the current time.
1606 * @param[in] wake the time when the event loop will wake up.
1607 * @param[in] uctx the network
1608 */
1610{
1611 fr_network_t *nr = talloc_get_type_abort(uctx, fr_network_t);
1612
1613 if (fr_heap_num_elements(nr->replies) > 0) return 1;
1614
1615 return 0;
1616}
1617
1618/** Handle replies after all FD and timer events have been serviced
1619 *
1620 * @param el the event loop
1621 * @param now the current time (mostly)
1622 * @param uctx the fr_network_t
1623 */
1625{
1627 fr_network_t *nr = talloc_get_type_abort(uctx, fr_network_t);
1628
1629 /*
1630 * Pull the replies off of our global heap, and try to
1631 * push them to the individual sockets.
1632 */
1633 while ((cd = fr_heap_pop(&nr->replies)) != NULL) {
1634 fr_listen_t *li;
1636
1637 li = cd->listen;
1638
1639 /*
1640 * @todo - cache this somewhere so we don't need
1641 * to do an rbtree lookup for every packet.
1642 */
1643 s = fr_rb_find(nr->sockets, &(fr_network_socket_t){ .listen = li });
1644
1645 /*
1646 * This shouldn't happen, but be safe...
1647 */
1648 if (!s) {
1649 fr_message_done(&cd->m);
1650 continue;
1651 }
1652
1653 if (cd->m.status != FR_MESSAGE_LOCALIZED) {
1654 fr_assert(s->outstanding > 0);
1655 s->outstanding--;
1656 }
1657
1658 /*
1659 * Just mark the message done, and skip it.
1660 */
1661 if (s->dead) {
1662 fr_message_done(&cd->m);
1663
1664 /*
1665 * No more packets, it's safe to delete
1666 * the socket.
1667 */
1668 if (!s->outstanding) talloc_free(s);
1669
1670 continue;
1671 }
1672
1673 /*
1674 * No data to write to the socket, so we skip the message.
1675 */
1676 if (!cd->m.data_size) {
1677 fr_message_done(&cd->m);
1678 continue;
1679 }
1680
1681 /*
1682 * No pending message, let's try writing it.
1683 *
1684 * If there is a pending message, then we're
1685 * waiting for IO write to become ready.
1686 */
1687 if (!s->pending) {
1688 fr_assert(!s->blocked);
1689 (void) fr_heap_insert(&s->waiting, cd);
1690 fr_network_write(nr->el, s->listen->fd, 0, s);
1691 }
1692 }
1693}
1694
1695/** Stop a network thread in an orderly way
1696 *
1697 * @param[in] nr the network to stop
1698 */
1700{
1702
1703 (void) talloc_get_type_abort(nr, fr_network_t);
1704
1705 /*
1706 * Close the network sockets
1707 */
1708 {
1709 fr_network_socket_t **sockets;
1710 size_t len;
1711 size_t i;
1712
1713 if (fr_rb_flatten_inorder(nr, (void ***)&sockets, nr->sockets) < 0) return -1;
1714 len = talloc_array_length(sockets);
1715
1716 for (i = 0; i < len; i++) {
1717 /*
1718 * Force to zero so we don't trigger asserts
1719 * if packets are being processed and the
1720 * server exits.
1721 */
1722 sockets[i]->outstanding = 0;
1723 talloc_free(sockets[i]);
1724 }
1725
1726 talloc_free(sockets);
1727 }
1728
1729
1730 /*
1731 * Clean up all outstanding replies.
1732 *
1733 * We can't do this after signalling the
1734 * workers to close, because they free
1735 * their message sets, and we end up
1736 * getting random use-after-free errors
1737 * as there's a race between the network
1738 * popping replies, and the workers
1739 * freeing their message sets.
1740 *
1741 * This isn't perfect, and we might still
1742 * lose some replies, but it's good enough
1743 * for now.
1744 *
1745 * @todo - call transport "done" for the reply, so that
1746 * it knows the replies are done, too.
1747 */
1748 while ((cd = fr_heap_pop(&nr->replies)) != NULL) {
1749 fr_message_done(&cd->m);
1750 }
1751
1752 /*
1753 * Signal the workers that we're closing
1754 *
1755 * nr->num_workers is decremented every
1756 * time a worker closes a socket.
1757 *
1758 * When nr->num_workers == 0, the event
1759 * loop (fr_network()) will exit.
1760 */
1761 {
1762 int i;
1763
1764 for (i = 0; i < nr->num_workers; i++) {
1765 fr_network_worker_t *worker = nr->workers[i];
1766
1768 }
1769 }
1770
1773 nr->exiting = true;
1775
1776 return 0;
1777}
1778
1779/** Read handler for signal pipe
1780 *
1781 */
1782static void _signal_pipe_read(UNUSED fr_event_list_t *el, int fd, UNUSED int flags, void *uctx)
1783{
1784 fr_network_t *nr = talloc_get_type_abort(uctx, fr_network_t);
1785 uint8_t buff;
1786
1787 if (read(fd, &buff, sizeof(buff)) < 0) {
1788 ERROR("Failed reading signal - %s", fr_syserror(errno));
1789 return;
1790 }
1791
1792 fr_assert(buff == 1);
1793
1794 /*
1795 * fr_network_stop() will signal the workers
1796 * to exit (by closing their channels).
1797 *
1798 * When we get the ack, we decrement our
1799 * nr->num_workers counter.
1800 *
1801 * When the counter reaches 0, the event loop
1802 * exits.
1803 */
1804 DEBUG2("Signalled to exit");
1805
1806 if (unlikely(fr_network_destroy(nr) < 0)) {
1807 PERROR("Failed stopping network");
1808 }
1809}
1810
1811/** The main network worker function.
1812 *
1813 * @param[in] nr the network data structure to run.
1814 */
1816{
1817 /*
1818 * Run until we're told to exit AND the number of
1819 * workers has dropped to zero.
1820 *
1821 * This is important as if we exit too early we
1822 * free the channels out from underneath the
1823 * workers and they read uninitialised memory.
1824 *
1825 * Whenever a worker ACKs our close notification
1826 * nr->num_workers is decremented, so when
1827 * nr->num_workers == 0, all workers have ACKd
1828 * our close and are no longer using the channel.
1829 */
1830 while (likely(!(nr->exiting && (nr->num_workers == 0)))) {
1831 bool wait_for_event;
1832 int num_events;
1833
1834 /*
1835 * There are runnable requests. We still service
1836 * the event loop, but we don't wait for events.
1837 */
1838 wait_for_event = (fr_heap_num_elements(nr->replies) == 0);
1839
1840 /*
1841 * Check the event list. If there's an error
1842 * (e.g. exit), we stop looping and clean up.
1843 */
1844 DEBUG4("Gathering events - %s", wait_for_event ? "will wait" : "Will not wait");
1845 num_events = fr_event_corral(nr->el, fr_time(), wait_for_event);
1846 DEBUG4("%u event(s) pending%s",
1847 num_events == -1 ? 0 : num_events, num_events == -1 ? " - event loop exiting" : "");
1848 if (num_events < 0) break;
1849
1850 /*
1851 * Service outstanding events.
1852 */
1853 if (num_events > 0) {
1854 DEBUG4("Servicing event(s)");
1855 fr_event_service(nr->el);
1856 }
1857 }
1858 return;
1859}
1860
1861/** Signal a network thread to exit
1862 *
1863 * @note Request to exit will be processed asynchronously.
1864 *
1865 * @param[in] nr the network data structure to manage
1866 * @return
1867 * - 0 on success.
1868 * - -1 on failure.
1869 */
1871{
1872 if (write(nr->signal_pipe[1], &(uint8_t){ 0x01 }, 1) < 0) {
1873 fr_strerror_printf("Failed signalling network thread to exit - %s", fr_syserror(errno));
1874 return -1;
1875 }
1876
1877 return 0;
1878}
1879
1880/** Free any resources associated with a network thread
1881 *
1882 */
1884{
1885 if (nr->signal_pipe[0] >= 0) close(nr->signal_pipe[0]);
1886 if (nr->signal_pipe[1] >= 0) close(nr->signal_pipe[1]);
1887
1888 return 0;
1889}
1890
1891/** Create a network
1892 *
1893 * @param[in] ctx The talloc ctx
1894 * @param[in] el The event list
1895 * @param[in] name Networker identifier.
1896 * @param[in] logger The destination for all logging messages
1897 * @param[in] lvl Log level
1898 * @param[in] config configuration structure.
1899 * @return
1900 * - NULL on error
1901 * - fr_network_t on success
1902 */
1903fr_network_t *fr_network_create(TALLOC_CTX *ctx, fr_event_list_t *el, char const *name,
1904 fr_log_t const *logger, fr_log_lvl_t lvl,
1906{
1907 fr_network_t *nr;
1908
1909 nr = talloc_zero(ctx, fr_network_t);
1910 if (!nr) {
1911 fr_strerror_const("Failed allocating memory");
1912 return NULL;
1913 }
1914 talloc_set_destructor(nr, _fr_network_free);
1915
1916 nr->name = talloc_strdup(nr, name);
1917
1918 nr->thread_id = pthread_self();
1919 nr->el = el;
1920 nr->log = logger;
1921 nr->lvl = lvl;
1922
1924 nr->num_workers = 0;
1925 nr->signal_pipe[0] = -1;
1926 nr->signal_pipe[1] = -1;
1927 if (config) nr->config = *config;
1928
1929 nr->aq_control = fr_atomic_queue_alloc(nr, 1024);
1930 if (!nr->aq_control) {
1931 talloc_free(nr);
1932 return NULL;
1933 }
1934
1935 nr->control = fr_control_create(nr, el, nr->aq_control);
1936 if (!nr->control) {
1937 fr_strerror_const_push("Failed creating control queue");
1938 fail:
1939 talloc_free(nr);
1940 return NULL;
1941 }
1942
1943 /*
1944 * @todo - rely on thread-local variables. And then the
1945 * various users of this can check if (rb == nr->rb), and
1946 * if so, skip the whole control plane / kevent /
1947 * whatever roundabout thing.
1948 */
1950 if (!nr->rb) {
1951 fr_strerror_const_push("Failed creating ring buffer");
1952 fail2:
1953 talloc_free(nr->control);
1954 goto fail;
1955 }
1956
1958 fr_strerror_const_push("Failed adding channel callback");
1959 goto fail2;
1960 }
1961
1963 fr_strerror_const_push("Failed adding socket callback");
1964 goto fail2;
1965 }
1966
1968 fr_strerror_const_push("Failed adding socket callback");
1969 goto fail2;
1970 }
1971
1973 fr_strerror_const_push("Failed adding worker callback");
1974 goto fail2;
1975 }
1976
1978 fr_strerror_const_push("Failed adding packet injection callback");
1979 goto fail2;
1980 }
1981
1982 /*
1983 * Create the various heaps.
1984 */
1986 if (!nr->sockets) {
1987 fr_strerror_const_push("Failed creating listen tree for sockets");
1988 goto fail2;
1989 }
1990
1992 if (!nr->sockets_by_num) {
1993 fr_strerror_const_push("Failed creating number tree for sockets");
1994 goto fail2;
1995 }
1996
1997 nr->replies = fr_heap_alloc(nr, reply_cmp, fr_channel_data_t, channel.heap_id, 0);
1998 if (!nr->replies) {
1999 fr_strerror_const_push("Failed creating heap for replies");
2000 goto fail2;
2001 }
2002
2003 if (fr_event_pre_insert(nr->el, fr_network_pre_event, nr) < 0) {
2004 fr_strerror_const("Failed adding pre-check to event list");
2005 goto fail2;
2006 }
2007
2008 if (fr_event_post_insert(nr->el, fr_network_post_event, nr) < 0) {
2009 fr_strerror_const("Failed inserting post-processing event");
2010 goto fail2;
2011 }
2012
2013 if (pipe(nr->signal_pipe) < 0) {
2014 fr_strerror_printf("Failed initialising signal pipe - %s", fr_syserror(errno));
2015 goto fail2;
2016 }
2017 if (fr_nonblock(nr->signal_pipe[0]) < 0) goto fail2;
2018 if (fr_nonblock(nr->signal_pipe[1]) < 0) goto fail2;
2019
2020 if (fr_event_fd_insert(nr, NULL, nr->el, nr->signal_pipe[0], _signal_pipe_read, NULL, NULL, nr) < 0) {
2021 fr_strerror_const("Failed inserting event for signal pipe");
2022 goto fail2;
2023 }
2024
2025 return nr;
2026}
2027
2028int fr_network_stats(fr_network_t const *nr, int num, uint64_t *stats)
2029{
2030 if (num < 0) return -1;
2031 if (num == 0) return 0;
2032
2033 stats[0] = nr->stats.in;
2034 if (num >= 2) stats[1] = nr->stats.out;
2035 if (num >= 3) stats[2] = nr->stats.dup;
2036 if (num >= 4) stats[3] = nr->stats.dropped;
2037 if (num >= 5) stats[4] = nr->num_workers;
2038
2039 if (num <= 5) return num;
2040
2041 return 5;
2042}
2043
2044void fr_network_stats_log(fr_network_t const *nr, fr_log_t const *log)
2045{
2046 int i;
2047
2048 /*
2049 * Dump all of the channel statistics.
2050 */
2051 for (i = 0; i < nr->max_workers; i++) {
2052 if (!nr->workers[i]) continue;
2053
2054 fr_channel_stats_log(nr->workers[i]->channel, log, __FILE__, __LINE__);
2055 }
2056}
2057
2058static int cmd_stats_self(FILE *fp, UNUSED FILE *fp_err, void *ctx, UNUSED fr_cmd_info_t const *info)
2059{
2060 fr_network_t const *nr = ctx;
2061
2062 fprintf(fp, "count.in\t%" PRIu64 "\n", nr->stats.in);
2063 fprintf(fp, "count.out\t%" PRIu64 "\n", nr->stats.out);
2064 fprintf(fp, "count.dup\t%" PRIu64 "\n", nr->stats.dup);
2065 fprintf(fp, "count.dropped\t%" PRIu64 "\n", nr->stats.dropped);
2066 fprintf(fp, "count.sockets\t%u\n", fr_rb_num_elements(nr->sockets));
2067
2068 return 0;
2069}
2070
2071static int cmd_socket_list(FILE *fp, UNUSED FILE *fp_err, void *ctx, UNUSED fr_cmd_info_t const *info)
2072{
2073 fr_network_t const *nr = ctx;
2076
2077 // @todo - note that this isn't thread-safe!
2078
2079 for (s = fr_rb_iter_init_inorder(&iter, nr->sockets);
2080 s != NULL;
2081 s = fr_rb_iter_next_inorder(&iter)) {
2082 if (!s->listen->app_io->get_name) {
2083 fprintf(fp, "%s\n", s->listen->app_io->common.name);
2084 } else {
2085 fprintf(fp, "%d\t%s\n", s->number, s->listen->app_io->get_name(s->listen));
2086 }
2087 }
2088 return 0;
2089}
2090
2091static int cmd_stats_socket(FILE *fp, FILE *fp_err, void *ctx, fr_cmd_info_t const *info)
2092{
2093 fr_network_t const *nr = ctx;
2095
2096 s = fr_rb_find(nr->sockets_by_num, &(fr_network_socket_t){ .number = info->box[0]->vb_uint32 });
2097 if (!s) {
2098 fprintf(fp_err, "No such socket number '%s'.\n", info->argv[0]);
2099 return -1;
2100 }
2101
2102 fprintf(fp, "count.in\t%" PRIu64 "\n", s->stats.in);
2103 fprintf(fp, "count.out\t%" PRIu64 "\n", s->stats.out);
2104 fprintf(fp, "count.dup\t%" PRIu64 "\n", s->stats.dup);
2105 fprintf(fp, "count.dropped\t%" PRIu64 "\n", s->stats.dropped);
2106
2107 return 0;
2108}
2109
2110
2112 {
2113 .parent = "stats",
2114 .name = "network",
2115 .help = "Statistics for network threads.",
2116 .read_only = true
2117 },
2118
2119 {
2120 .parent = "stats network",
2121 .add_name = true,
2122 .name = "self",
2123 .func = cmd_stats_self,
2124 .help = "Show statistics for a specific network thread.",
2125 .read_only = true
2126 },
2127
2128 {
2129 .parent = "stats network",
2130 .add_name = true,
2131 .name = "socket",
2132 .syntax = "INTEGER",
2133 .func = cmd_stats_socket,
2134 .help = "Show statistics for a specific socket",
2135 .read_only = true
2136 },
2137
2138 {
2139 .parent = "show",
2140 .name = "network",
2141 .help = "Show information about network threads.",
2142 .read_only = true
2143 },
2144
2145 {
2146 .parent = "show network",
2147 .add_name = true,
2148 .name = "socket",
2149 .syntax = "list",
2150 .func = cmd_socket_list,
2151 .help = "List the sockets associated with this network thread.",
2152 .read_only = true
2153 },
2154
2156};
static int const char char buffer[256]
Definition acutest.h:576
fr_io_close_t close
Close the transport.
Definition app_io.h:60
fr_io_data_read_t read
Read from a socket to a data buffer.
Definition app_io.h:47
module_t common
Common fields to all loadable modules.
Definition app_io.h:34
fr_io_signal_t error
There was an error on the socket.
Definition app_io.h:59
fr_app_event_list_set_t event_list_set
Called by the network thread to pass an event list for use by the app_io_t.
Definition app_io.h:36
fr_io_data_inject_t inject
Inject a packet into a socket.
Definition app_io.h:50
fr_io_data_vnode_t vnode
Handle notifications that the VNODE has changed.
Definition app_io.h:52
fr_io_data_write_t write
Write from a data buffer to a socket.
Definition app_io.h:48
fr_io_name_t get_name
get the socket name
Definition app_io.h:70
Public structure describing an I/O path for a protocol.
Definition app_io.h:33
fr_app_priority_get_t priority
Assign a priority to the packet.
Definition application.h:90
#define fr_atexit_thread_local(_name, _free, _uctx)
Definition atexit.h:221
fr_atomic_queue_t * fr_atomic_queue_alloc(TALLOC_CTX *ctx, size_t size)
Create fixed-size atomic queue.
Structure to hold the atomic queue.
#define RCSID(id)
Definition build.h:485
#define CMP(_a, _b)
Same as CMP_PREFER_SMALLER use when you don't really care about ordering, you just want an ordering.
Definition build.h:112
#define unlikely(_x)
Definition build.h:383
#define UNUSED
Definition build.h:317
char const * cf_section_name2(CONF_SECTION const *cs)
Return the second identifier of a CONF_SECTION.
Definition cf_util.c:1184
void * fr_channel_requestor_uctx_get(fr_channel_t *ch)
Get network-specific data from a channel.
Definition channel.c:922
fr_table_num_sorted_t const channel_signals[]
Definition channel.c:153
bool fr_channel_recv_reply(fr_channel_t *ch)
Receive a reply message from the channel.
Definition channel.c:408
int fr_channel_signal_responder_close(fr_channel_t *ch)
Signal a responder that the channel is closing.
Definition channel.c:824
int fr_channel_send_request(fr_channel_t *ch, fr_channel_data_t *cd)
Send a request message into the channel.
Definition channel.c:306
int fr_channel_set_recv_reply(fr_channel_t *ch, void *uctx, fr_channel_recv_callback_t recv_reply)
Definition channel.c:930
fr_channel_event_t fr_channel_service_message(fr_time_t when, fr_channel_t **p_channel, void const *data, size_t data_size)
Service a control-plane message.
Definition channel.c:685
void fr_channel_requestor_uctx_add(fr_channel_t *ch, void *uctx)
Add network-specific data to a channel.
Definition channel.c:910
void fr_channel_stats_log(fr_channel_t const *ch, fr_log_t const *log, char const *file, int line)
Definition channel.c:963
A full channel, which consists of two ends.
Definition channel.c:144
fr_message_t m
the message header
Definition channel.h:105
fr_channel_event_t
Definition channel.h:67
@ FR_CHANNEL_NOOP
Definition channel.h:74
@ FR_CHANNEL_EMPTY
Definition channel.h:75
@ FR_CHANNEL_CLOSE
Definition channel.h:72
@ FR_CHANNEL_ERROR
Definition channel.h:68
@ FR_CHANNEL_DATA_READY_REQUESTOR
Definition channel.h:70
@ FR_CHANNEL_OPEN
Definition channel.h:71
@ FR_CHANNEL_DATA_READY_RESPONDER
Definition channel.h:69
void * packet_ctx
Packet specific context for holding client information, and other proto_* specific information that n...
Definition channel.h:142
fr_listen_t * listen
for tracking packet transport, etc.
Definition channel.h:146
#define PRIORITY_NORMAL
Definition channel.h:151
#define PRIORITY_NOW
Definition channel.h:149
uint32_t priority
Priority of this packet.
Definition channel.h:140
Channel information which is added to a message.
Definition channel.h:104
char const * parent
e.g. "show module"
Definition command.h:52
#define CMD_TABLE_END
Definition command.h:62
char const ** argv
text version of commands
Definition command.h:42
#define FR_CONTROL_ID_INJECT
Definition control.h:60
#define FR_CONTROL_ID_DIRECTORY
Definition control.h:59
#define FR_CONTROL_ID_CHANNEL
Definition control.h:56
#define FR_CONTROL_ID_LISTEN
Definition control.h:57
#define FR_CONTROL_ID_WORKER
Definition control.h:58
#define FR_CONTROL_MAX_SIZE
Definition control.h:51
#define FR_CONTROL_MAX_MESSAGES
Definition control.h:50
static fr_ring_buffer_t * rb
#define fr_cond_assert(_x)
Calls panic_action ifndef NDEBUG, else logs error and evaluates to value of _x.
Definition debug.h:139
#define fr_cond_assert_msg(_x, _fmt,...)
Calls panic_action ifndef NDEBUG, else logs error and evaluates to value of _x.
Definition debug.h:156
#define fr_fatal_assert_msg(_x, _fmt,...)
Calls panic_action ifndef NDEBUG, else logs error and causes the server to exit immediately with code...
Definition debug.h:184
#define MEM(x)
Definition debug.h:36
#define ERROR(fmt,...)
Definition dhcpclient.c:41
static int sockfd
Definition dhcpclient.c:56
fr_dict_t const * fr_dict_proto_dict(fr_dict_t const *dict)
Definition dict_util.c:4974
#define fr_event_fd_insert(...)
Definition event.h:248
fr_event_filter_t
The type of filter to install for an FD.
Definition event.h:83
@ FR_EVENT_FILTER_VNODE
Filter for vnode subfilters.
Definition event.h:85
@ FR_EVENT_FILTER_IO
Combined filter for read/write functions/.
Definition event.h:84
#define fr_event_filter_update(...)
Definition event.h:240
#define fr_event_filter_insert(...)
Definition event.h:235
#define FR_EVENT_RESUME(_s, _f)
Re-add the filter for a func from kevent.
Definition event.h:132
#define FR_EVENT_SUSPEND(_s, _f)
Temporarily remove the filter for a func from kevent.
Definition event.h:116
fr_event_fd_cb_t extend
Additional files were added to a directory.
Definition event.h:199
Callbacks for the FR_EVENT_FILTER_IO filter.
Definition event.h:189
Structure describing a modification to a filter's state.
Definition event.h:97
Callbacks for the FR_EVENT_FILTER_VNODE filter.
Definition event.h:196
int fr_heap_insert(fr_heap_t **hp, void *data)
Insert a new element into the heap.
Definition heap.c:146
void * fr_heap_pop(fr_heap_t **hp)
Remove a node from the heap.
Definition heap.c:322
unsigned int fr_heap_index_t
Definition heap.h:80
#define fr_heap_alloc(_ctx, _cmp, _type, _field, _init)
Creates a heap that can be used with non-talloced elements.
Definition heap.h:100
static unsigned int fr_heap_num_elements(fr_heap_t *h)
Return the number of elements in the heap.
Definition heap.h:179
#define FR_HEAP_INDEX_INVALID
Definition heap.h:83
The main heap structure.
Definition heap.h:66
uint64_t out
Definition base.h:43
uint64_t dup
Definition base.h:44
uint64_t dropped
Definition base.h:45
uint64_t in
Definition base.h:42
int fr_control_callback_add(fr_control_t *c, uint32_t id, void *ctx, fr_control_callback_t callback)
Register a callback for an ID.
Definition control.c:417
int fr_control_message_send(fr_control_t *c, fr_ring_buffer_t *rb, uint32_t id, void *data, size_t data_size)
Send a control-plane message.
Definition control.c:343
fr_control_t * fr_control_create(TALLOC_CTX *ctx, fr_event_list_t *el, fr_atomic_queue_t *aq)
Create a control-plane signaling path.
Definition control.c:149
The control structure.
Definition control.c:79
size_t num_messages
for the message ring buffer
Definition listen.h:53
bool non_socket_listener
special internal listener that does not use sockets.
Definition listen.h:46
char const * name
printable name for this socket - set by open
Definition listen.h:29
void const * app_instance
Definition listen.h:39
size_t default_message_size
copied from app_io, but may be changed
Definition listen.h:52
fr_app_t const * app
Definition listen.h:38
CONF_SECTION * server_cs
CONF_SECTION of the server.
Definition listen.h:41
bool no_write_callback
sometimes we don't need to do writes
Definition listen.h:45
int fd
file descriptor for this socket - set by open
Definition listen.h:28
fr_dict_t const * dict
dictionary for this listener
Definition listen.h:30
bool needs_full_setup
Set to true to avoid the short cut when adding the listener.
Definition listen.h:47
fr_app_io_t const * app_io
I/O path functions.
Definition listen.h:32
fr_ring_buffer_t * rb
ring buffer for my control-plane messages
Definition network.c:126
fr_cmd_table_t cmd_network_table[]
Definition network.c:2111
size_t fr_network_listen_outstanding(fr_network_t *nr, fr_listen_t *li)
Get the number of outstanding packets.
Definition network.c:823
size_t written
however much we did in a partial write
Definition network.c:92
int fr_network_listen_send_packet(fr_network_t *nr, fr_listen_t *parent, fr_listen_t *li, const uint8_t *buffer, size_t buflen, fr_time_t recv_time, void *packet_ctx)
Send a packet to the worker.
Definition network.c:781
fr_atomic_queue_t * aq_control
atomic queue for control messages sent to me
Definition network.c:122
static int cmd_stats_socket(FILE *fp, FILE *fp_err, void *ctx, fr_cmd_info_t const *info)
Definition network.c:2091
int fr_network_listen_add(fr_network_t *nr, fr_listen_t *li)
Add a fr_listen_t to a network.
Definition network.c:236
bool suspended
whether or not we're suspended.
Definition network.c:117
int fr_network_worker_add(fr_network_t *nr, fr_worker_t *worker)
Add a worker to a network in a different thread.
Definition network.c:305
int fr_network_destroy(fr_network_t *nr)
Stop a network thread in an orderly way.
Definition network.c:1699
fr_network_t * nr
O(N) issues in talloc.
Definition network.c:77
fr_io_stats_t stats
Definition network.c:70
fr_listen_t * listen
Definition network.c:52
static void fr_network_listen_callback(void *ctx, void const *data, size_t data_size, UNUSED fr_time_t now)
Handle a network control message callback for a new listener.
Definition network.c:1325
uint8_t * packet
Definition network.c:53
static int cmd_stats_self(FILE *fp, UNUSED FILE *fp_err, void *ctx, UNUSED fr_cmd_info_t const *info)
Definition network.c:2058
fr_log_t const * log
log destination
Definition network.c:119
int fr_network_directory_add(fr_network_t *nr, fr_listen_t *li)
Add a "watch directory" call to a network.
Definition network.c:290
static int _fr_network_free(fr_network_t *nr)
Free any resources associated with a network thread.
Definition network.c:1883
static void fr_network_inject_callback(void *ctx, void const *data, size_t data_size, UNUSED fr_time_t now)
Handle a network control message callback for a packet sent to a socket.
Definition network.c:1573
fr_heap_index_t heap_id
for the sockets_by_num heap
Definition network.c:79
#define RTT(_old, _new)
Definition network.c:496
void fr_network(fr_network_t *nr)
The main network worker function.
Definition network.c:1815
fr_message_set_t * ms
message buffers for this socket.
Definition network.c:89
int fr_network_listen_delete(fr_network_t *nr, fr_listen_t *li)
Delete a socket from a network.
Definition network.c:271
int num_blocked
number of blocked workers
Definition network.c:138
static void fr_network_worker_started_callback(void *ctx, void const *data, size_t data_size, fr_time_t now)
char const * name
Network ID for logging.
Definition network.c:113
void fr_network_worker_add_self(fr_network_t *nr, fr_worker_t *worker)
Add a worker to a network in the same thread.
Definition network.c:325
unsigned int outstanding
number of outstanding packets sent to the worker
Definition network.c:86
static _Thread_local fr_ring_buffer_t * fr_network_rb
Definition network.c:49
int number
unique ID
Definition network.c:78
static fr_event_update_t const resume_write[]
Definition network.c:1143
fr_time_delta_t predicted
predicted processing time for one packet
Definition network.c:64
static int fr_network_pre_event(fr_time_t now, fr_time_delta_t wake, void *uctx)
fr_worker_t * worker
worker pointer
Definition network.c:69
int fr_network_sendto_worker(fr_network_t *nr, fr_listen_t *li, void *packet_ctx, uint8_t const *data, size_t data_len, fr_time_t recv_time)
Definition network.c:1048
int fr_network_exit(fr_network_t *nr)
Signal a network thread to exit.
Definition network.c:1870
#define MAX_WORKERS
Definition network.c:47
int fr_network_listen_inject(fr_network_t *nr, fr_listen_t *li, uint8_t const *packet, size_t packet_len, fr_time_t recv_time)
Inject a packet for a listener to read.
Definition network.c:410
fr_listen_t * listen
I/O ctx and functions.
Definition network.c:87
int num_sockets
actually a counter...
Definition network.c:141
fr_rb_node_t listen_node
rbtree node for looking up by listener.
Definition network.c:74
static void fr_network_vnode_extend(UNUSED fr_event_list_t *el, int sockfd, int fflags, void *ctx)
Get a notification that a vnode changed.
Definition network.c:1093
static void _signal_pipe_read(UNUSED fr_event_list_t *el, int fd, UNUSED int flags, void *uctx)
Read handler for signal pipe.
Definition network.c:1782
#define OUTSTANDING(_x)
Definition network.c:613
static int8_t reply_cmp(void const *one, void const *two)
Definition network.c:157
int num_workers
number of active workers
Definition network.c:137
static int8_t socket_num_cmp(void const *one, void const *two)
Definition network.c:186
int num_pending_workers
number of workers we're waiting to start.
Definition network.c:139
fr_rb_tree_t * sockets
list of sockets we're managing, ordered by the listener
Definition network.c:134
pthread_t thread_id
for self
Definition network.c:115
fr_log_lvl_t lvl
debug log level
Definition network.c:120
int signal_pipe[2]
Pipe for signalling the worker in an orderly way.
Definition network.c:143
fr_channel_data_t * pending
the currently pending partial packet
Definition network.c:94
static int8_t socket_listen_cmp(void const *one, void const *two)
Definition network.c:179
static void fr_network_write(UNUSED fr_event_list_t *el, UNUSED int sockfd, UNUSED int flags, void *ctx)
Write packets to the network.
Definition network.c:1156
fr_event_list_t * el
our event list
Definition network.c:128
fr_heap_t * replies
replies from the worker, ordered by priority / origin time
Definition network.c:130
static int fr_network_send_request(fr_network_t *nr, fr_channel_data_t *cd)
Send a message on the "best" channel.
Definition network.c:620
void fr_network_stats_log(fr_network_t const *nr, fr_log_t const *log)
Definition network.c:2044
fr_heap_t * waiting
packets waiting to be written
Definition network.c:95
int fr_network_stats(fr_network_t const *nr, int num, uint64_t *stats)
Definition network.c:2028
fr_heap_index_t heap_id
workers are in a heap
Definition network.c:62
bool blocked
is this worker blocked?
Definition network.c:66
static void fr_network_read(UNUSED fr_event_list_t *el, int sockfd, UNUSED int flags, void *ctx)
Read a packet from the network.
Definition network.c:880
static bool is_network_thread(fr_network_t const *nr)
Definition network.c:224
fr_rb_node_t num_node
rbtree node for looking up by number.
Definition network.c:75
static void fr_network_error(UNUSED fr_event_list_t *el, UNUSED int sockfd, int flags, int fd_errno, void *ctx)
Handle errors for a socket.
Definition network.c:1118
fr_io_stats_t stats
Definition network.c:96
static fr_ring_buffer_t * fr_network_rb_init(void)
Initialise thread local storage.
Definition network.c:206
fr_channel_data_t * cd
cached in case of allocation & read error
Definition network.c:90
static int fr_network_listen_add_self(fr_network_t *nr, fr_listen_t *listen)
Definition network.c:1339
static void fr_network_suspend(fr_network_t *nr)
Definition network.c:457
bool dead
is it dead?
Definition network.c:83
size_t leftover
leftover data from a previous read
Definition network.c:91
static void fr_network_post_event(fr_event_list_t *el, fr_time_t now, void *uctx)
fr_network_worker_t * workers[MAX_WORKERS]
each worker
Definition network.c:149
fr_time_t recv_time
Definition network.c:55
static void fr_network_unsuspend(fr_network_t *nr)
Definition network.c:476
fr_rb_tree_t * sockets_by_num
ordered by number;
Definition network.c:135
fr_network_config_t config
configuration
Definition network.c:148
void fr_network_listen_read(fr_network_t *nr, fr_listen_t *li)
Signal the network to read from a listener.
Definition network.c:336
static int8_t waiting_cmp(void const *one, void const *two)
Definition network.c:168
fr_io_stats_t stats
Definition network.c:132
static int _fr_network_rb_free(void *arg)
Definition network.c:197
static void fr_network_recv_reply(void *ctx, fr_channel_t *ch, fr_channel_data_t *cd)
Callback which handles a message being received on the network side.
Definition network.c:504
int max_workers
maximum number of allowed workers
Definition network.c:140
void fr_network_listen_write(fr_network_t *nr, fr_listen_t *li, uint8_t const *packet, size_t packet_len, void *packet_ctx, fr_time_t request_time)
Inject a packet for a listener to write.
Definition network.c:362
bool exiting
are we exiting?
Definition network.c:146
fr_event_filter_t filter
what type of filter it is
Definition network.c:81
static void fr_network_socket_dead(fr_network_t *nr, fr_network_socket_t *s)
Definition network.c:839
static void fr_network_directory_callback(void *ctx, void const *data, size_t data_size, UNUSED fr_time_t now)
Handle a network control message callback for a new "watch directory".
Definition network.c:1452
fr_channel_t * channel
channel to the worker
Definition network.c:68
static fr_event_update_t const pause_write[]
Definition network.c:1138
fr_network_t * fr_network_create(TALLOC_CTX *ctx, fr_event_list_t *el, char const *name, fr_log_t const *logger, fr_log_lvl_t lvl, fr_network_config_t const *config)
Create a network.
Definition network.c:1903
static int _network_socket_free(fr_network_socket_t *s)
Definition network.c:1281
static void fr_network_channel_callback(void *ctx, void const *data, size_t data_size, fr_time_t now)
Handle a network control message callback for a channel.
Definition network.c:552
static int cmd_socket_list(FILE *fp, UNUSED FILE *fp_err, void *ctx, UNUSED fr_cmd_info_t const *info)
Definition network.c:2071
fr_time_delta_t cpu_time
how much CPU time this worker has spent
Definition network.c:63
fr_control_t * control
the control plane
Definition network.c:124
bool blocked
is it blocked?
Definition network.c:84
Associate a worker thread with a network thread.
Definition network.c:61
uint32_t max_outstanding
Definition network.h:46
#define PERROR(_fmt,...)
Definition log.h:228
#define DEBUG3(_fmt,...)
Definition log.h:266
#define DEBUG4(_fmt,...)
Definition log.h:267
#define RATE_LIMIT_GLOBAL(_log, _fmt,...)
Rate limit messages using a global limiting entry.
Definition log.h:641
void fr_event_service(fr_event_list_t *el)
Service any outstanding timer or file descriptor events.
Definition event.c:2194
int fr_event_pre_delete(fr_event_list_t *el, fr_event_status_cb_t callback, void *uctx)
Delete a pre-event callback from the event list.
Definition event.c:1976
int fr_event_corral(fr_event_list_t *el, fr_time_t now, bool wait)
Gather outstanding timer and file descriptor events.
Definition event.c:2059
talloc_free(reap)
int fr_event_post_delete(fr_event_list_t *el, fr_event_post_cb_t callback, void *uctx)
Delete a post-event callback from the event list.
Definition event.c:2030
int fr_event_pre_insert(fr_event_list_t *el, fr_event_status_cb_t callback, void *uctx)
Add a pre-event callback to the event list.
Definition event.c:1954
int fr_event_fd_delete(fr_event_list_t *el, int fd, fr_event_filter_t filter)
Remove a file descriptor from the event loop.
Definition event.c:1203
int fr_event_post_insert(fr_event_list_t *el, fr_event_post_cb_t callback, void *uctx)
Add a post-event callback to the event list.
Definition event.c:2008
Stores all information relating to an event list.
Definition event.c:377
void fr_log(fr_log_t const *log, fr_log_type_t type, char const *file, int line, char const *fmt,...)
Send a server log message to its destination.
Definition log.c:580
fr_log_lvl_t
Definition log.h:67
@ L_DBG
Only displayed when debugging is enabled.
Definition log.h:59
static fr_event_update_t pause_read[]
Definition master.c:170
static fr_event_update_t resume_read[]
Definition master.c:175
unsigned int uint32_t
long int ssize_t
unsigned char uint8_t
int fr_message_done(fr_message_t *m)
Mark a message as done.
Definition message.c:190
fr_message_t * fr_message_alloc(fr_message_set_t *ms, fr_message_t *m, size_t actual_packet_size)
Allocate packet data for a message.
Definition message.c:988
fr_message_t * fr_message_localize(TALLOC_CTX *ctx, fr_message_t *m, size_t message_size)
Localize a message by copying it to local storage.
Definition message.c:242
fr_message_t * fr_message_reserve(fr_message_set_t *ms, size_t reserve_size)
Reserve a message.
Definition message.c:934
fr_message_t * fr_message_alloc_reserve(fr_message_set_t *ms, fr_message_t *m, size_t actual_packet_size, size_t leftover, size_t reserve_size)
Allocate packet data for a message, and reserve a new message.
Definition message.c:1077
fr_message_set_t * fr_message_set_create(TALLOC_CTX *ctx, int num_messages, size_t message_size, size_t ring_buffer_size)
Create a message set.
Definition message.c:127
A Message set, composed of message headers and ring buffer data.
Definition message.c:95
size_t rb_size
cache-aligned size in the ring buffer
Definition message.h:51
fr_time_t when
when this message was sent
Definition message.h:47
uint8_t * data
pointer to the data in the ring buffer
Definition message.h:49
size_t data_size
size of the data in the ring buffer
Definition message.h:50
@ FR_MESSAGE_USED
Definition message.h:39
@ FR_MESSAGE_LOCALIZED
Definition message.h:40
fr_message_status_t status
free, used, done, etc.
Definition message.h:45
int fr_nonblock(UNUSED int fd)
Definition misc.c:293
static const conf_parser_t config[]
Definition base.c:183
#define fr_assert(_expr)
Definition rad_assert.h:38
#define DEBUG2(fmt,...)
Definition radclient.h:43
static fr_app_io_t app_io
uint32_t fr_rand(void)
Return a 32-bit random number.
Definition rand.c:105
uint32_t fr_rb_num_elements(fr_rb_tree_t *tree)
Return how many nodes there are in a tree.
Definition rb.c:781
void * fr_rb_iter_init_inorder(fr_rb_iter_inorder_t *iter, fr_rb_tree_t *tree)
Initialise an in-order iterator.
Definition rb.c:824
void * fr_rb_iter_next_inorder(fr_rb_iter_inorder_t *iter)
Return the next node.
Definition rb.c:850
void * fr_rb_find(fr_rb_tree_t const *tree, void const *data)
Find an element in the tree, returning the data, not the node.
Definition rb.c:577
bool fr_rb_insert(fr_rb_tree_t *tree, void const *data)
Insert data into a tree.
Definition rb.c:626
bool fr_rb_delete(fr_rb_tree_t *tree, void const *data)
Remove node and free data (if a free function was specified)
Definition rb.c:741
#define fr_rb_inline_talloc_alloc(_ctx, _type, _field, _data_cmp, _data_free)
Allocs a red black that verifies elements are of a specific talloc type.
Definition rb.h:246
int fr_rb_flatten_inorder(TALLOC_CTX *ctx, void **out[], fr_rb_tree_t *tree)
Iterator structure for in-order traversal of an rbtree.
Definition rb.h:321
The main red black tree structure.
Definition rb.h:73
fr_ring_buffer_t * fr_ring_buffer_create(TALLOC_CTX *ctx, size_t size)
Create a ring buffer.
Definition ring_buffer.c:64
static char const * name
static char buff[sizeof("18446744073709551615")+3]
Definition size_tests.c:41
#define fr_time()
Allow us to arbitrarily manipulate time.
Definition state_test.c:8
Definition log.h:96
char const * fr_syserror(int num)
Guaranteed to be thread-safe version of strerror.
Definition syserror.c:243
#define fr_table_str_by_value(_table, _number, _def)
Convert an integer to a string.
Definition table.h:772
#define talloc_get_type_abort_const
Definition talloc.h:282
static fr_time_delta_t fr_time_delta_from_msec(int64_t msec)
Definition time.h:575
static fr_time_delta_t fr_time_delta_add(fr_time_delta_t a, fr_time_delta_t b)
Definition time.h:255
#define fr_time_delta_lt(_a, _b)
Definition time.h:285
#define fr_time_wrap(_time)
Definition time.h:145
#define fr_time_delta_ispos(_a)
Definition time.h:290
static int8_t fr_time_cmp(fr_time_t a, fr_time_t b)
Compare two fr_time_t values.
Definition time.h:916
A time delta, a difference in time measured in nanoseconds.
Definition time.h:80
"server local" time.
Definition time.h:69
close(uq->fd)
static fr_event_list_t * el
static fr_slen_t parent
Definition pair.h:845
void fr_perror(char const *fmt,...)
Print the current error to stderr with a prefix.
Definition strerror.c:732
#define fr_strerror_printf(_fmt,...)
Log to thread local error buffer.
Definition strerror.h:64
#define fr_strerror_const_push(_msg)
Definition strerror.h:227
#define fr_strerror_const(_msg)
Definition strerror.h:223
static fr_slen_t data
Definition value.h:1274
fr_dict_t const * virtual_server_dict_by_cs(CONF_SECTION const *server_cs)
Return the namespace for the virtual server specified by a config section.
fr_channel_t * fr_worker_channel_create(fr_worker_t *worker, TALLOC_CTX *ctx, fr_control_t *master)
Create a channel to the worker.
Definition worker.c:1654
int fr_worker_listen_cancel(fr_worker_t *worker, fr_listen_t const *li)
Definition worker.c:1681
A worker which takes packets from a master, and processes them.
Definition worker.c:99