From: Alan T. DeKok Date: Fri, 2 Dec 2016 21:10:51 +0000 (-0500) Subject: allow ack for channel close X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=d30ffe68ee470b0d07af9c1dc940ba3fb84e6902;p=thirdparty%2Ffreeradius-server.git allow ack for channel close --- diff --git a/src/util/channel.c b/src/util/channel.c index 9aaea1b8ace..ae2ef065852 100644 --- a/src/util/channel.c +++ b/src/util/channel.c @@ -87,6 +87,7 @@ fr_channel_t *fr_channel_create(TALLOC_CTX *ctx, int kq_master, int kq_worker) { fr_time_t when; fr_channel_t *ch; + struct kevent events[10]; ch = talloc_zero(ctx, fr_channel_t); if (!ch) return NULL; @@ -121,6 +122,28 @@ fr_channel_t *fr_channel_create(TALLOC_CTX *ctx, int kq_master, int kq_worker) ch->active = true; + /* + * Enable each individual EVFILT_USER event. + */ + EV_SET(&events[0], FR_CHANNEL_SIGNAL_OPEN, EVFILT_USER, EV_ADD | EV_ONESHOT, NOTE_FFCOPY, 0, NULL); + EV_SET(&events[1], FR_CHANNEL_SIGNAL_CLOSE, EVFILT_USER, EV_ADD | EV_ONESHOT, NOTE_FFCOPY, 0, NULL); + EV_SET(&events[2], FR_CHANNEL_SIGNAL_DATA_TO_WORKER, EVFILT_USER, EV_ADD | EV_ONESHOT, NOTE_FFCOPY, 0, NULL); + if (kevent(kq_worker, events, 3, NULL, 0, NULL) < 0) { + talloc_free(ch); + return NULL; + } + + /* + * And the same for the master KQ. + */ + EV_SET(&events[0], FR_CHANNEL_SIGNAL_WORKER_SLEEPING, EVFILT_USER, EV_ADD | EV_ONESHOT, NOTE_FFCOPY, 0, NULL); + EV_SET(&events[1], FR_CHANNEL_SIGNAL_CLOSE, EVFILT_USER, EV_ADD | EV_ONESHOT, NOTE_FFCOPY, 0, NULL); + EV_SET(&events[2], FR_CHANNEL_SIGNAL_DATA_FROM_WORKER, EVFILT_USER, EV_ADD | EV_ONESHOT, NOTE_FFCOPY, 0, NULL); + if (kevent(kq_master, events, 3, NULL, 0, NULL) < 0) { + talloc_free(ch); + return NULL; + } + return ch; } @@ -156,7 +179,7 @@ static int fr_channel_data_ready(fr_channel_t *ch, fr_time_t when, fr_channel_en * that a thread listening on multiple channels can * receive events unique to each one. */ - EV_SET(&kev, which, EVFILT_USER, EV_ADD, 0, 0, ch); + EV_SET(&kev, which, EVFILT_USER, EV_ENABLE, NOTE_TRIGGER, 0, ch); return kevent(end->kq, &kev, 1, NULL, 0, NULL); } @@ -423,7 +446,7 @@ int fr_channel_worker_sleeping(fr_channel_t *ch) * that a thread listening on multiple channels can * receive events unique to each one. */ - EV_SET(&kev, FR_CHANNEL_SIGNAL_WORKER_SLEEPING, EVFILT_USER, EV_ADD, 0, end->ack, ch); + EV_SET(&kev, FR_CHANNEL_SIGNAL_WORKER_SLEEPING, EVFILT_USER, EV_ENABLE, NOTE_TRIGGER, end->ack, ch); return kevent(end->kq, &kev, 1, NULL, 0, NULL); } @@ -492,11 +515,10 @@ fr_channel_event_t fr_channel_service_kevent(int kq, struct kevent const *kev, f } /* - * Only the master can signal that a channel should be - * closed. + * Each end can signal the channel to close. */ if (kev->ident == FR_CHANNEL_SIGNAL_CLOSE) { - rad_assert(kq == ch->end[TO_WORKER].kq); + rad_assert(kq == ch->end[kev->fflags].kq); *p_channel = ch; return FR_CHANNEL_CLOSE; @@ -562,9 +584,32 @@ bool fr_channel_active(fr_channel_t *ch) * - <0 on error * - 0 on success */ -int fr_channel_signal_close(fr_channel_t *ch) +int fr_channel_signal_close(fr_channel_t *ch, bool ack) { + struct kevent kev; + ch->active = false; - return 0; + EV_SET(&kev, FR_CHANNEL_SIGNAL_CLOSE, EVFILT_USER, EV_ENABLE, NOTE_TRIGGER | NOTE_FFCOPY, ack, ch); + + return kevent(ch->end[ack].kq, &kev, 1, NULL, 0, NULL); +} + +/** Send a channel to a KQ + * + * @param[in] kq the kqueue to send the channel + * @param[in] ch the channel + * @return + * - <0 on error + * - 0 on success + */ +int fr_channel_signal_open(int kq, fr_channel_t *ch) +{ + struct kevent kev; + + rad_assert(kq == ch->end[TO_WORKER].kq); + + EV_SET(&kev, FR_CHANNEL_SIGNAL_OPEN, EVFILT_USER, EV_ENABLE, NOTE_TRIGGER, 0, ch); + + return kevent(kq, &kev, 1, NULL, 0, NULL); } diff --git a/src/util/channel.h b/src/util/channel.h index 37e5f090cf9..35b660281da 100644 --- a/src/util/channel.h +++ b/src/util/channel.h @@ -114,7 +114,9 @@ int fr_channel_worker_sleeping(fr_channel_t *ch) CC_HINT(nonnull); fr_channel_event_t fr_channel_service_kevent(int kq, struct kevent const *kev, fr_time_t when, fr_channel_t **p_channel) CC_HINT(nonnull); bool fr_channel_active(fr_channel_t *ch) CC_HINT(nonnull); -int fr_channel_signal_close(fr_channel_t *ch) CC_HINT(nonnull); + +int fr_channel_signal_open(int kq, fr_channel_t *ch) CC_HINT(nonnull); +int fr_channel_signal_close(fr_channel_t *ch, bool ack) CC_HINT(nonnull); #ifdef __cplusplus } diff --git a/src/util/receiver.c b/src/util/receiver.c index b52f3fb6bbb..4de4931e346 100644 --- a/src/util/receiver.c +++ b/src/util/receiver.c @@ -230,7 +230,7 @@ int fr_receiver_destroy(fr_receiver_t *rc) * closing/ */ while ((worker = fr_heap_pop(rc->workers)) != NULL) { - fr_channel_signal_close(worker->channel); + fr_channel_signal_close(worker->channel, false); (void) fr_heap_insert(rc->closing, worker); } @@ -251,8 +251,15 @@ int fr_receiver_destroy(fr_receiver_t *rc) return 0; } +/** The main network worker function. + * + * @param[in] rc th receiver data structure to run. + */ void fr_receiver(fr_receiver_t *rc) { + /* + * The receiver is entirely event driven. + */ while (fr_event_loop(rc->el) == 0) { /* nothing */ } diff --git a/src/util/schedule.c b/src/util/schedule.c index a6ee56851e4..5f7cd18b879 100644 --- a/src/util/schedule.c +++ b/src/util/schedule.c @@ -30,6 +30,8 @@ RCSID("$Id$") #include #include +#include + #ifdef HAVE_PTHREAD_H #include #define PTHREAD_MUTEX_LOCK pthread_mutex_lock @@ -93,6 +95,19 @@ typedef struct fr_schedule_worker_t { fr_worker_t *worker; //!< the worker data structure } fr_schedule_worker_t; +/** + * A data structure to track network threads / receivers. + */ +typedef struct fr_schedule_receiver_t { + pthread_t pthread_id; //!< the thread of this receiver + + int kq; //!< the receivers KQ + fr_event_list_t *el; //!< the receivers event list + + fr_receiver_t *rc; //!< the receive data structure +} fr_schedule_receiver_t; + + /** * The scheduler */ @@ -422,6 +437,15 @@ int fr_schedule_destroy(fr_schedule_t *sc) } +#if 0 +int fr_schedule_socket_add(fr_schedule_t *sc, int fd, fr_transport_t *transport, void *ctx) +{ + // send it to a receivers KQ as transport / ctx + // it receives it via the USERFILT, and adds the transport / ctx + // transport_ctx is largely rad_listen_t, which is a transport-specific socket +} +#endif + /* * @todo single threaded mode. Instead of having function * specific to single threaded mode, just fix the event loop. diff --git a/src/util/worker.c b/src/util/worker.c index 67d8003c129..32c13f8c957 100644 --- a/src/util/worker.c +++ b/src/util/worker.c @@ -573,7 +573,7 @@ void fr_worker_destroy(fr_worker_t *worker) * automatically freed when our talloc context is freed. */ for (i = 0; i < worker->num_channels; i++) { - fr_channel_signal_close(worker->channel[i]); + fr_channel_signal_close(worker->channel[i], true); } }