allow ack for channel close

diff --git a/src/util/channel.c b/src/util/channel.c
index 9aaea1b8ace1a7518993aa922df14317fc64d536..ae2ef065852ba5e1b589ef0ab1ba5d068ad97d41 100644 (file)
--- 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 37e5f090cf9009cfd14945eaae986836612dfdec..35b660281dabd98fc61e093f1a76ab74f9eabadf 100644 (file)
--- 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 b52f3fb6bbb90def3bfc80b058adb0061def6060..4de4931e3461baa5e337040c483abdb65bacb6aa 100644 (file)
--- 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 a6ee56851e4733994cbc50e5268478c20d2de025..5f7cd18b87926cad8a3aab67a87bdc16d3f37ec3 100644 (file)
--- a/src/util/schedule.c
+++ b/src/util/schedule.c
@@ -30,6 +30,8 @@ RCSID("$Id$")
 #include <freeradius-devel/util/schedule.h>
 #include <freeradius-devel/rbtree.h>
 
+#include <freeradius-devel/util/receiver.h>
+
 #ifdef HAVE_PTHREAD_H
 #include <pthread.h>
 #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 67d8003c129983b54704a57d8456b9a418cc8ed1..32c13f8c9574c4b330a44f18b2fdc71cc4b553ab 100644 (file)
--- 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);
        }
 }