/*
 *   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: 5f85964fe7b49c741b07ec97a8a9aec32345a67f $
 *
 * @brief Network / worker thread scheduling
 * @file io/schedule.c
 *
 * @copyright 2016 Alan DeKok (aland@freeradius.org)
 */
RCSID("$Id: 5f85964fe7b49c741b07ec97a8a9aec32345a67f $")

#define LOG_DST sc->log

#include <freeradius-devel/autoconf.h>

#include <freeradius-devel/io/schedule.h>
#include <freeradius-devel/util/dlist.h>
#include <freeradius-devel/util/rb.h>
#include <freeradius-devel/util/syserror.h>
#include <freeradius-devel/server/trigger.h>

#include <pthread.h>

/*
 *	Other OS's have sem_init, OS X doesn't.
 */
#ifdef HAVE_SEMAPHORE_H
#include <semaphore.h>
#endif

#define SEMAPHORE_LOCKED	(0)

#ifdef __APPLE__
#include <mach/task.h>
#include <mach/mach_init.h>
#include <mach/semaphore.h>

#undef sem_t
#define sem_t semaphore_t
#undef sem_init
#define sem_init(s,p,c) semaphore_create(mach_task_self(),s,SYNC_POLICY_FIFO,c)
#undef sem_wait
#define sem_wait(s) semaphore_wait(*s)
#undef sem_post
#define sem_post(s) semaphore_signal(*s)
#undef sem_destroy
#define sem_destroy(s) semaphore_destroy(mach_task_self(),*s)
#endif	/* __APPLE__ */

#define SEM_WAIT_INTR(_x) do {if (sem_wait(_x) == 0) break;} while (errno == EINTR)

/**
 *  Track the child thread status.
 */
typedef enum fr_schedule_child_status_t {
	FR_CHILD_FREE = 0,			//!< child is free
	FR_CHILD_INITIALIZING,			//!< initialized, but not running
	FR_CHILD_RUNNING,			//!< running, and in the running queue
	FR_CHILD_EXITED,			//!< exited, and in the exited queue
	FR_CHILD_FAIL				//!< failed, and in the exited queue
} fr_schedule_child_status_t;

/** Scheduler specific information for worker threads
 *
 * Wraps a fr_worker_t, tracking additional information that
 * the scheduler uses.
 */
typedef struct {
	TALLOC_CTX	*ctx;			//!< our allocation ctx
	fr_event_list_t	*el;			//!< our event list
	pthread_t	pthread_id;		//!< the thread of this worker

	unsigned int	id;			//!< a unique ID
	int		uses;			//!< how many network threads are using it
	fr_time_t	cpu_time;		//!< how much CPU time this worker has used

	fr_dlist_t	entry;			//!< our entry into the linked list of workers

	fr_schedule_t	*sc;			//!< the scheduler we are running under

	fr_schedule_child_status_t status;	//!< status of the worker
	fr_worker_t	*worker;		//!< the worker data structure
} fr_schedule_worker_t;

/** Scheduler specific information for network threads
 *
 * Wraps a fr_network_t, tracking additional information that
 * the scheduler uses.
 */
typedef struct {
	TALLOC_CTX	*ctx;			//!< our allocation ctx
	pthread_t	pthread_id;		//!< the thread of this network

	unsigned int	id;			//!< a unique ID

	fr_dlist_t	entry;			//!< our entry into the linked list of networks

	fr_schedule_t	*sc;			//!< the scheduler we are running under

	fr_schedule_child_status_t status;	//!< status of the worker
	fr_network_t	*nr;			//!< the receive data structure

	fr_event_timer_t const *ev;		//!< timer for stats_interval
} fr_schedule_network_t;


/**
 *  The scheduler
 */
struct fr_schedule_s {
	bool		running;		//!< is the scheduler running?

	CONF_SECTION	*cs;			//!< thread pool configuration section
	fr_event_list_t	*el;			//!< event list for single-threaded mode.

