From: Alan T. DeKok Date: Mon, 28 Nov 2016 21:54:30 +0000 (-0500) Subject: added receiver code X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=ca5eccaab6f52a7988360a0ebfad196402a29c49;p=thirdparty%2Ffreeradius-server.git added receiver code Basic infrastructure for now. More later. --- diff --git a/src/util/all.mk b/src/util/all.mk index a12f1c35752..203d5c06544 100644 --- 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 index 00000000000..b52f3fb6bbb --- /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 + */ +RCSID("$Id$") + +#include + +#include +#include +#include +#include + +#include + +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 index 00000000000..cfb77d438ee --- /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 + */ +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 */