schedule.c

Go to the documentation of this file.

/*
 *   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: 35ea2bcfd8ff947c13a2b99033621ce0acccbc4c $
 *
 * @brief Network / worker thread scheduling
 * @file io/schedule.c
 *
 * @copyright 2016 Alan DeKok (aland@freeradius.org)
 */
RCSID("$Id: 35ea2bcfd8ff947c13a2b99033621ce0acccbc4c $")
 
#define LOG_DST sc->log
 
#include <freeradius-devel/autoconf.h>
 
#include <freeradius-devel/io/schedule.h>
#include <freeradius-devel/io/thread.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 <freeradius-devel/util/semaphore.h>
 
#include <pthread.h>
 
/** Scheduler specific information for worker threads
 *
 * Wraps a fr_worker_t, tracking additional information that
 * the scheduler uses.
 */
typedef struct {
        fr_thread_t     thread;                 //!< common thread structure - must be first!
 
        int             uses;                   //!< how many network threads are using it
        fr_time_t       cpu_time;               //!< how much CPU time this worker has used
 
        fr_schedule_t   *sc;                    //!< the scheduler we are running under
 
        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 {
        fr_thread_t     thread;                 //!< common thread structure - must be first!
 
        fr_schedule_t   *sc;                    //!< the scheduler we are running under
 
        fr_network_t    *nr;                    //!< the receive data structure
 
        fr_timer_t      *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.
        bool            single_threaded;        //!< true if running in 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
 
        fr_sem_t        *worker_sem;            //!< for inter-thread signaling
        fr_sem_t        *network_sem;           //!< for inter-thread signaling
        fr_sem_t        *coord_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 = -1;        //!< 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;
}
 
/** Explicitly set the worker id for the current thread
 *
 * **Only to be used in test programs like unit_test_module**
 */
void fr_schedule_worker_id_set(int id)
{
        worker_id = id;
}
 
/** Entry point for worker threads
 *
 * @param[in] arg       the fr_schedule_worker_t
 * @return NULL
 */
static void *fr_schedule_worker_thread(void *arg)
{
        fr_schedule_worker_t            *sw = talloc_get_type_abort(arg, fr_schedule_worker_t);
        fr_schedule_t                   *sc = sw->sc;
        fr_thread_status_t              status = FR_THREAD_FAIL;
        char                            worker_name[32];
 
        worker_id = sw->thread.id;              /* Store the current worker ID */
 
        snprintf(worker_name, sizeof(worker_name), "Worker %d", sw->thread.id);
 
#ifdef HAVE_PTHREAD_SETNAME_NP
#  ifdef __APPLE__
        pthread_setname_np(worker_name);
#  else
        pthread_setname_np(pthread_self(), worker_name);
#  endif
#endif
 
        if (fr_thread_setup(&sw->thread, worker_name) < 0) goto fail;
 
        sw->worker = fr_worker_alloc(sw->thread.ctx, sw->thread.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->thread.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->thread.ctx, sw->thread.el, cs) < 0) {
                        PERROR("%s - Worker thread instantiation failed", worker_name);
                        goto fail;
                }
        }
 
        /*
         *      Add this worker to all network threads.
         */
        fr_dlist_foreach(&sc->networks, fr_schedule_network_t, sn) {
                if (unlikely(fr_network_worker_add(sn->nr, sw->worker) < 0)) {
                        PERROR("%s - Failed adding worker to network %u", worker_name, sn->thread.id);
                        goto fail;      /* FIXME - Should maybe try to undo partial adds? */
                }
        }
 
        /*
         *      Tell the originator that the thread has started.
         */
        fr_thread_start(&sw->thread, sc->worker_sem);
 
        /*
         *      Do all of the work.
         */
        fr_worker(sw->worker);
 
        status = FR_THREAD_EXITED;
 