	fr_log_t	*log;			//!< log destination
	fr_log_lvl_t	lvl;			//!< log level

	fr_schedule_config_t *config;		//!< configuration

	unsigned int	num_workers_exited;	//!< number of exited workers

	sem_t		worker_sem;		//!< for inter-thread signaling
	sem_t		network_sem;		//!< for inter-thread signaling

	fr_schedule_thread_instantiate_t	worker_thread_instantiate;	//!< thread instantiation callback
	fr_schedule_thread_detach_t		worker_thread_detach;

	fr_dlist_head_t	workers;		//!< list of workers
	fr_dlist_head_t	networks;		//!< list of networks

	fr_network_t	*single_network;	//!< for single-threaded mode
	fr_worker_t	*single_worker;		//!< for single-threaded mode
};

static _Thread_local int worker_id;		//!< Internal ID of the current worker thread.

/** Return the worker id for the current thread
 *
 * @return worker ID
 */
int fr_schedule_worker_id(void)
{
	return worker_id;
}

/** Entry point for worker threads
 *
 * @param[in] arg	the fr_schedule_worker_t
 * @return NULL
 */
static void *fr_schedule_worker_thread(void *arg)
{
	TALLOC_CTX			*ctx;
	fr_schedule_worker_t		*sw = talloc_get_type_abort(arg, fr_schedule_worker_t);
	fr_schedule_t			*sc = sw->sc;
	fr_schedule_child_status_t	status = FR_CHILD_FAIL;
	fr_schedule_network_t		*sn;
	char				worker_name[32];

#ifndef __APPLE__
	/*
	 * This ifdef is because macOS doesn't use pthread_signmask in its
	 * setcontext function, and seems to apply the signal mask of the thread
	 * to the entire process when setcontext is called.
	 *
	 *  * frame #0: 0x00000001934118b0 libsystem_kernel.dylib`sigprocmask
	 *  frame #1: 0x0000000193481f3c libsystem_platform.dylib`setcontext + 44
	 *  frame #2: 0x0000000100f27298 libcrypto.3.dylib`async_fibre_swapcontext + 52
	 *  frame #3: 0x0000000100f274a0 libcrypto.3.dylib`ASYNC_start_job + 496
	 *  frame #4: 0x0000000100b17884 libssl.3.dylib`ssl_start_async_job + 116
	 *  frame #5: 0x0000000100b17804 libssl.3.dylib`ssl_read_internal + 356
	 *  frame #6: 0x0000000100b17a0c libssl.3.dylib`SSL_read + 28
	 *  frame #7: 0x00000001004f5b94 libfreeradius-tls.dylib`tls_session_async_handshake_cont(p_result=0x0000000112815c7c, priority=0x0000000112815edc, request=0x0000000112815a80, uctx=0x0000000139160060) at session.c:1366:26
	 */
	sigset_t			sigset;

	sigfillset(&sigset);

	/*
	 *	Ensure workers aren't interrupted by signals.
	 *	The main thread, and main event loop are mostly
	 *	idle, so they can handle signals.
	 */
	pthread_sigmask(SIG_BLOCK, &sigset, NULL);
#endif

	worker_id = sw->id;		/* Store the current worker ID */

	snprintf(worker_name, sizeof(worker_name), "Worker %d", sw->id);

	sw->ctx = ctx = talloc_init("%s", worker_name);
	if (!ctx) {
		ERROR("%s - Failed allocating memory", worker_name);
		goto fail;
	}

	INFO("%s - Starting", worker_name);

	sw->el = fr_event_list_alloc(ctx, NULL, NULL);
	if (!sw->el) {
		PERROR("%s - Failed creating event list", worker_name);
		goto fail;
	}


	sw->worker = fr_worker_create(ctx, sw->el, worker_name, sc->log, sc->lvl, &sc->config->worker);
	if (!sw->worker) {
		PERROR("%s - Failed creating worker", worker_name);
		goto fail;
	}

	/*
	 *	@todo make this a registry
	 */
	if (sc->worker_thread_instantiate) {
		CONF_SECTION	*cs;
		char		section_name[32];

		snprintf(section_name, sizeof(section_name), "%u", sw->id);

		cs = cf_section_find(sc->cs, "worker", section_name);
		if (!cs) cs = cf_section_find(sc->cs, "worker", NULL);

		if (sc->worker_thread_instantiate(sw->ctx, sw->el, cs) < 0) {
			PERROR("%s - Worker thread instantiation failed", worker_name);
			goto fail;
		}
	}

	sw->status = FR_CHILD_RUNNING;

	/*
	 *	Add this worker to all network threads.
	 */
	for (sn = fr_dlist_head(&sc->networks);
	     sn != NULL;
	     sn = fr_dlist_next(&sc->networks, sn)) {
		(void) fr_network_worker_add(sn->nr, sw->worker);
	}

	DEBUG3("%s - Started", worker_name);

	/*
	 *	Tell the originator that the thread has started.
	 */
	sem_post(&sc->worker_sem);

	/*
	 *	Do all of the work.
	 */
	fr_worker(sw->worker);

	status = FR_CHILD_EXITED;

fail:
	sw->status = status;

	if (sw->worker) {
		fr_worker_destroy(sw->worker);
		sw->worker = NULL;
	}

	INFO("%s - Exiting", worker_name);

	if (sc->worker_thread_detach) sc->worker_thread_detach(NULL);	/* Fixme once we figure out what uctx should be */

	/*
	 *	Not looping at this point, but may catch timer/fd
	 *	insertions being done after the thread should have
	 *	exited.
	 */
	if (sw->el) fr_event_loop_exit(sw->el, 1);

	/*
	 *	Tell the scheduler we're done.
	 */
	sem_post(&sc->worker_sem);

	talloc_free(ctx);

	return NULL;
}


