--- /dev/null
+/*
+ * channel_test.c Tests for channels
+ *
+ * Version: $Id$
+ *
+ * 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
+ *
+ * Copyright 2016 Alan DeKok <aland@freeradius.org>
+ */
+
+RCSID("$Id$")
+
+#include <freeradius-devel/util/channel.h>
+#include <freeradius-devel/rad_assert.h>
+
+#ifdef HAVE_GETOPT_H
+# include <getopt.h>
+#endif
+
+#ifdef HAVE_PTHREAD_H
+#include <pthread.h>
+#endif
+
+#include <sys/event.h>
+
+#define MAX_MESSAGES (2048)
+#define MAX_KEVENTS (10)
+
+#define MPRINT1 if (debug_lvl) printf
+#define MPRINT2 if (debug_lvl > 1) printf
+
+static int debug_lvl = 0;
+static int kq_master, kq_worker;
+static int max_messages = 10;
+static int max_outstanding = 1;
+
+static void NEVER_RETURNS usage(void)
+{
+ fprintf(stderr, "usage: channel_test [OPTS]\n");
+ fprintf(stderr, " -x Debugging mode.\n");
+
+ exit(1);
+}
+
+#if 0
+SHIT
+
+- we have to have the array of events to listen for on every call to kevent()
+- which upsets some assumptions about code / API encapsulation
+- the behavior of EV_ADD and EV_ENABLE seem to be reversed from the documentation
+ - tho https://wiki.netbsd.org/tutorials/kqueue_tutorial/ says to do EV_ADD | EV_ENABLE
+-
+#endif
+
+static void *channel_master(void *arg)
+{
+ bool running, signaled_close;
+ int rcode, i;
+ int num_outstanding, num_messages;
+ int num_replies;
+ fr_message_set_t *ms;
+ TALLOC_CTX *ctx;
+ fr_channel_t *channel = arg;
+
+ int num_listen_events;
+ struct kevent listen_events[MAX_KEVENTS];
+ struct kevent received_events[MAX_KEVENTS];
+
+ ctx = talloc_init("channel_master");
+ if (!ctx) _exit(1);
+
+ ms = fr_message_set_create(ctx, MAX_MESSAGES, sizeof(fr_channel_data_t), MAX_MESSAGES * 1024);
+ if (!ms) {
+ fprintf(stderr, "Failed creating message set\n");
+ exit(1);
+ }
+
+ MPRINT1("Master started.\n");
+
+ /*
+ * Signal the worker that the channel is open
+ */
+ rcode = fr_channel_signal_open(kq_worker, channel);
+ if (rcode < 0) {
+ fprintf(stderr, "Failed signaling open: %s\n", strerror(errno));
+ exit(1);
+ }
+
+ /*
+ * Bootstrap the queue with messages.
+ */
+ num_replies = num_outstanding = num_messages = 0;
+
+ running = true;
+ signaled_close = false;
+
+ while (running) {
+ fr_time_t now;
+ int num_to_send;
+ fr_channel_data_t *cd, *reply;
+
+ /*
+ * Ensure we have outstanding messages.
+ */
+ if (num_messages >= max_messages) {
+ MPRINT1("Master DONE sending\n");
+ goto check_close;
+ }
+
+ num_to_send = max_outstanding - num_outstanding;
+ if ((num_messages + num_to_send) > max_messages) {
+ num_to_send = max_messages - num_messages;
+ }
+ MPRINT1("Master sending %d messages\n", num_to_send);
+
+ for (i = 0; i < num_to_send; i++) {
+ cd = (fr_channel_data_t *) fr_message_alloc(ms, NULL, 100);
+ rad_assert(cd != NULL);
+
+ num_outstanding++;
+ num_messages++;
+
+ cd->m.when = fr_time();
+ memcpy(cd->m.data, &num_messages, sizeof(num_messages));
+
+ MPRINT1("Master sent message %d\n", num_messages);
+ rcode = fr_channel_send_request(channel, cd, &reply);
+ if (rcode < 0) {
+ fprintf(stderr, "Failed sending request: %s\n", strerror(errno));
+ }
+ rad_assert(rcode == 0);
+ if (reply) {
+ num_replies++;
+ num_outstanding--;
+ MPRINT1("Master got reply %d, outstanding=%d, %d/%d sent.\n",
+ num_replies, num_outstanding, num_messages, max_messages);
+ fr_message_done(&reply->m);
+ }
+ }
+
+ /*
+ * Signal close only when done.
+ */
+check_close:
+ if (!signaled_close && (num_messages >= max_messages) && (num_outstanding == 0)) {
+ rcode = fr_channel_signal_close(channel, false);
+ if (rcode < 0) {
+ fprintf(stderr, "Failed signaling close: %s\n", strerror(errno));
+ exit(1);
+ }
+
+ signaled_close = true;
+ }
+
+ MPRINT1("Master waiting on events.\n");
+ rad_assert(num_messages <= max_messages);
+
+ num_listen_events = fr_channel_add_kevent_receiver(channel, listen_events, MAX_KEVENTS);
+ rcode = kevent(kq_master, listen_events, num_listen_events, received_events, MAX_KEVENTS, NULL);
+
+ MPRINT1("Master kevent returned %d\n", rcode);
+
+ if (rcode < 0) {
+ if (rcode == EINTR) continue;
+
+ fprintf(stderr, "Failed waiting for kevent: %s\n", strerror(errno));
+ exit(1);
+ }
+
+ if (rcode == 0) continue;
+
+ now = fr_time();
+
+ /*
+ * Service the events.
+ */
+ for (i = 0; i < rcode; i++) {
+ fr_channel_event_t ce;
+ fr_channel_t *new_channel;
+
+ ce = fr_channel_service_kevent(kq_worker, &received_events[i], now, &new_channel);
+ MPRINT1("\tWorker Got channel event %d\n", ce);
+
+ if (ce == FR_CHANNEL_DATA_READY_RECEIVER) {
+ MPRINT1("Master got data ready signal\n");
+
+ reply = fr_channel_recv_reply(channel);
+ if (!reply) continue;
+
+ do {
+ num_replies++;
+ num_outstanding--;
+ MPRINT1("Master got reply %d, outstanding=%d, %d/%d sent.\n",
+ num_replies, num_outstanding, num_messages, max_messages);
+ fr_message_done(&reply->m);
+ } while ((reply = fr_channel_recv_reply(channel)) != NULL);
+
+ continue;
+ }
+
+ if (ce == FR_CHANNEL_CLOSE) {
+ MPRINT1("Master received close signal\n");
+ rad_assert(signaled_close == true);
+ running = false;
+ continue;
+ }
+
+ fprintf(stderr, "Master got unexpected CE %d\n", ce);
+
+ /*
+ * Not written yet!
+ */
+ rad_assert(0 == 1);
+ }
+ }
+
+ MPRINT1("Master signaling worker to exit.\n");
+
+
+ // wait for final set of messages
+
+
+ MPRINT1("Master exiting.\n");
+
+ /*
+ * Force all messages to be garbage collected
+ */
+ MPRINT2("GC\n");
+ fr_message_set_gc(ms);
+
+ if (debug_lvl > 1) fr_message_set_debug(ms, stdout);
+
+ /*
+ * After the garbage collection, all messages marked "done" MUST also be marked "free".
+ */
+ rcode = fr_message_set_messages_used(ms);
+ MPRINT2("Master messages used = %d\n", rcode);
+ rad_assert(rcode == 0);
+
+ talloc_free(ctx);
+
+ return NULL;
+}
+
+static void *channel_worker(void *arg)
+{
+ bool running = true;
+ int rcode;
+ int worker_messages = 0;
+ fr_message_set_t *ms;
+ TALLOC_CTX *ctx;
+ fr_channel_t *channel = arg;
+
+ int num_listen_events;
+ struct kevent listen_events[MAX_KEVENTS];
+ struct kevent received_events[MAX_KEVENTS];
+
+ ctx = talloc_init("channel_worker");
+ if (!ctx) _exit(1);
+
+ ms = fr_message_set_create(ctx, MAX_MESSAGES, sizeof(fr_channel_data_t), MAX_MESSAGES * 1024);
+ if (!ms) {
+ fprintf(stderr, "Failed creating message set\n");
+ exit(1);
+ }
+
+ MPRINT1("\tWorker started.\n");
+
+ while (running) {
+ int i;
+ fr_time_t now;
+ fr_channel_t *new_channel;
+
+ MPRINT1("\tWorker waiting on events.\n");
+
+ num_listen_events = fr_channel_add_kevent_worker(channel, listen_events, MAX_KEVENTS);
+ rcode = kevent(kq_worker, listen_events, num_listen_events, received_events, MAX_KEVENTS, NULL);
+
+ MPRINT1("\tWorker kevent returned %d events\n", rcode);
+
+ if (rcode < 0) {
+ if (rcode == EINTR) continue;
+
+ fprintf(stderr, "Failed waiting for kevent: %s\n", strerror(errno));
+ exit(1);
+ }
+
+ if (rcode == 0) continue;
+
+ now = fr_time();
+
+ for (i = 0; i < rcode; i++) {
+ fr_channel_event_t ce;
+
+ ce = fr_channel_service_kevent(kq_worker, &received_events[i], now, &new_channel);
+ MPRINT1("\tWorker Got channel event %d\n", ce);
+
+ if (ce == FR_CHANNEL_OPEN) {
+ MPRINT1("\tWorker received a new channel\n");
+ rad_assert(new_channel == channel);
+ continue;
+ }
+
+ if (ce == FR_CHANNEL_CLOSE) {
+ fr_channel_data_t *cd;
+
+ MPRINT1("\tWorker requested to close the channel.\n");
+ running = false;
+
+ /*
+ * Drain the input before we ACK the exit.
+ */
+ while ((cd = fr_channel_recv_request(channel)) != NULL) {
+ worker_messages++;
+ MPRINT1("\tWorker got message %d\n", worker_messages);
+ fr_message_done(&cd->m);
+ }
+
+ (void) fr_channel_signal_close(channel, true);
+ continue;
+ }
+
+ if (ce == FR_CHANNEL_DATA_READY_WORKER) {
+ fr_channel_data_t *cd, *reply;
+
+ MPRINT1("\tWorker got data ready signal\n");
+
+ cd = fr_channel_recv_request(channel);
+ rad_assert(cd != NULL);
+ while (cd) {
+ int message_id;
+
+ worker_messages++;
+
+ rad_assert(cd->m.data != NULL);
+ memcpy(&message_id, cd->m.data, sizeof(message_id));
+ MPRINT1("\tWorker got message %d (says %d)\n", worker_messages, message_id);
+
+ reply = (fr_channel_data_t *) fr_message_alloc(ms, NULL, 100);
+ rad_assert(reply != NULL);
+
+ reply->m.when = fr_time();
+ fr_message_done(&cd->m);
+
+ MPRINT1("\tWorker sending reply to messages %d\n", worker_messages);
+ rcode = fr_channel_send_reply(channel, reply, &cd);
+ if (rcode < 0) {
+ fprintf(stderr, "Failed sending reply: %s\n", strerror(errno));
+ }
+ rad_assert(rcode == 0);
+ }
+ continue;
+ }
+
+ fprintf(stderr, "Worker got unexpected CE %d\n", ce);
+
+ /*
+ * Not written yet!
+ */
+ rad_assert(0 == 1);
+ }
+ }
+
+ MPRINT1("\tWorker exiting.\n");
+
+ /*
+ * Force all messages to be garbage collected
+ */
+ MPRINT2("Worker GC\n");
+ fr_message_set_gc(ms);
+
+ if (debug_lvl > 1) fr_message_set_debug(ms, stdout);
+
+ /*
+ * After the garbage collection, all messages marked "done" MUST also be marked "free".
+ */
+ rcode = fr_message_set_messages_used(ms);
+ rad_assert(rcode == 0);
+
+ talloc_free(ctx);
+
+ return NULL;
+}
+
+
+
+int main(int argc, char *argv[])
+{
+ int c;
+ fr_channel_t *channel;
+ TALLOC_CTX *autofree = talloc_init("main");
+ pthread_attr_t attr;
+ pthread_t master_id, worker_id;
+
+ fr_time_start();
+
+ while ((c = getopt(argc, argv, "hm:o:x")) != EOF) switch (c) {
+ case 'x':
+ debug_lvl++;
+ break;
+
+ case 'm':
+ max_messages = atoi(optarg);
+ break;
+
+ case 'o':
+ max_outstanding = atoi(optarg);
+ break;
+
+ case 'h':
+ default:
+ usage();
+ }
+
+#if 0
+ argc -= (optind - 1);
+ argv += (optind - 1);
+#endif
+
+ kq_master = kqueue();
+ rad_assert(kq_master >= 0);
+
+ kq_worker = kqueue();
+ rad_assert(kq_worker >= 0);
+
+ channel = fr_channel_create(autofree, kq_master, kq_worker);
+ if (!channel) {
+ fprintf(stderr, "channel_test: Failed to create channel\n");
+ exit(1);
+ }
+
+ /*
+ * Start the two threads, with the channel.
+ */
+ (void) pthread_attr_init(&attr);
+ (void) pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
+
+ (void) pthread_create(&master_id, &attr, channel_master, channel);
+ (void) pthread_create(&worker_id, &attr, channel_worker, channel);
+
+ (void) pthread_join(master_id, NULL);
+ (void) pthread_join(worker_id, NULL);
+
+ close(kq_master);
+ close(kq_worker);
+
+ talloc_free(autofree);
+
+ return 0;
+}
projects / thirdparty / freeradius-server.git / commitdiff
