/src/freeradius-server/src/lib/io/schedule.c

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

#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_coord_thread_join();

  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;
}

Coverage Report

Created: 2026-05-11 06:44