static void stats_timer(fr_event_list_t *el, fr_time_t now, void *uctx)
{
	fr_schedule_network_t		*sn = talloc_get_type_abort(uctx, fr_schedule_network_t);

	fr_network_stats_log(sn->nr, sn->sc->log);

	(void) fr_event_timer_at(sn, el, &sn->ev, fr_time_add(now, sn->sc->config->stats_interval), stats_timer, sn);
}

/** Initialize and run the network thread.
 *
 * @param[in] arg the fr_schedule_network_t
 * @return NULL
 */
static void *fr_schedule_network_thread(void *arg)
{
	TALLOC_CTX			*ctx;
	fr_schedule_network_t		*sn = talloc_get_type_abort(arg, fr_schedule_network_t);
	fr_schedule_t			*sc = sn->sc;
	fr_schedule_child_status_t	status = FR_CHILD_FAIL;
	fr_event_list_t			*el;
	char				network_name[32];

#ifndef __APPLE__
	/*
	 * This ifdef is because macOS doesn't use pthread_signmask in its
	 * setcontext function, and seems to apply the signal mask of the thread
	 * to the entire process when setcontext is called.
	 *
	 *  * frame #0: 0x00000001934118b0 libsystem_kernel.dylib`sigprocmask
	 *  frame #1: 0x0000000193481f3c libsystem_platform.dylib`setcontext + 44
	 *  frame #2: 0x0000000100f27298 libcrypto.3.dylib`async_fibre_swapcontext + 52
	 *  frame #3: 0x0000000100f274a0 libcrypto.3.dylib`ASYNC_start_job + 496
	 *  frame #4: 0x0000000100b17884 libssl.3.dylib`ssl_start_async_job + 116
	 *  frame #5: 0x0000000100b17804 libssl.3.dylib`ssl_read_internal + 356
	 *  frame #6: 0x0000000100b17a0c libssl.3.dylib`SSL_read + 28
	 *  frame #7: 0x00000001004f5b94 libfreeradius-tls.dylib`tls_session_async_handshake_cont(p_result=0x0000000112815c7c, priority=0x0000000112815edc, request=0x0000000112815a80, uctx=0x0000000139160060) at session.c:1366:26
	 */
	sigset_t			sigset;

	sigfillset(&sigset);

	/*
	 *	Ensure workers aren't interrupted by signals.
	 *	The main thread, and main event loop are mostly
	 *	idle, so they can handle signals.
	 */
	pthread_sigmask(SIG_BLOCK, &sigset, NULL);
#endif

	snprintf(network_name, sizeof(network_name), "Network %d", sn->id);

	INFO("%s - Starting", network_name);

	sn->ctx = ctx = talloc_init("%s", network_name);
	if (!ctx) {
		ERROR("%s - Failed allocating memory", network_name);
		goto fail;
	}

	el = fr_event_list_alloc(ctx, NULL, NULL);
	if (!el) {
		PERROR("%s - Failed creating event list", network_name);
		goto fail;
	}

	sn->nr = fr_network_create(ctx, el, network_name, sc->log, sc->lvl, &sc->config->network);
	if (!sn->nr) {
		PERROR("%s - Failed creating network", network_name);
		goto fail;
	}

	sn->status = FR_CHILD_RUNNING;

	/*
	 *	Tell the originator that the thread has started.
	 */
	sem_post(&sc->network_sem);

	DEBUG3("%s - Started", network_name);

	/*
	 *	Print out statistics for this network IO handler.
	 */
	if (fr_time_delta_ispos(sc->config->stats_interval)) {
		(void) fr_event_timer_in(sn, el, &sn->ev, sn->sc->config->stats_interval, stats_timer, sn);
	}
	/*
	 *	Call the main event processing loop of the network
	 *	thread Will not return until the worker is about
	 *      to exit.
	 */
	fr_network(sn->nr);

	status = FR_CHILD_EXITED;

fail:
	sn->status = status;

	INFO("%s - Exiting", network_name);

	/*
	 *	Tell the scheduler we're done.
	 */
	sem_post(&sc->network_sem);

	talloc_free(ctx);

	return NULL;
}

