--- /dev/null
+/*
+ * worker_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/worker.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_CONTROL_PLANE (1024)
+#define MAX_KEVENTS (10)
+#define MAX_WORKERS (1024)
+
+#define MPRINT1 if (debug_lvl) printf
+#define MPRINT2 if (debug_lvl > 1) printf
+
+typedef struct fr_schedule_worker_t {
+ int id; //!< ID of the worker 0..N
+ pthread_t pthread_id; //!< pthread ID of the worker
+ fr_worker_t *worker; //!< pointer to the worker
+ fr_channel_t *ch; //!< channel for communicating with the worker
+} fr_schedule_worker_t;
+
+static int debug_lvl = 0;
+static int kq_master;
+static fr_atomic_queue_t *aq_master;
+static int max_messages = 10;
+static int max_control_plane = 0;
+static int max_outstanding = 1;
+static bool touch_memory = false;
+static int num_workers = 1;
+static fr_schedule_worker_t workers[MAX_WORKERS];
+
+static void NEVER_RETURNS usage(void)
+{
+ fprintf(stderr, "usage: worker_test [OPTS]\n");
+ fprintf(stderr, " -c <control-plane> Size of the control plane queue.\n");
+ fprintf(stderr, " -m <messages> Send number of messages.\n");
+ fprintf(stderr, " -o <outstanding> Keep number of messages outstanding.\n");
+ fprintf(stderr, " -t Touch memory for fake packets.\n");
+ fprintf(stderr, " -w N Create N workers. Default is 1.\n");
+ fprintf(stderr, " -x Debugging mode.\n");
+
+ exit(1);
+}
+
+static int test_decode(void const *packet_ctx, uint8_t *const data, size_t data_len, REQUEST *request)
+{
+ uint32_t number;
+
+ /*
+ * The data is the packet number.
+ */
+ memcpy(&number, data, sizeof(number));
+ request->number = number;
+
+ MPRINT1("\t\tDECODE <<< request %zd - %p data %p size %zd\n", request->number, packet_ctx, data, data_len);
+ return 0;
+}
+
+static ssize_t test_encode(void const *packet_ctx, REQUEST *request, uint8_t *const data, size_t data_len)
+{
+ MPRINT1("\t\tENCODE >>> request %zd - data %p %p size %zd\n", request->number, packet_ctx, data, data_len);
+
+ return data_len;
+}
+
+static fr_transport_final_t test_process(REQUEST *request, fr_transport_action_t action)
+{
+ MPRINT1("\t\tPROCESS --- request %zd action %d\n", request->number, action);
+ return FR_TRANSPORT_REPLY;
+}
+
+static fr_transport_t transport = {
+ .name = "worker-test",
+ .id = 1,
+ .decode = test_decode,
+ .encode = test_encode,
+ .process = test_process,
+};
+
+static fr_transport_t *transports = &transport;
+
+static void *worker_thread(void *arg)
+{
+ TALLOC_CTX *ctx;
+ fr_worker_t *worker;
+ fr_schedule_worker_t *sw;
+
+ sw = (fr_schedule_worker_t *) arg;
+
+ MPRINT1("\tWorker %d started.\n", sw->id);
+
+ ctx = talloc_init("worker");
+ if (!ctx) _exit(1);
+
+ worker = sw->worker = fr_worker_create(ctx, 1, &transports);
+ if (!worker) {
+ fprintf(stderr, "worker_test: Failed to create the worker\n");
+ exit(1);
+ }
+
+
+ MPRINT1("\tWorker %d looping.\n", sw->id);
+ fr_worker(worker);
+
+ sw->worker = NULL;
+ MPRINT1("\tWorker %d exiting.\n", sw->id);
+
+ talloc_free(ctx);
+ return NULL;
+}
+
+
+static void master_process(void)
+{
+ bool running, signaled_close;
+ int rcode, i, num_events, which_worker;
+ int num_outstanding, num_messages;
+ int num_replies;
+ fr_message_set_t *ms;
+ TALLOC_CTX *ctx;
+ fr_channel_t *ch;
+ fr_channel_event_t ce;
+ pthread_attr_t attr;
+ fr_schedule_worker_t *sw;
+ struct kevent events[MAX_KEVENTS];
+
+ ctx = talloc_init("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");
+
+ /*
+ * Create the worker threads.
+ */
+ (void) pthread_attr_init(&attr);
+ (void) pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
+
+ for (i = 0; i < num_workers; i++) {
+ workers[i].id = i;
+ (void) pthread_create(&workers[i].pthread_id, &attr, worker_thread, &workers[i]);
+ }
+
+ MPRINT1("Master created %d workers.\n", num_workers);
+
+ /*
+ * Busy loop because that's fine for the test
+ */
+ num_outstanding = 0;
+ while (num_outstanding < num_workers) {
+ for (i = 0; i < num_workers; i++) {
+ if (!workers[i].worker) continue;
+ if (workers[i].ch != NULL) continue;
+
+ /*
+ * Create the channel and signal the
+ * worker that it is open
+ */
+ MPRINT1("Master creating channel to worker %d.\n", num_workers);
+ workers[i].ch = fr_worker_channel_create(ctx, workers[i].worker, kq_master, aq_master);
+ rad_assert(workers[i].ch != NULL);
+
+ (void) fr_channel_master_ctx_add(workers[i].ch, &workers[i]);
+
+ rcode = fr_channel_signal_open(workers[i].ch);
+ if (rcode < 0) {
+ fprintf(stderr, "Failed signaling open %d: %s\n", i, strerror(errno));
+ exit(1);
+ }
+
+ num_outstanding++;
+ }
+ }
+
+ MPRINT1("Master created all channels.\n");
+
+ /*
+ * Bootstrap the queue with messages.
+ */
+ num_replies = num_outstanding = num_messages = 0;
+ which_worker = 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();
+
+ if (touch_memory) {
+ size_t j, k;
+
+ for (j = k = 0; j < cd->m.data_size; j++) {
+ k += cd->m.data[j];
+ }
+
+ cd->m.data[4] = k;
+ }
+
+ memcpy(cd->m.data, &num_messages, sizeof(num_messages));
+
+ MPRINT1("Master sent message %d to worker %d\n", num_messages, which_worker);
+ rcode = fr_channel_send_request(workers[which_worker].ch, cd, &reply);
+ if (rcode < 0) {
+ fprintf(stderr, "Failed sending request: %s\n", strerror(errno));
+ }
+ which_worker++;
+ if (which_worker >= num_workers) which_worker = 0;
+
+ 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)) {
+ MPRINT1("Master signaling workers to exit.\n");
+
+ for (i = 0; i < num_workers; i++) {
+ printf("Worker %d\n", i);
+ fr_worker_debug(workers[i].worker, stdout);
+
+ rcode = fr_channel_signal_worker_close(workers[i].ch);
+ if (rcode < 0) {
+ fprintf(stderr, "Failed signaling close %d: %s\n", i, strerror(errno));
+ exit(1);
+ }
+ }
+ signaled_close = true;
+ }
+
+ MPRINT1("Master waiting on events.\n");
+ rad_assert(num_messages <= max_messages);
+
+ num_events = kevent(kq_master, NULL, 0, events, MAX_KEVENTS, NULL);
+ MPRINT1("Master kevent returned %d\n", num_events);
+
+ if (num_events < 0) {
+ if (num_events == EINTR) continue;
+
+ fprintf(stderr, "Failed waiting for kevent: %s\n", strerror(errno));
+ exit(1);
+ }
+
+ if (num_events == 0) continue;
+
+ /*
+ * Service the events.
+ *
+ * @todo this should NOT take a channel pointer
+ */
+ for (i = 0; i < num_events; i++) {
+ (void) fr_channel_service_kevent(workers[0].ch, aq_master, &events[i]);
+ }
+
+ now = fr_time();
+
+ MPRINT1("Master servicing control-plane\n");
+
+ while ((ce = fr_channel_service_aq(aq_master, now, &ch)) != FR_CHANNEL_EMPTY) {
+ MPRINT1("Master got channel event %d\n", ce);
+
+ switch (ce) {
+ case FR_CHANNEL_DATA_READY_RECEIVER:
+ MPRINT1("Master got data ready signal\n");
+
+ reply = fr_channel_recv_reply(ch);
+ if (!reply) {
+ MPRINT1("Master SIGNAL WITH NO DATA!\n");
+ 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(ch)) != NULL);
+ break;
+
+ case FR_CHANNEL_CLOSE:
+ sw = fr_channel_master_ctx_get(ch);
+ rad_assert(sw != NULL);
+
+ MPRINT1("Master received close signal for worker %d\n", sw->id);
+ rad_assert(signaled_close == true);
+
+
+ /*
+ * Tell the event loop to exit, and signal the worker
+ * so that it stops waiting on the KQ.
+ */
+ (void) fr_worker_exit(sw->worker);
+ (void) pthread_kill(sw->pthread_id, SIGTERM);
+ running = false;
+ break;
+
+ case FR_CHANNEL_NOOP:
+ break;
+
+ default:
+ fprintf(stderr, "Master got unexpected CE %d\n", ce);
+
+ /*
+ * Not written yet!
+ */
+ rad_assert(0 == 1);
+ break;
+ } /* switch over signal returned */
+ } /* drain the control plane */
+ } /* loop until told to exit */
+
+ MPRINT1("Master exiting.\n");
+
+ /*
+ * Busy-wait for the workers to exit;
+ */
+ do {
+ num_outstanding = num_workers;
+
+ for (i = 0; i < num_workers; i++) {
+ if (workers[i].worker == NULL) num_outstanding--;
+ }
+ } while (num_outstanding > 0);
+
+ /*
+ * 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);
+
+}
+
+static void sig_ignore(int sig)
+{
+ (void) signal(sig, sig_ignore);
+}
+
+int main(int argc, char *argv[])
+{
+ int c;
+ TALLOC_CTX *autofree = talloc_init("main");
+
+ fr_time_start();
+
+ while ((c = getopt(argc, argv, "c:hm:o:tw:x")) != EOF) switch (c) {
+ case 'x':
+ debug_lvl++;
+ break;
+
+ case 'c':
+ max_control_plane = atoi(optarg);
+ break;
+
+ case 'm':
+ max_messages = atoi(optarg);
+ break;
+
+ case 'o':
+ max_outstanding = atoi(optarg);
+ break;
+
+ case 't':
+ touch_memory = true;
+ break;
+
+ case 'w':
+ num_workers = atoi(optarg);
+ if ((num_workers <= 0) || (num_workers >= MAX_WORKERS)) usage();
+ break;
+
+ case 'h':
+ default:
+ usage();
+ }
+
+ if (max_outstanding > max_messages) max_outstanding = max_messages;
+
+ if (!max_control_plane) {
+ max_control_plane = MAX_CONTROL_PLANE;
+ if (max_outstanding > max_control_plane) max_control_plane = max_outstanding;
+ }
+
+#if 0
+ argc -= (optind - 1);
+ argv += (optind - 1);
+#endif
+
+ kq_master = kqueue();
+ rad_assert(kq_master >= 0);
+
+ aq_master = fr_atomic_queue_create(autofree, max_control_plane);
+ rad_assert(aq_master != NULL);
+
+ signal(SIGTERM, sig_ignore);
+
+ if (debug_lvl) {
+ setvbuf(stdout, NULL, _IONBF, 0);
+ }
+
+ master_process();
+
+ close(kq_master);
+
+ talloc_free(autofree);
+
+ return 0;
+}
projects / thirdparty / freeradius-server.git / commitdiff
