From: Alan T. DeKok Date: Sat, 3 Dec 2016 16:58:44 +0000 (-0500) Subject: channel test code X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=e20a012d086391d4f9bd97d39767a0e2bd9f6fc7;p=thirdparty%2Ffreeradius-server.git channel test code --- diff --git a/src/tests/util/all.mk b/src/tests/util/all.mk index bac2567fa4f..346ca0d3e04 100644 --- a/src/tests/util/all.mk +++ b/src/tests/util/all.mk @@ -1 +1,8 @@ SUBMAKEFILES := ring_buffer_test.mk message_set_test.mk atomic_queue_test.mk + +# +# This requires pthread. +# +ifneq "$(findstring thread,${CFLAGS})" "" +SUBMAKEFILES += channel_test.mk +endif diff --git a/src/tests/util/channel_test.c b/src/tests/util/channel_test.c new file mode 100644 index 00000000000..58b2243c32e --- /dev/null +++ b/src/tests/util/channel_test.c @@ -0,0 +1,462 @@ +/* + * 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 + */ + +RCSID("$Id$") + +#include +#include + +#ifdef HAVE_GETOPT_H +# include +#endif + +#ifdef HAVE_PTHREAD_H +#include +#endif + +#include + +#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; +}