/** Creates a new thread using our standard set of options
 *
 * New threads are:
 * - Joinable, i.e. you can call pthread_join on them to confirm they've exited
 * - Immune to catchable signals.
 *
 * @param[out] thread		handled that was created by pthread_create.
 * @param[in] func		entry point for the thread.
 * @param[in] arg		Argument to pass to func.
 * @return
 *	- 0 on success.
 *	- -1 on failure.
 */
int fr_schedule_pthread_create(pthread_t *thread, void *(*func)(void *), void *arg)
{
	pthread_attr_t			attr;
	int				ret;

	/*
	 *	Set the thread to wait around after it's exited
	 *	so it can be joined.  This is more of a useful
	 *	mechanism for the parent to determine if all
	 *	the threads have exited so it can continue with
	 *	a graceful shutdown.
	 */
	pthread_attr_init(&attr);
	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

	ret = pthread_create(thread, &attr, func, arg);
	if (ret != 0) {
		fr_strerror_printf("Failed creating thread: %s", fr_syserror(ret));
		return -1;
	}

	return 0;
}

/** Create a scheduler and spawn the child threads.
 *
 * @param[in] ctx				talloc context.
 * @param[in] el				event list, only for single-threaded mode.
 * @param[in] logger				destination for all logging messages.
 * @param[in] lvl				log level.
 * @param[in] worker_thread_instantiate		callback for new worker threads.
 * @param[in] worker_thread_detach		callback to destroy resources
 *						allocated by worker_thread_instantiate.
 * @param[in] config				configuration for the scheduler
 * @return
 *	- NULL on error
 *	- fr_schedule_t new scheduler
 */