fail:
        if (sw->worker) {
                fr_worker_destroy(sw->worker);
                sw->worker = NULL;
        }
 
        if (sc->worker_thread_detach) sc->worker_thread_detach(NULL);   /* Fixme once we figure out what uctx should be */
 
        fr_thread_exit(&sw->thread, status, sc->worker_sem);
 
        return NULL;
}
 
 
static void stats_timer(fr_timer_list_t *tl, 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_timer_at(sn, tl, &sn->ev, fr_time_add(now, sn->sc->config->stats_interval), false, 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)
{
        fr_schedule_network_t           *sn = talloc_get_type_abort(arg, fr_schedule_network_t);
        fr_schedule_t                   *sc = sn->sc;
        fr_thread_status_t              status = FR_THREAD_FAIL;
        char                            network_name[32];
 
        snprintf(network_name, sizeof(network_name), "Network %d", sn->thread.id);
 
#ifdef HAVE_PTHREAD_SETNAME_NP
#  ifdef __APPLE__
        pthread_setname_np(network_name);
#  else
        pthread_setname_np(pthread_self(), network_name);
#  endif
#endif
 
        if (fr_thread_setup(&sn->thread, network_name) < 0) goto fail;
 
        sn->nr = fr_network_create(sn->thread.ctx, sn->thread.el, network_name, sc->log, sc->lvl, &sc->config->network);
        if (!sn->nr) {
                PERROR("%s - Failed creating network", network_name);
                goto fail;
        }
 
        /*
         *      Tell the originator that the thread has started.
         */
        fr_thread_start(&sn->thread, sc->network_sem);
 
        /*
         *      Print out statistics for this network IO handler.
         */
        if (fr_time_delta_ispos(sc->config->stats_interval)) {
                (void) fr_timer_in(sn, sn->thread.el->tl, &sn->ev, sn->sc->config->stats_interval, false, 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_THREAD_EXITED;
 
fail:
        fr_thread_exit(&sn->thread, status, sc->network_sem);
 
        return NULL;
}
 
/** Create a scheduler and spawn the child threads.
 *
 * @param[in] ctx                               talloc context.
 * @param[in] single_threaded                   no workers are spawned, everything runs in a common event loop.
 * @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,
                                  bool single_threaded,
                                  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;
        }
 
        /*
         *      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;
        }
 
        sc->el = el;
        sc->single_threaded = single_threaded;
        sc->log = logger;
        sc->lvl = lvl;
        sc->cs = sc->config->cs;
 
        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 (single_threaded) {
                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;
                }
 
                if (fr_coords_create(sc, el) < 0) {
                        PERROR("Failed creating coordinators");
                        if (unlikely(fr_network_destroy(sc->single_network) < 0)) {
                                PERROR("Failed destroying network");
                        }
                        goto pre_instantiate_st_fail;
                }
 
                worker_id = 0;
                sc->single_worker = fr_worker_alloc(sc, el, "Worker", sc->log, sc->lvl, &sc->config->worker);
                if (!sc->single_worker) {
                        PERROR("Failed creating worker");
                        if (unlikely(fr_network_destroy(sc->single_network) < 0)) {
                                PERROR("Failed destroying 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:
                                if (unlikely(fr_network_destroy(sc->single_network) < 0)) {
                                        PERROR("Failed destroying 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;
                }
 
                if (fr_coord_pre_event_insert(el) < 0) {
                        fr_strerror_const("Failed adding coordinator 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;
                }
 
                if (fr_coord_post_event_insert(el) < 0) {
                        fr_strerror_const("Failed adding coordinator post-processing to event list");
                        goto st_fail;
                }
 
                return sc;
        }
 
        /*
         *      Create the lists which hold the workers and networks.
         */
        fr_dlist_init(&sc->workers, fr_schedule_worker_t, thread.entry);
        fr_dlist_init(&sc->networks, fr_schedule_network_t, thread.entry);
 
        sc->network_sem = fr_sem_alloc();
        if (!sc->network_sem) {
        sem_fail:
                ERROR("Failed creating semaphore: %s", fr_syserror(errno));
                fr_sem_free(sc->network_sem);
                fr_sem_free(sc->worker_sem);
                talloc_free(sc);
                return NULL;
        }
 
        sc->worker_sem = fr_sem_alloc();
        if (!sc->worker_sem) goto sem_fail;
 
        sc->coord_sem = fr_sem_alloc();
        if (!sc->coord_sem) goto sem_fail;
 
        /*
         *      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->thread.id = i;
                sn->sc = sc;
                sn->thread.status = FR_THREAD_INITIALIZING;
 
                if (fr_thread_create(&sn->thread.pthread_id, fr_schedule_network_thread, sn) < 0) {
                        talloc_free(sn);
                        PERROR("Failed creating network %u", i);
                        break;
                }
 
                fr_dlist_insert_head(&sc->networks, sn);
        }
 
        /*
         *      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.
         */
        if (fr_thread_wait_list(sc->network_sem, &sc->networks) < 0) {
                fr_schedule_destroy(&sc);
                return NULL;
        }
 
        /*
         *      Create the coordination threads
         */
        if (fr_coord_start(sc->config->max_workers, sc->coord_sem) < 0) {
                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->thread.id = i;
                sw->sc = sc;
                sw->thread.status = FR_THREAD_INITIALIZING;
 
                if (fr_thread_create(&sw->thread.pthread_id, fr_schedule_worker_thread, sw) < 0) {
                        talloc_free(sw);
                        PERROR("Failed creating worker %u", i);
                        break;
                }
 
                fr_dlist_insert_head(&sc->workers, sw);
        }
 
        /*
         *      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.
         */
        if (fr_thread_wait_list(sc->worker_sem, &sc->workers) < 0) {
                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");
                mt_fail:
                        fr_schedule_destroy(&sc);
                        return NULL;
                }
        }
 
        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 mt_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));
 
        /*
         *      Instantiate thread-local data for the main thread too.
         *      In single-threaded mode this is done above.  In
         *      multi-worker mode the main thread also needs module
         *      thread data so that triggers can use module xlats.
         */
        if (sc->worker_thread_instantiate &&
            unlikely((sc->worker_thread_instantiate(sc, el, NULL) < 0))) {
                PERROR("Main thread instantiation failed");
                goto mt_fail;
        }
 
        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->single_threaded) {
                /*
                 *      Destroy the network side first.  It tells the
                 *      workers to close.
                 */
                if (unlikely(fr_network_destroy(sc->single_network) < 0)) {
                        ERROR("Failed destroying network");
                }
                fr_worker_destroy(sc->single_worker);
                fr_coords_destroy();
 
                goto done;
        } else {
                /*
                 *      Detach thread-local data for the main thread.
                 *      Worker threads handle their own detach, but
                 *      the main thread was instantiated explicitly
                 *      by fr_schedule_create.
                 */
                if (sc->worker_thread_detach) sc->worker_thread_detach(NULL);
        }
 
        /*
         *      Signal each network thread to exit.
         */
        fr_dlist_foreach(&sc->networks, fr_schedule_network_t, sne) {
                if (fr_network_exit(sne->nr) < 0) {
                        PERROR("Failed signaling network %i to exit", sne->thread.id);
                }
        }
 
        /*
         *      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_pop_head(&sc->networks)) != NULL) {
                /*
                 *      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->thread.pthread_id, NULL)) != 0) {
                        ERROR("Failed joining network %i: %s", sn->thread.id, fr_syserror(ret));
                } else {
                        DEBUG2("Network %i joined (cleaned up)", sn->thread.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_pop_head(&sc->workers)) != NULL) {
                /*
                 *      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->thread.pthread_id, NULL)) != 0) {
                        ERROR("Failed joining worker %i: %s", sw->thread.id, fr_syserror(ret));
                } else {
                        DEBUG2("Worker %i joined (cleaned up)", sw->thread.id);
                }
        }
 
        fr_sem_free(sc->coord_sem);
        fr_sem_free(sc->network_sem);
        fr_sem_free(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->single_threaded) {
                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->single_threaded) {
                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;
}