added receiver code

Basic infrastructure for now.  More later.

diff --git a/src/util/all.mk b/src/util/all.mk
index a12f1c357525d58669f5fdd534b0e43f68e3c6fb..203d5c065444364c346eb9762224610692e726a3 100644 (file)
--- a/src/util/all.mk
+++ b/src/util/all.mk
@@ -1,6 +1,7 @@
 TARGET := libfreeradius-util.a
 
-SOURCES        :=      ring_buffer.c message.c atomic_queue.c queue.c time.c channel.c track.c worker.c schedule.c
+SOURCES        :=      ring_buffer.c message.c atomic_queue.c queue.c time.c channel.c track.c worker.c \
+               schedule.c receiver.c
 
 TGT_PREREQS    := libfreeradius-radius.la
 TGT_LDLIBS     := $(LIBS)

diff --git a/src/util/receiver.c b/src/util/receiver.c
new file mode 100644 (file)
index 0000000..b52f3fb
--- /dev/null
+++ b/src/util/receiver.c
@@ -0,0 +1,259 @@
+/*
+ *   This program is free software; you can redistribute it and/or modify
+ *   it under the terms of the GNU General Public License as published by
+ *   the Free Software Foundation; either version 2 of the License, or
+ *   (at your option) any later version.
+ *
+ *   This program is distributed in the hope that it will be useful,
+ *   but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *   GNU General Public License for more details.
+ *
+ *   You should have received a copy of the GNU General Public License
+ *   along with this program; if not, write to the Free Software
+ *   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
+ */
+
+/**
+ * $Id$
+ *
+ * @brief Receiver of socket data, which sends messages to the workers.
+ * @file util/receiver.c
+ *
+ * @copyright 2016 Alan DeKok <aland@freeradius.org>
+ */
+RCSID("$Id$")
+
+#include <talloc.h>
+
+#include <freeradius-devel/event.h>
+#include <freeradius-devel/util/queue.h>
+#include <freeradius-devel/util/worker.h>
+#include <freeradius-devel/util/receiver.h>
+
+#include <freeradius-devel/rad_assert.h>
+
+typedef struct fr_receiver_worker_t {
+       int                     heap_id;
+       fr_time_t               cpu_time;
+       fr_time_t               processing_time;
+
+       fr_channel_t            *channel;
+       fr_worker_t             *worker;
+} fr_receiver_worker_t;
+
+struct fr_receiver_t {
+       int                     kq;
+       fr_event_list_t         *el;
+
+       fr_heap_t               *replies;
+       fr_heap_t               *workers;
+       fr_heap_t               *closing;
+
+       uint32_t                num_transports; //!< how many transport layers we have
+       fr_transport_t          **transports;   //!< array of active transports.
+};
+
+static int worker_cmp(void const *one, void const *two)
+{
+       fr_receiver_worker_t const *a = one;
+       fr_receiver_worker_t const *b = two;
+
+       if (a->cpu_time < b->cpu_time) return -1;
+       if (a->cpu_time > b->cpu_time) return +1;
+
+       return 0;
+}
+
+static int reply_cmp(void const *one, void const *two)
+{
+       fr_channel_data_t const *a = one;
+       fr_channel_data_t const *b = two;
+
+       if (a->priority < b->priority) return -1;
+       if (a->priority > b->priority) return +1;
+
+       if (a->m.when < b->m.when) return -1;
+       if (a->m.when > b->m.when) return +1;
+
+       return 0;
+}
+
+#if 0
+/** Send a message on the "best" channel.
+ *
+ */
+int fr_receiver_send_request(fr_receiver_t *rc, fr_channel_data_t *cd)
+{
+       fr_receiver_worker_t *worker;
+       fr_channel_data_t *reply;
+
+#ifndef NDEBUG
+       (void) talloc_get_type_abort(rc, fr_receiver_t);
+#endif
+
+       worker = fr_heap_pop(rc->workers);
+       if (!worker) return 0;
+
+       /*
+        *      Send the message to the channel.  If we fail, recurse.
+        *      That's easier than manually tracking the channel we
+        *      popped off of the heap.
+        *
+        *      The only practical reason why the channel send will
+        *      fail is because the recipient is not servicing it's
+        *      queue.  When that happens, just hand the request to
+        *      another channel.
+        *
+        *      If we run out of channels to use, the caller needs to
+        *      allocate another one, and hand it to the scheduler.
+        */
+       if (fr_channel_send_request(worker->channel, cd, &reply) < 0) {
+               int rcode;
+
+               rcode = fr_receiver_send_request(rc, cd);
+
+               /*
+                *      Mark this channel as still busy, for some
+                *      future time.  This process ensures that we
+                *      don't immediately pop it off the heap and try
+                *      to send it another request.
+                */
+               worker->cpu_time = cd->m.when + worker->processing_time;
+               (void) fr_heap_insert(rc->workers, worker);
+
+               return rcode;
+       }
+
+       /*
+        *      We're projecting that the worker will use more CPU
+        *      time to process this request.  The CPU time will be
+        *      updated with a more accurate number when we receive a
+        *      reply from this channel.
+        */
+       worker->cpu_time += worker->processing_time;
+
+       /*
+        *      Insert the worker back into the scheduler.
+        */
+       (void) fr_heap_insert(rc->workers, worker);
+
+       /*
+        *      If we have a reply, push it onto our local queue, and
+        *      poll for more replies.
+        */
+       if (reply) {
+               do {
+                       reply->channel.ch = worker->channel;
+                       (void) fr_heap_insert(rc->replies, reply);
+               } while ((reply = fr_channel_recv_reply(worker->channel)) != NULL);
+       }
+
+       return 0;
+}
+#endif
+
+/** Create a receiver
+ *
+ * @param[in] ctx the talloc ctx
+ * @param[in] num_transports the number of transports in the transport array
+ * @param[in] transports the array of transports.
+ * @return
+ *     - NULL on error
+ *     - fr_receiver_t on success
+ */
+fr_receiver_t *fr_receiver_create(TALLOC_CTX *ctx, uint32_t num_transports, fr_transport_t **transports)
+{
+       fr_receiver_t *rc;
+
+       if (!num_transports || !transports) return NULL;
+
+       rc = talloc_zero(ctx, fr_receiver_t);
+       if (!rc) return NULL;
+
+       rc->el = fr_event_list_create(rc, NULL, NULL);
+       if (!rc->el) {
+               talloc_free(rc);
+               return NULL;
+       }
+
+       rc->kq = fr_event_list_kq(rc->el);
+       rad_assert(rc->kq >= 0);
+
+       rc->replies = fr_heap_create(reply_cmp, offsetof(fr_channel_data_t, channel.heap_id));
+       if (!rc->replies) {
+               talloc_free(rc);
+               return NULL;
+       }
+
+       rc->workers = fr_heap_create(worker_cmp, offsetof(fr_channel_data_t, channel.heap_id));
+       if (!rc->workers) {
+               talloc_free(rc);
+               return NULL;
+       }
+
+       rc->closing = fr_heap_create(worker_cmp, offsetof(fr_channel_data_t, channel.heap_id));
+       if (!rc->closing) {
+               talloc_free(rc);
+               return NULL;
+       }
+
+       // insert our kevent handler
+       // start off with a channel?
+       // i.e. get new sockets from that channel?
+
+       rc->num_transports = num_transports;
+       rc->transports = transports;
+
+       return rc;
+}
+
+
+/** Destroy a receiver
+ *
+ * @param[in] rc the receiver
+ * @return
+ *     - <0 on error
+ *     - 0 on success
+ */
+int fr_receiver_destroy(fr_receiver_t *rc)
+{
+       fr_receiver_worker_t *worker;
+       fr_channel_data_t *cd;
+
+#ifndef NDEBUG
+       (void) talloc_get_type_abort(rc, fr_receiver_t);
+#endif
+
+       /*
+        *      Pop all of the workers, and signal them that we're
+        *      closing/
+        */
+       while ((worker = fr_heap_pop(rc->workers)) != NULL) {
+               fr_channel_signal_close(worker->channel);
+               (void) fr_heap_insert(rc->closing, worker);
+       }
+
+       /*
+        *      @todo wait for all workers to acknowledge the channel
+        *      close.
+        */
+
+       /*
+        *      Clean up all of the replies.
+        */
+       while ((cd = fr_heap_pop(rc->replies)) != NULL) {
+               fr_message_done(&cd->m);
+       }
+
+       talloc_free(rc);
+
+       return 0;
+}
+
+void fr_receiver(fr_receiver_t *rc)
+{
+       while (fr_event_loop(rc->el) == 0) {
+               /* nothing */
+       }
+}

diff --git a/src/util/receiver.h b/src/util/receiver.h
new file mode 100644 (file)
index 0000000..cfb77d4
--- /dev/null
+++ b/src/util/receiver.h
@@ -0,0 +1,42 @@
+/*
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 2 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program; if not, write to the Free Software
+ *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
+ */
+#ifndef _FR_RECEIVE_H
+#define _FR_RECEIVE_H
+/**
+ * $Id$
+ *
+ * @file util/receiver.h
+ * @brief Receive packets
+ *
+ * @copyright 2016 Alan DeKok <aland@freeradius.org>
+ */
+RCSIDH(receiver_h, "$Id$")
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct fr_receiver_t fr_receiver_t;
+
+fr_receiver_t *fr_receiver_create(TALLOC_CTX *ctx, uint32_t num_transports, fr_transport_t **transports);
+int fr_receiver_destroy(fr_receiver_t *rc) CC_HINT(nonnull);
+void fr_receiver(fr_receiver_t *rc) CC_HINT(nonnull);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* _FR_RECEIVER_H */