fr_schedule_t *fr_schedule_create(TALLOC_CTX *ctx, fr_event_list_t *el,
				  fr_log_t *logger, fr_log_lvl_t lvl,
				  fr_schedule_thread_instantiate_t worker_thread_instantiate,
				  fr_schedule_thread_detach_t worker_thread_detach,
				  fr_schedule_config_t *config)
{
	unsigned int i;
	fr_schedule_worker_t *sw, *next_sw;
	fr_schedule_network_t *sn, *next_sn;
	fr_schedule_t *sc;

	sc = talloc_zero(ctx, fr_schedule_t);
	if (!sc) {
		fr_strerror_const("Failed allocating memory");
		return NULL;
	}

	sc->config = config;
	sc->el = el;
	sc->log = logger;
	sc->lvl = lvl;

	sc->worker_thread_instantiate = worker_thread_instantiate;
	sc->worker_thread_detach = worker_thread_detach;
	sc->running = true;

	/*
	 *	If we're single-threaded, create network / worker, and insert them into the event loop.
	 */
	if (el) {
		sc->single_network = fr_network_create(sc, el, "Network", sc->log, sc->lvl, &sc->config->network);
		if (!sc->single_network) {
			PERROR("Failed creating network");
		pre_instantiate_st_fail:
			talloc_free(sc);
			return NULL;
		}

		sc->single_worker = fr_worker_create(sc, el, "Worker", sc->log, sc->lvl, &sc->config->worker);
		if (!sc->single_worker) {
			PERROR("Failed creating worker");
			fr_network_destroy(sc->single_network);
			goto pre_instantiate_st_fail;
		}

		/*
		 *	Parent thread-specific data from the single_worker
		 */
		if (sc->worker_thread_instantiate) {
			CONF_SECTION *subcs;

			subcs = cf_section_find(sc->cs, "worker", "0");
			if (!subcs) subcs = cf_section_find(sc->cs, "worker", NULL);

			if (sc->worker_thread_instantiate(sc->single_worker, el, subcs) < 0) {
				PERROR("Worker thread instantiation failed");
			destroy_both:
				fr_network_destroy(sc->single_network);
				fr_worker_destroy(sc->single_worker);
				goto pre_instantiate_st_fail;
			}
		}

		if (fr_command_register_hook(NULL, "0", sc->single_worker, cmd_worker_table) < 0) {
			PERROR("Failed adding worker commands");
		st_fail:
			if (sc->worker_thread_detach) sc->worker_thread_detach(NULL);
			goto destroy_both;
		}

		if (fr_command_register_hook(NULL, "0", sc->single_network, cmd_network_table) < 0) {
			PERROR("Failed adding network commands");
			goto st_fail;
		}

		/*
		 *	Register the worker with the network, so
		 *	things like fr_network_send_request() work.
		 */
		fr_network_worker_add_self(sc->single_network, sc->single_worker);
		DEBUG("Scheduler created in single-threaded mode");

		if (fr_event_pre_insert(el, fr_worker_pre_event, sc->single_worker) < 0) {
			fr_strerror_const("Failed adding pre-check to event list");
			goto st_fail;
		}

		/*
		 *	Add the event which processes request_t packets.
		 */
		if (fr_event_post_insert(el, fr_worker_post_event, sc->single_worker) < 0) {
			fr_strerror_const("Failed inserting post-processing event");
			goto st_fail;
		}

		return sc;
	}

	/*
	 *	Parse any scheduler-specific configuration.
	 */
	if (!config) {
		MEM(sc->config = talloc_zero(sc, fr_schedule_config_t));
		sc->config->max_networks = 1;
		sc->config->max_workers = 4;
	} else {
		sc->config = config;

		if (sc->config->max_networks < 1) sc->config->max_networks = 1;
		if (sc->config->max_networks > 64) sc->config->max_networks = 64;
		if (sc->config->max_workers < 1) sc->config->max_workers = 1;
		if (sc->config->max_workers > 64) sc->config->max_workers = 64;
	}

	/*
	 *	Create the lists which hold the workers and networks.
	 */
	fr_dlist_init(&sc->workers, fr_schedule_worker_t, entry);
	fr_dlist_init(&sc->networks, fr_schedule_network_t, entry);

	memset(&sc->network_sem, 0, sizeof(sc->network_sem));
	if (sem_init(&sc->network_sem, 0, SEMAPHORE_LOCKED) != 0) {
		ERROR("Failed creating semaphore: %s", fr_syserror(errno));
		talloc_free(sc);
		return NULL;
	}

	memset(&sc->worker_sem, 0, sizeof(sc->worker_sem));
	if (sem_init(&sc->worker_sem, 0, SEMAPHORE_LOCKED) != 0) {
		ERROR("Failed creating semaphore: %s", fr_syserror(errno));
		talloc_free(sc);
		return NULL;
	}

	/*
	 *	Create the network threads first.
	 */
	for (i = 0; i < sc->config->max_networks; i++) {
		DEBUG3("Creating %u/%u networks", i + 1, sc->config->max_networks);

		/*
		 *	Create a worker "glue" structure
		 */
		sn = talloc_zero(sc, fr_schedule_network_t);
		if (!sn) {
			ERROR("Network %u - Failed allocating memory", i);
			break;
		}

		sn->id = i;
		sn->sc = sc;
		sn->status = FR_CHILD_INITIALIZING;
		fr_dlist_insert_head(&sc->networks, sn);

		if (fr_schedule_pthread_create(&sn->pthread_id, fr_schedule_network_thread, sn) < 0) {
			PERROR("Failed creating network %u", i);
			break;
		}
	}

	/*
	 *	Wait for all of the networks to signal us that either
	 *	they've started, OR there's been a problem and they
	 *	can't start.
	 */
	for (i = 0; i < (unsigned int)fr_dlist_num_elements(&sc->networks); i++) {
		DEBUG3("Waiting for semaphore from network %u/%u",
		       i + 1, (unsigned int)fr_dlist_num_elements(&sc->networks));
		SEM_WAIT_INTR(&sc->network_sem);
	}

	/*
	 *	See if all of the networks have started.
	 */
	for (sn = fr_dlist_head(&sc->networks);
	     sn != NULL;
	     sn = next_sn) {
		next_sn = fr_dlist_next(&sc->networks, sn);

		if (sn->status != FR_CHILD_RUNNING) {
			fr_dlist_remove(&sc->networks, sn);
			continue;
		}
	}

	/*
	 *	Failed to start some workers, refuse to do anything!
	 */
	if ((unsigned int)fr_dlist_num_elements(&sc->networks) < sc->config->max_networks) {
		fr_schedule_destroy(&sc);
		return NULL;
	}

	/*
	 *	Create all of the workers.
	 */
	for (i = 0; i < sc->config->max_workers; i++) {
		DEBUG3("Creating %u/%u workers", i + 1, sc->config->max_workers);

		/*
		 *	Create a worker "glue" structure
		 */
		sw = talloc_zero(sc, fr_schedule_worker_t);
		if (!sw) {
			ERROR("Worker %u - Failed allocating memory", i);
			break;
		}

		sw->id = i;
		sw->sc = sc;
		sw->status = FR_CHILD_INITIALIZING;
		fr_dlist_insert_head(&sc->workers, sw);

		if (fr_schedule_pthread_create(&sw->pthread_id, fr_schedule_worker_thread, sw) < 0) {
			PERROR("Failed creating worker %u", i);
			break;
		}
	}

	/*
	 *	Wait for all of the workers to signal us that either
	 *	they've started, OR there's been a problem and they
	 *	can't start.
	 */
	for (i = 0; i < (unsigned int)fr_dlist_num_elements(&sc->workers); i++) {
		DEBUG3("Waiting for semaphore from worker %u/%u",
		       i + 1, (unsigned int)fr_dlist_num_elements(&sc->workers));
		SEM_WAIT_INTR(&sc->worker_sem);
	}

	/*
	 *	See if all of the workers have started.
	 */
	for (sw = fr_dlist_head(&sc->workers);
	     sw != NULL;
	     sw = next_sw) {

		next_sw = fr_dlist_next(&sc->workers, sw);

		if (sw->status != FR_CHILD_RUNNING) {
			fr_dlist_remove(&sc->workers, sw);
			continue;
		}
	}

	/*
	 *	Failed to start some workers, refuse to do anything!
	 */
	if ((unsigned int)fr_dlist_num_elements(&sc->workers) < sc->config->max_workers) {
		fr_schedule_destroy(&sc);
		return NULL;
	}

	for (sw = fr_dlist_head(&sc->workers), i = 0;
	     sw != NULL;
	     sw = next_sw, i++) {
		char buffer[32];

		next_sw = fr_dlist_next(&sc->workers, sw);

		snprintf(buffer, sizeof(buffer), "%d", i);
		if (fr_command_register_hook(NULL, buffer, sw->worker, cmd_worker_table) < 0) {
			PERROR("Failed adding worker commands");
			goto st_fail;
		}
	}

	for (sn = fr_dlist_head(&sc->networks), i = 0;
	     sn != NULL;
	     sn = next_sn, i++) {
		char buffer[32];

		next_sn = fr_dlist_next(&sc->networks, sn);

		snprintf(buffer, sizeof(buffer), "%d", i);
		if (fr_command_register_hook(NULL, buffer, sn->nr, cmd_network_table) < 0) {
			PERROR("Failed adding network commands");
			goto st_fail;
		}
	}

	if (sc) INFO("Scheduler created successfully with %u networks and %u workers",
		     sc->config->max_networks, (unsigned int)fr_dlist_num_elements(&sc->workers));

	return sc;
}

/** Destroy a scheduler, and tell its child threads to exit.
 *
 * @note This may be called with no worker or network threads in the case of a
 *	 instantiation error.  This function _should_ deal with that condition
 *	 gracefully.
 *
 * @param[in] sc_to_free the scheduler
 * @return
 *	- <0 on error
 *	- 0 on success
 */
int fr_schedule_destroy(fr_schedule_t **sc_to_free)
{
	fr_schedule_t		*sc = *sc_to_free;
	unsigned int		i;
	fr_schedule_worker_t	*sw;
	fr_schedule_network_t	*sn;
	int			ret;

	if (!sc) return 0;

	sc->running = false;

	/*
	 *	Single threaded mode: kill the only network / worker we have.
	 */
	if (sc->el) {
		/*
		 *	Destroy the network side first.  It tells the
		 *	workers to close.
		 */
		fr_network_destroy(sc->single_network);
		fr_worker_destroy(sc->single_worker);
		goto done;
	}

	/*
	 *	Signal each network thread to exit.
	 */
	for (sn = fr_dlist_head(&sc->networks);
	     sn != NULL;
	     sn = fr_dlist_next(&sc->networks, sn)) {
		fr_network_exit(sn->nr);
	}

	/*
	 *	If the network threads are running, tell them to exit,
	 *	and wait for them to do so.  Each network thread tells
	 *	all of its worker threads that it's exiting.  It then
	 *	closes the channels.  When the workers see that there
	 *	are no input channels, they exit, too.
	 */
	for (i = 0; i < (unsigned int)fr_dlist_num_elements(&sc->networks); i++) {
		DEBUG2("Scheduler - Waiting for semaphore indicating network exit %u/%u", i + 1,
		       (unsigned int)fr_dlist_num_elements(&sc->networks));
		SEM_WAIT_INTR(&sc->network_sem);
	}
	DEBUG2("Scheduler - All networks indicated exit complete");

	while ((sn = fr_dlist_head(&sc->networks)) != NULL) {
		fr_dlist_remove(&sc->networks, sn);

		/*
		 *	Ensure that the thread has exited before
		 *	cleaning up the context.
		 *
		 *	This also ensures that the child threads have
		 *	exited before the main thread cleans up the
		 *	module instances.
		 */
		if ((ret = pthread_join(sn->pthread_id, NULL)) != 0) {
			ERROR("Failed joining network %i: %s", sn->id, fr_syserror(ret));
		} else {
			DEBUG2("Network %i joined (cleaned up)", sn->id);
		}
	}

	/*
	 *	Wait for all worker threads to finish.  THEN clean up
	 *	modules.  Otherwise, the modules will be removed from
	 *	underneath the workers!
	 */
	for (i = 0; i < (unsigned int)fr_dlist_num_elements(&sc->workers); i++) {
		DEBUG2("Scheduler - Waiting for semaphore indicating worker exit %u/%u", i + 1,
		       (unsigned int)fr_dlist_num_elements(&sc->workers));
		SEM_WAIT_INTR(&sc->worker_sem);
	}
	DEBUG2("Scheduler - All workers indicated exit complete");

	/*
	 *	Clean up the exited workers.
	 */
	while ((sw = fr_dlist_head(&sc->workers)) != NULL) {
		fr_dlist_remove(&sc->workers, sw);

		/*
		 *	Ensure that the thread has exited before
		 *	cleaning up the context.
		 *
		 *	This also ensures that the child threads have
		 *	exited before the main thread cleans up the
		 *	module instances.
		 */
		if ((ret = pthread_join(sw->pthread_id, NULL)) != 0) {
			ERROR("Failed joining worker %i: %s", sw->id, fr_syserror(ret));
		} else {
			DEBUG2("Worker %i joined (cleaned up)", sw->id);
		}
	}

	sem_destroy(&sc->network_sem);
	sem_destroy(&sc->worker_sem);
done:
	/*
	 *	Now that all of the workers are done, we can return to
	 *	the caller, and have it dlclose() the modules.
	 */
	talloc_free(sc);
	*sc_to_free = NULL;

	return 0;
}

/** Add a fr_listen_t to a scheduler.
 *
 * @param[in] sc the scheduler
 * @param[in] li the ctx and callbacks for the transport.
 * @return
 *	- NULL on error
 *	- the fr_network_t that the socket was added to.
 */
fr_network_t *fr_schedule_listen_add(fr_schedule_t *sc, fr_listen_t *li)
{
	fr_network_t *nr;

	(void) talloc_get_type_abort(sc, fr_schedule_t);

	if (sc->el) {
		nr = sc->single_network;
	} else {
		fr_schedule_network_t *sn;

		/*
		 *	@todo - round robin it among the listeners?
		 *	or maybe add it to the same parent thread?
		 */
		sn = fr_dlist_head(&sc->networks);
		nr = sn->nr;
	}

	if (fr_network_listen_add(nr, li) < 0) return NULL;

	return nr;
}

/** Add a directory NOTE_EXTEND to a scheduler.
 *
 * @param[in] sc the scheduler
 * @param[in] li the ctx and callbacks for the transport.
 * @return
 *	- NULL on error
 *	- the fr_network_t that the socket was added to.
 */
fr_network_t *fr_schedule_directory_add(fr_schedule_t *sc, fr_listen_t *li)
{
	fr_network_t *nr;

	(void) talloc_get_type_abort(sc, fr_schedule_t);

	if (sc->el) {
		nr = sc->single_network;
	} else {
		fr_schedule_network_t *sn;

		/*
		 *	@todo - round robin it among the listeners?
		 *	or maybe add it to the same parent thread?
		 */
		sn = fr_dlist_head(&sc->networks);
		nr = sn->nr;
	}

	if (fr_network_directory_add(nr, li) < 0) return NULL;

	return nr;
}
