scheduler.c

Go to the documentation of this file.

/*
 SPDX-FileCopyrightText: © 2010, 2011, 2012 Hewlett-Packard Development Company, L.P.
 
 SPDX-License-Identifier: GPL-2.0-only
*/
/* local includes */
#include <libfossrepo.h>
#include <agent.h>
#include <database.h>
#include <event.h>
#include <host.h>
#include <interface.h>
#include <scheduler.h>
#include <fossconfig.h>
 
/* std library includes */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
 
/* unix system includes */
#include <dirent.h>
#include <fcntl.h>
#include <signal.h>
#include <sys/mman.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#include <pwd.h>
#include <grp.h>
 
/* glib includes */
#include <glib.h>
#include <gio/gio.h>
 
#define TEST_ERROR(error, ...)                                     \
  if(error)                                                        \
  {                                                                \
    log_printf("ERROR %s.%d: %s\n",                                \
      __FILE__, __LINE__, error->message);                         \
    log_printf("ERROR %s.%d: ", __FILE__, __LINE__);               \
    log_printf(__VA_ARGS__);                                       \
    log_printf("\n");                                              \
    g_clear_error(&error);                                         \
    continue;                                                      \
  }
 
/* global flags */
int verbose = 0;        
int closing = 0;        
 
GThread* main_thread;   
 
#define SELECT_DECLS(type, name, l_op, w_op, val) type CONF_##name = val;
CONF_VARIABLES_TYPES(SELECT_DECLS)
#undef SELECT_DECLS
 
/* ************************************************************************** */
/* **** signals and events ************************************************** */
/* ************************************************************************** */
 
#define MASK_SIGCHLD (1 << 0)
#define MASK_SIGALRM (1 << 1)
#define MASK_SIGTERM (1 << 2)
#define MASK_SIGQUIT (1 << 3)
#define MASK_SIGHUP  (1 << 4)
 
int sigmask = 0;
 
void scheduler_sig_handle(int signo)
{
  /* Anywhere you see a "#if __GNUC__" the code is checking if GCC is the
   * compiler. This is because the __sync... set of functions are the GCC
   * version of atomics.
   *
   * This means that if you aren't compiling with GCC, you can have a race
   * condition that results in a signal being lost during the
   * signal_scheduler() function.
   *
   * What could happen:
   *   1. signal_scheduler() reads value of sigmask
   *   2. scheduler receives a SIG**** and sets the correct bit in sigmask
   *   3. signal_scheduler() clears sigmask by setting it to 0
   *
   * In this set of events, a signal has been lost. If this is a sigchld this
   * could be very bad as a job could never get marked as finished.
   */
  switch(signo)
  {
#if __GNUC__
    case SIGCHLD: __sync_fetch_and_or(&sigmask, MASK_SIGCHLD); break;
    case SIGTERM: __sync_fetch_and_or(&sigmask, MASK_SIGTERM); break;
    case SIGQUIT: __sync_fetch_and_or(&sigmask, MASK_SIGQUIT); break;
    case SIGHUP:  __sync_fetch_and_or(&sigmask, MASK_SIGHUP);  break;
#else
    case SIGCHLD: sigmask |= MASK_SIGCHLD; break;
    case SIGALRM: sigmask |= MASK_SIGALRM; break;
    case SIGTERM: sigmask |= MASK_SIGTERM; break;
    case SIGQUIT: sigmask |= MASK_SIGQUIT; break;
    case SIGHUP:  sigmask |= MASK_SIGHUP ; break;
#endif
  }
}
 
void scheduler_signal(scheduler_t* scheduler)
{
  // the last time an update was run
  static time_t last_update = 0;
 
  // copy of the mask
  guint mask;
 
  /* this will get sigmask and set it to 0 */
#if __GNUC__
  mask = __sync_fetch_and_and(&sigmask, 0);
#else
  mask = sigmask;
  sigmask = 0;
#endif
 
  /* initialize last_update */
  if(last_update == 0)
    last_update = time(NULL);
 
  /* signal: SIGCHLD
   *
   * A SIGCHLD has been received since the last time signal_scheduler() was
   * called. Get all agents that have finished since last this happened and
   * create an event for each.
   */
  if(mask & MASK_SIGCHLD)
  {
    pid_t n;          // the next pid that has died
    pid_t* pass;
    int status;       // status returned by waitpit()
 
    /* get all of the dead children's pids */
    while((n = waitpid(-1, &status, WNOHANG)) > 0)
    {
      V_SCHED("SIGNALS: received sigchld for pid %d\n", n);
      pass = g_new0(pid_t, 3);  /* [0]=pid [1]=status [2]=retry */
      pass[0] = n;
      pass[1] = status;
      event_signal(agent_death_event, pass);
    }
  }
 
  /* signal: SIGTERM
   *
   * A SIGTERM has been received. simply set the closing flag to 1 so that the
   * scheduler will gracefully shutdown as all the agents finish running.
   */
  if(mask & MASK_SIGTERM)
  {
    V_SCHED("SIGNALS: Scheduler received terminate signal, shutting down gracefully\n");
    event_signal(scheduler_close_event, (void*)0);
  }
 
  /* signal: SIGQUIT
   *
   * A SIGQUIT has been received. Queue a scheduler_close_event so that the
   * scheduler will immediately stop running. This will cause all the agents to
   * be forcefully killed.
   */
  if(mask & MASK_SIGQUIT)
  {
    V_SCHED("SIGNALS: Scheduler received quit signal, shutting down scheduler\n");
    event_signal(scheduler_close_event, (void*)1);
  }
 
  /* signal: SIGHUP
   *
   * A SIGHUP has been received. reload the configuration files for the
   * scheduler. This will run here instead of being queued as an event.
   */
  if(mask & MASK_SIGHUP)
  {
    V_SCHED("SIGNALS: Scheduler received SGIHUP, reloading configuration data\n");
    scheduler_config_event(scheduler, NULL);
  }
 
  /* Finish by checking if an agent update needs to be performed.
   *
   * Every CONF_agent_update_interval, the agents and database should be
   * updated. The agents need to be updated to check for dead and unresponsive
   * agents. The database is updated to make sure that a new job hasn't been
   * scheduled without the scheduler being informed.
   */
  if((time(NULL) - last_update) > CONF_agent_update_interval )
  {
    V_SPECIAL("SIGNALS: Performing agent and database update\n");
    event_signal(agent_update_event, NULL);
    event_signal(database_update_event, NULL);
    last_update = time(NULL);
  }
}
 
/* ************************************************************************** */
/* **** The actual scheduler ************************************************ */
/* ************************************************************************** */
 
scheduler_t* scheduler_init(gchar* sysconfigdir, log_t* log)
{
  scheduler_t* ret = g_new0(scheduler_t, 1);
 
  ret->process_name  = NULL;
  ret->s_pid         = getpid();
  ret->s_daemon      = FALSE;
  ret->s_startup     = FALSE;
  ret->s_pause       = TRUE;
 
  ret->sysconfig     = NULL;
  ret->sysconfigdir  = g_strdup(sysconfigdir);
  ret->logdir        = LOG_DIR;
  ret->logcmdline    = FALSE;
  ret->main_log      = log;
  ret->host_queue    = NULL;
 
  ret->i_created     = FALSE;
  ret->i_terminate   = FALSE;
  ret->i_port        = 0;
  ret->server        = NULL;
  ret->workers       = NULL;
  ret->cancel        = NULL;
 
  ret->scheduler_version = NULL;
 
  ret->job_queue     = g_sequence_new(NULL);
 
  ret->db_conn       = NULL;
  ret->host_url      = NULL;
  ret->email_subject = NULL;
  ret->email_header  = NULL;
  ret->email_footer  = NULL;
  ret->email_command = NULL;
 
  /* This regex should find:
   *   1. One or more capital letters followed by a ':' followed by white space,
   *      followed by a number
   *   2. One or more capital letters followed by a ':' followed by white space,
   *      followed by a number, followed by white space, followed by a number
   *
   * Examples:
   *   HEART: 1 2   -> matches
   *   HEART: 1     -> matches
   *   HEART:       -> does not match
   *
   */
  ret->parse_agent_msg = g_regex_new(
      "([A-Z]+):([ \t]+)(\\d+)(([ \t]+)(\\d))?",
      0, 0, NULL);
 
  /* This regex should find:
   *   1. A '$' followed by any combination of capital letters or underscore
   *   2. A '$' followed by any combination of capital letters or underscore,
   *      followed by a '.' followed by alphabetic characters or underscore,
   *      followed by a '.' followed by alphabetic characters or underscore
   *
   * Examples:
   *   $HELLO             -> matches
   *   $SIMPLE_NAME       -> matches
   *   $DB.table.column   -> matches
   *   $bad               -> does not match
   *   $DB.table          -> does not match
   */
  ret->parse_db_email      = g_regex_new(
      "\\$([A-Z_]*)(\\.([a-zA-Z_]*)\\.([a-zA-Z_]*))?",
      0, 0, NULL);
 
  /* This regex should match:
   *   1. a set of alphabetical characters
   *   2. a set of alphabetical characters, followed by white space, followed by
   *      a number
   *   3. a set of alphabetical characters, followed by white space, followed by
   *      a number, followed by white space, followed by a string in quotes.
   *
   *
   * Examples:
   *   close                   -> matches
   *   stop                    -> matches
   *   pause 10                -> matches
   *   kill 10 "hello world"   -> matches
   *   pause 10 10             -> does not match
   *   kill "hello world" 10   -> does not match
   *
   *
   */
  ret->parse_interface_cmd = g_regex_new(
      "(\\w+)(\\s+(-?\\d+))?(\\s+((-?\\d+)|(\"(.*)\")))?",
      0, G_REGEX_MATCH_NEWLINE_LF, NULL);
 
  ret->meta_agents = g_tree_new_full(string_compare, NULL, NULL,
      (GDestroyNotify)meta_agent_destroy);
  ret->agents      = g_tree_new_full(int_compare,    NULL, NULL,
      (GDestroyNotify)agent_destroy);
  ret->host_list = g_tree_new_full(string_compare, NULL, NULL,
      (GDestroyNotify)host_destroy);
  ret->job_list     = g_tree_new_full(int_compare, NULL, NULL,
      (GDestroyNotify)job_destroy);
 
  main_log = log;
 
  return ret;
}
 
void scheduler_destroy(scheduler_t* scheduler)
{
 
  event_loop_destroy();
 
  if(scheduler->main_log)
  {
    log_destroy(scheduler->main_log);
    main_log = NULL;
  }
 
  if(scheduler->scheduler_version) g_free(scheduler->scheduler_version);
  if(scheduler->process_name) g_free(scheduler->process_name);
  if(scheduler->sysconfig)    fo_config_free(scheduler->sysconfig);
  if(scheduler->sysconfigdir) g_free(scheduler->sysconfigdir);
  if(scheduler->host_queue)   g_list_free(scheduler->host_queue);
  if(scheduler->workers)      g_thread_pool_free(scheduler->workers, FALSE, TRUE);
 
  if(scheduler->email_subject) g_free(scheduler->email_subject);
  if(scheduler->email_command) g_free(scheduler->email_command);
 
  g_sequence_free(scheduler->job_queue);
 
  g_regex_unref(scheduler->parse_agent_msg);
  g_regex_unref(scheduler->parse_db_email);
  g_regex_unref(scheduler->parse_interface_cmd);
 
  g_tree_unref(scheduler->meta_agents);
  g_tree_unref(scheduler->agents);
  g_tree_unref(scheduler->host_list);
  g_tree_unref(scheduler->job_list);
 
  if (scheduler->db_conn) PQfinish(scheduler->db_conn);
 
  g_free(scheduler);
}
 
static gboolean isMaxLimitReached(meta_agent_t* agent)
{
  // max_run < 0 means unlimited; only enforce a non-negative limit.
  if (agent != NULL && agent->max_run >= 0 && agent->max_run <= agent->run_count)
  {
    return TRUE;
  }
  return FALSE;
}
 
/* ************************************************************************** */
/* **** Stale CHECKEDOUT job reaper ***************************************** */
/* ************************************************************************** */
 
typedef struct {
  GList*  list; 
  time_t  now;  
} stale_ctx;
 
static gboolean collect_stale_jobs(gpointer key, gpointer val, gpointer data)
{
  job_t* j = (job_t*)val;
  stale_ctx* ctx = (stale_ctx*)data;
 
  // Skip non-CHECKEDOUT jobs and internal startup test jobs (id < 0).
  if(j == NULL || j->status != JB_CHECKEDOUT || j->id < 0)
  {
    return FALSE;
  }
 
  // Not stale if it still has agents assigned.
  if(j->running_agents  != NULL ||
     j->finished_agents != NULL ||
     j->failed_agents   != NULL)
  {
    return FALSE;
  }
 
  // Grace period: only flag as stale after CONF_agent_update_interval seconds beyond checkout time.
  if((ctx->now - j->checkedout_at) <= CONF_agent_update_interval)
  {
    return FALSE;
  }
 
  ctx->list = g_list_prepend(ctx->list, j);
  return FALSE;
}
 
static void reap_stale_jobs(scheduler_t* scheduler)
{
  static time_t last_stale_check = 0;
  time_t now = time(NULL);
 
  // Never reap during shutdown: CHECKEDOUT-with-no-agents jobs are legitimately in-flight, not stale.
  if(closing)
  {
    return;
  }
 
  // Rate-limit: bail immediately if the check interval hasn't elapsed.
  if((now - last_stale_check) < CONF_agent_update_interval)
  {
    return;
  }
 
  last_stale_check = now;
 
  // Skip the tree walk entirely when there are no jobs.
  if(g_tree_nnodes(scheduler->job_list) == 0)
  {
    return;
  }
 
  stale_ctx ctx = { NULL, now };
  g_tree_foreach(scheduler->job_list, collect_stale_jobs, &ctx);
 
  // Nothing stale - common case, exit fast.
  if(ctx.list == NULL)
  {
    return;
  }
 
  for(GList* iter = ctx.list; iter != NULL; iter = iter->next)
  {
    job_t* j = (job_t*)iter->data;
 
    log_printf("WARNING %s.%d: JOB[%d] type:%s is CHECKEDOUT with no agents"
               " - failing as stale\n",
               __FILE__, __LINE__, j->id, j->agent_type);
 
    g_free(j->message);
    j->message = g_strdup("job checked out but no agent was ever started (stale)");
    job_fail_event(scheduler, j);
 
    // Remove from job_queue so the scheduler doesn't try to assign a fresh agent to this FAILED job.
    {
      GSequenceIter* sit = g_sequence_get_begin_iter(scheduler->job_queue);
      while(!g_sequence_iter_is_end(sit))
      {
        if(g_sequence_get(sit) == j)
        {
          g_sequence_remove(sit);
          break;
        }
        sit = g_sequence_iter_next(sit);
      }
    }
 
    /* no agents were spawned, so this removes the job from job_list and frees it */
    job_remove_agent(j, scheduler->job_list, NULL);
  }
 
  g_list_free(ctx.list);
}
 
void scheduler_update(scheduler_t* scheduler)
{
  /* queue used to hold jobs if an exclusive job enters the system */
  static job_t*  job  = NULL;
  static host_t* host = NULL;
  static int lockout = 0;
 
  /* locals */
  int n_agents = g_tree_nnodes(scheduler->agents);
  int n_jobs   = active_jobs(scheduler->job_list);
 
  /* check to see if we are in and can exit the startup state */
  if(scheduler->s_startup && n_agents == 0)
  {
    event_signal(database_update_event, NULL);
    scheduler->s_startup = 0;
  }
 
  /* check if we are able to close the scheduler */
  if(closing && n_agents == 0 && n_jobs == 0)
  {
    event_loop_terminate();
    return;
  }
 
  if(lockout && n_agents == 0 && n_jobs == 0)
    lockout = 0;
 
  if(job == NULL && !lockout)
  {
    /* blocked jobs are set aside in skipped_jobs and re-inserted (sorted) after
     * the loop, so a blocked high-priority job doesn't starve the rest */
    GList* skipped_jobs = NULL;
 
    while((job = peek_job(scheduler->job_queue)) != NULL)
    {
      // Check the max limit of running agents
      if (isMaxLimitReached(
          g_tree_lookup(scheduler->meta_agents, job->agent_type)))
      {
        V_SCHED("JOB_INIT: Unable to run agent %s due to max_run limit.\n",
            job->agent_type);
        // max_run reached: reset stale timer and defer to next cycle.
        job->checkedout_at = time(NULL);
        next_job(scheduler->job_queue);
        skipped_jobs = g_list_prepend(skipped_jobs, job);
        job = NULL;
        continue;
      }
      // check if the agent is required to run on local host
      if(is_meta_special(
          g_tree_lookup(scheduler->meta_agents, job->agent_type), SAG_LOCAL))
      {
        host = g_tree_lookup(scheduler->host_list, LOCAL_HOST);
        if(!(host->running < host->max))
        {
          // Local host full: reset stale timer and defer to next cycle.
          job->checkedout_at = time(NULL);
          next_job(scheduler->job_queue);
          skipped_jobs = g_list_prepend(skipped_jobs, job);
          job = NULL;
          continue;
        }
      }
      // check if the job is required to run on a specific machine
      else if((job->required_host != NULL))
      {
        host = g_tree_lookup(scheduler->host_list, job->required_host);
        if(host != NULL)
        {
          if(!(host->running < host->max))
          {
            // Required host full: reset stale timer and defer to next cycle.
            job->checkedout_at = time(NULL);
            next_job(scheduler->job_queue);
            skipped_jobs = g_list_prepend(skipped_jobs, job);
            job = NULL;
            continue;
          }
        } else {
          g_free(job->message);
          job->message = g_strdup("ERROR: jq_host not in the agent list!");
          job_fail_event(scheduler, job);
          // Pop from queue first so job is not in queue when job_remove_agent frees it.
          next_job(scheduler->job_queue);
          job_remove_agent(job, scheduler->job_list, NULL);
          job = NULL;
          continue;
        }
      }
      // the generic case, this can run anywhere, find a place
      else if((host = get_host(&(scheduler->host_queue), 1)) == NULL)
      {
        // No host available: refresh stale timer on every queued job so reap_stale_jobs()
        // doesn't misidentify legitimately blocked jobs as orphans.
        time_t now = time(NULL);
        GSequenceIter* sit = g_sequence_get_begin_iter(scheduler->job_queue);
        while(!g_sequence_iter_is_end(sit))
        {
          ((job_t*)g_sequence_get(sit))->checkedout_at = now;
          sit = g_sequence_iter_next(sit);
        }
        job = NULL;
        break;
      }
 
      next_job(scheduler->job_queue);
      // Reset stale timer: job is now actively being dispatched.
      job->checkedout_at = time(NULL);
      if(is_meta_special(
          g_tree_lookup(scheduler->meta_agents, job->agent_type), SAG_EXCLUSIVE))
      {
        V_SCHED("JOB_INIT: exclusive, postponing initialization\n");
        break;
      }
 
      V_SCHED("Starting JOB[%d].%s\n", job->id, job->agent_type);
      agent_init(scheduler, host, job);
      job = NULL;
    }
 
    // Restore skipped jobs to the queue in priority order.
    for(GList* sl = skipped_jobs; sl != NULL; sl = sl->next)
    {
      g_sequence_insert_sorted(scheduler->job_queue, sl->data, job_compare, NULL);
    }
    g_list_free(skipped_jobs);
  }
 
  if(job != NULL && n_agents == 0 && n_jobs == 0)
  {
    agent_init(scheduler, host, job);
    lockout = 1;
    job  = NULL;
    host = NULL;
  }
 
  // Exclusive job is held in the static `job` pointer; reset its stale timer so the reaper doesn't free it.
  if(job != NULL)
  {
    job->checkedout_at = time(NULL);
  }
 
  reap_stale_jobs(scheduler);
 
  if(scheduler->s_pause)
  {
    scheduler->s_startup = 1;
    scheduler->s_pause = 0;
  }
}
 
/* ************************************************************************** */
/* **** main utility functions ********************************************** */
/* ************************************************************************** */
 
#define GU_HEADER "DIRECTORIES"
#define GU_GROUP  "PROJECTGROUP"
#define GU_USER   "PROJECTUSER"
 
void set_usr_grp(gchar* process_name, fo_conf* config)
{
  /* locals */
  struct group*  grp;
  struct passwd* pwd;
 
  char* group =
      fo_config_has_key(config, GU_HEADER, GU_GROUP) ?
      fo_config_get    (config, GU_HEADER, GU_GROUP, NULL) : PROJECT_GROUP;
  char* user  =
      fo_config_has_key(config, GU_HEADER, GU_USER)  ?
      fo_config_get    (config, GU_HEADER, GU_USER, NULL)  : PROJECT_USER;
 
  /* make sure group exists */
  grp = getgrnam(group);
  if(!grp)
  {
    fprintf(stderr, "FATAL %s.%d: could not find group \"%s\"\n",
        __FILE__, __LINE__, group);
    fprintf(stderr, "FATAL set_usr_grp() aborting due to error: %s\n",
        strerror(errno));
    exit(-1);
  }
 
  /* set the project group */
  setgroups(1, &(grp->gr_gid));
  if((setgid(grp->gr_gid) != 0) || (setegid(grp->gr_gid) != 0))
  {
    fprintf(stderr, "FATAL %s.%d: %s must be run as root or %s\n",
        __FILE__, __LINE__, process_name, user);
    fprintf(stderr, "FATAL Set group '%s' aborting due to error: %s\n",
        group, strerror(errno));
    exit(-1);
  }
 
  /* run as project user */
  pwd = getpwnam(user);
  if(!pwd)
  {
    fprintf(stderr, "FATAL %s.%d: user '%s' not found\n",
        __FILE__, __LINE__, user);
    exit(-1);
  }
 
  /* run as correct user, not as root or any other user */
  if((setuid(pwd->pw_uid) != 0) || (seteuid(pwd->pw_uid) != 0))
  {
    fprintf(stderr, "FATAL %s.%d: %s must run this as %s\n",
        __FILE__, __LINE__, process_name, user);
    fprintf(stderr, "FATAL SETUID aborting due to error: %s\n",
        strerror(errno));
    exit(-1);
  }
}
 
int kill_scheduler(int force)
{
  gchar f_name[FILENAME_MAX];
  struct dirent* ep;
  DIR* dp;
  FILE* file;
  gint num_killed = 0;
  pid_t s_pid = getpid();
 
  if((dp = opendir("/proc/")) == NULL)
  {
    fprintf(stderr, "ERROR %s.%d: Could not open /proc/ file system\n",
        __FILE__, __LINE__);
    exit(-1);
  }
 
  while((ep = readdir(dp)) != NULL)
  {
    if(string_is_num(ep->d_name))
    {
      snprintf(f_name, sizeof(f_name), "/proc/%s/cmdline", ep->d_name);
      if((file = fopen(f_name, "rt")))
      {
        if(fgets(f_name, sizeof(f_name), file) != NULL &&
            strstr(f_name, "fo_scheduler") && s_pid != atoi(ep->d_name))
        {
          NOTIFY("KILL: send signal to process %s\n", ep->d_name);
          if(force)
            kill(atoi(ep->d_name), SIGQUIT);
          else
            kill(atoi(ep->d_name), SIGTERM);
          num_killed++;
        }
 
        fclose(file);
      }
    }
  }
 
  closedir(dp);
 
  if(num_killed == 0)
    return -1;
  return 0;
}
 
void scheduler_clear_config(scheduler_t* scheduler)
{
  g_tree_clear(scheduler->meta_agents);
  g_tree_clear(scheduler->host_list);
 
  g_list_free(scheduler->host_queue);
  scheduler->host_queue = NULL;
 
  g_free(scheduler->host_url);
  g_free(scheduler->email_subject);
  g_free(scheduler->email_command);
  PQfinish(scheduler->db_conn);
  scheduler->db_conn       = NULL;
  scheduler->host_url      = NULL;
  scheduler->email_subject = NULL;
  scheduler->email_command = NULL;
 
  if(scheduler->default_header)
    munmap(scheduler->email_header, strlen(scheduler->email_header));
  if(scheduler->default_footer)
    munmap(scheduler->email_footer, strlen(scheduler->email_footer));
  scheduler->email_header  = NULL;
  scheduler->email_footer  = NULL;
 
  fo_config_free(scheduler->sysconfig);
  scheduler->sysconfig = NULL;
 
  /* keep scheduler_version: scheduler_foss_config() compares it against the
   * re-read COMMIT_HASH to detect a version change */
}
 
static gboolean g_tree_collect(gpointer key, gpointer value, gpointer data)
{
  GList** ret = (GList**)data;
 
  *ret = g_list_append(*ret, key);
 
  return 0;
}
 
void g_tree_clear(GTree* tree)
{
  GList* keys = NULL;
  GList* iter = NULL;
 
  g_tree_foreach(tree, g_tree_collect, &keys);
 
  for(iter = keys; iter != NULL; iter = iter->next)
    g_tree_remove(tree, iter->data);
 
  g_list_free(keys);
}
 
void scheduler_agent_config(scheduler_t* scheduler)
{
  DIR* dp;                  // directory pointer used to load meta agents;
  struct dirent* ep;        // information about directory
  gchar* dirname;           // holds the name of the current configuration file
  uint8_t max = -1;         // the number of agents to a host or number of one type running
  uint32_t special = 0;     // anything that is special about the agent (EXCLUSIVE)
  int32_t i;
  gchar* name;
  gchar* cmd;
  gchar* tmp;
  GError* error = NULL;
  fo_conf* config;
 
  dirname = g_strdup_printf("%s/%s/", scheduler->sysconfigdir, AGENT_CONF);
  if((dp = opendir(dirname)) == NULL)
  {
    FATAL("Could not open agent config directory: %s", dirname);
    return;
  }
  g_free(dirname);
 
  /* load the configuration for the agents */
  while((ep = readdir(dp)) != NULL)
  {
    if(ep->d_name[0] != '.')
    {
      dirname = g_strdup_printf("%s/%s/%s/%s.conf",
          scheduler->sysconfigdir, AGENT_CONF, ep->d_name, ep->d_name);
 
      config = fo_config_load(dirname, &error);
      if(error && error->code == fo_missing_file)
      {
        V_SCHED("CONFIG: Could not find %s\n", dirname);
        g_clear_error(&error);
        continue;
      }
      TEST_ERROR(error, "no additional info");
      V_SCHED("CONFIG: loading config file %s\n", dirname);
 
      if(!fo_config_has_group(config, "default"))
      {
        log_printf("ERROR: %s must have a \"default\" group\n", dirname);
        log_printf("ERROR: cause by %s.%d\n", __FILE__, __LINE__);
        continue;
      }
 
      special = 0;
      name = ep->d_name;
      max = fo_config_list_length(config, "default", "special", &error);
      TEST_ERROR(error, "%s: the special key should be of type list", dirname);
      for(i = 0; i < max; i++)
      {
        cmd = fo_config_get_list(config, "default", "special", i, &error);
        TEST_ERROR(error, "%s: failed to load element %d of special list",
            dirname, i)
 
        if(cmd[0] != '\0') {
          if(strncmp(cmd, "EXCLUSIVE", 9) == 0)
            special |= SAG_EXCLUSIVE;
          else if(strncmp(cmd, "NOEMAIL", 7) == 0)
            special |= SAG_NOEMAIL;
          else if(strncmp(cmd, "NOKILL", 6) == 0)
            special |= SAG_NOKILL;
          else if(strncmp(cmd, "LOCAL", 6) == 0)
            special |= SAG_LOCAL;
          else if(strlen(cmd) != 0)
            WARNING("%s: Invalid special type for agent %s: %s",
                dirname, name, cmd);
        }
      }
 
      cmd  = fo_config_get(config, "default", "command", &error);
      TEST_ERROR(error, "%s: the default group must have a command key", dirname);
      tmp  = fo_config_get(config, "default", "max", &error);
      TEST_ERROR(error, "%s: the default group must have a max key", dirname);
 
      if(!add_meta_agent(scheduler->meta_agents, name, cmd, (max = atoi(tmp)), special))
      {
        V_SCHED("CONFIG: could not create meta agent using %s\n", ep->d_name);
      }
      else if(TVERB_SCHED)
      {
        log_printf("CONFIG: added new agent\n");
        log_printf("    name = %s\n", name);
        log_printf(" command = %s\n", cmd);
        log_printf("     max = %d\n", max);
        log_printf(" special = %d\n", special);
      }
 
      g_free(dirname);
      fo_config_free(config);
    }
  }
 
  closedir(dp);
  event_signal(scheduler_test_agents, NULL);
}
 
void scheduler_foss_config(scheduler_t* scheduler)
{
  gchar*   tmp;                   // pointer into a string
  gchar**  keys;                  // list of host names grabbed from the config file
  int32_t  max = -1;              // the number of agents to a host or number of one type running
  int32_t  special = 0;           // anything that is special about the agent (EXCLUSIVE)
  gchar    addbuf[512];           // standard string buffer
  gchar    dirbuf[FILENAME_MAX];  // standard string buffer
  GError*  error = NULL;          // error return location
  int32_t  i;                     // indexing variable
  host_t*  host;                  // new hosts will be created in the loop
  fo_conf* version;               // information loaded from the version file
 
  if(scheduler->sysconfig != NULL)
    fo_config_free(scheduler->sysconfig);
 
  /* parse the config file */
  tmp = g_strdup_printf("%s/fossology.conf", scheduler->sysconfigdir);
  scheduler->sysconfig = fo_config_load(tmp, &error);
  if(error) FATAL("%s", error->message);
  g_free(tmp);
 
  /* set the user and group before proceeding */
  set_usr_grp(scheduler->process_name, scheduler->sysconfig);
 
  /* load the port setting */
  if(scheduler->i_port == 0)
    scheduler->i_port = atoi(fo_config_get(scheduler->sysconfig,
        "FOSSOLOGY", "port", &error));
 
  /* load the log directory */
  if(!scheduler->logcmdline)
  {
    if(fo_config_has_key(scheduler->sysconfig, "DIRECTORIES", "LOGDIR"))
      scheduler->logdir = fo_config_get(scheduler->sysconfig, "DIRECTORIES", "LOGDIR", &error);
    scheduler->main_log = log_new(scheduler->logdir, NULL, scheduler->s_pid);
 
    if(main_log)
    {
      log_destroy(main_log);
      main_log = scheduler->main_log;
    }
  }
 
  /* load the host settings */
  keys = fo_config_key_set(scheduler->sysconfig, "HOSTS", &special);
  for(i = 0; i < special; i++)
  {
    tmp = fo_config_get(scheduler->sysconfig, "HOSTS", keys[i], &error);
    if(error)
    {
      WARNING("%s\n", error->message);
      g_clear_error(&error);
      continue;
    }
 
    sscanf(tmp, "%s %s %d", addbuf, dirbuf, &max);
    host = host_init(keys[i], addbuf, dirbuf, max);
    host_insert(host, scheduler);
    if(TVERB_SCHED)
    {
      log_printf("CONFIG: added new host\n");
      log_printf("      name = %s\n", keys[i]);
      log_printf("   address = %s\n", addbuf);
      log_printf(" directory = %s\n", dirbuf);
      log_printf("       max = %d\n", max);
    }
  }
 
  if((tmp = fo_RepValidate(scheduler->sysconfig)) != NULL)
  {
    ERROR("configuration file failed repository validation");
    ERROR("The offending line: \"%s\"", tmp);
    g_free(tmp);
    exit(254);
  }
 
  /* load the version information */
  tmp = g_strdup_printf("%s/VERSION", scheduler->sysconfigdir);
  version = fo_config_load(tmp, &error);
  if(error) FATAL("%s", error->message);
  g_free(tmp);
 
  fo_config_join(scheduler->sysconfig, version, NULL);
 
  /* read COMMIT_HASH so we can detect a rebuilt binary on the next SIGHUP */
  {
    GError* ver_err = NULL;
    const gchar* new_ver = fo_config_get(version, "BUILD", "COMMIT_HASH", &ver_err);
    if (ver_err)
    {
      g_clear_error(&ver_err);
      new_ver = "unknown";
    }
    if (scheduler->scheduler_version == NULL)
    {
      /* first load: just store it */
      scheduler->scheduler_version = g_strdup(new_ver);
      log_printf("NOTE: scheduler version initialised to \"%s\"\n",
          scheduler->scheduler_version);
    }
    else if (strcmp(scheduler->scheduler_version, new_ver) != 0)
    {
      /* version changed: store it and respawn agents on the new binary */
      log_printf("NOTE: scheduler version changed: \"%s\" -> \"%s\"\n",
          scheduler->scheduler_version, new_ver);
      g_free(scheduler->scheduler_version);
      scheduler->scheduler_version = g_strdup(new_ver);
      event_signal(scheduler_version_refresh, NULL);
    }
  }
 
  fo_config_free(version);
 
  /* This will create the load and the print command for the special
   * configuration variables. This uses the l_op operation to load the variable
   * from the file and the w_op variable to write the variable to the log file.
   *
   * example:
   *   if this is in the CONF_VARIABLES_TYPES():
   *
   *     apply(char*, test_variable, NOOP, %s, "hello")
   *
   *   this is generated:
   *
   *     if(fo_config_has_key(sysconfig, "SCHEDULER", "test_variable")
   *       CONF_test_variable = fo_config_get(sysconfig, "SCHEDULER",
   *           "test_variable", NULL);
   *     V_SPECIAL("CONFIG: %s == %s\n", "test_variable", CONF_test_variable);
   *
   */
#define SELECT_CONF_INIT(type, name, l_op, w_op, val)                                  \
  if(fo_config_has_key(scheduler->sysconfig, "SCHEDULER", #name))                      \
    CONF_##name = l_op(fo_config_get(scheduler->sysconfig, "SCHEDULER", #name, NULL)); \
  V_SPECIAL("CONFIG: %s == " MK_STRING_LIT(w_op) "\n", #name, CONF_##name );
  CONF_VARIABLES_TYPES(SELECT_CONF_INIT)
#undef SELECT_CONF_INIT
}
 
int scheduler_daemonize(scheduler_t* scheduler)
{
  int ret = 0;
 
  /* daemonize the process */
  if((ret = daemon(0, 0)) != 0)
    return ret;
 
  scheduler->s_pid = getpid();
  return ret;
}
 
void scheduler_config_event(scheduler_t* scheduler, void* unused)
{
  if(scheduler->sysconfig)
    scheduler_clear_config(scheduler);
 
  scheduler_foss_config(scheduler);
  scheduler_agent_config(scheduler);
 
  database_init(scheduler);
  email_init(scheduler);
}
 
void scheduler_close_event(scheduler_t* scheduler, void* killed)
{
  closing = 1;
 
  if(killed) {
    kill_agents(scheduler);
  }
}
 
void scheduler_test_agents(scheduler_t* scheduler, void* unused)
{
  scheduler->s_startup = TRUE;
  test_agents(scheduler);
}
 
/* ************************************************************************** */
/* **** Version refresh ***************************************************** */
/* ************************************************************************** */
 
typedef struct
{
  scheduler_t* scheduler;
  int          count;
} version_refresh_ctx;
 
static gboolean version_refresh_kill_agent(int* pid_ptr, agent_t* agent,
    version_refresh_ctx* ctx)
{
  if (agent == NULL)
  {
    return FALSE;
  }
 
  agent_meta_version_reset(agent->type);
 
  /* leave agents that have already started; only respawn the not-yet-working ones */
  if (agent->status != AG_SPAWNED)
  {
    return FALSE;
  }
 
  /* kill() directly, not agent_kill(): return_code=0 makes the death event
   * re-queue the still-CHECKEDOUT job. Kill the process group too. */
  agent->return_code = 0;
  log_printf("NOTE: version refresh: sending SIGKILL to agent pid %d type \"%s\"\n",
      agent->pid, agent->type ? agent->type->name : "?");
  kill(-agent->pid, SIGKILL);
  ctx->count++;
  return FALSE;
}
 
void scheduler_version_refresh(scheduler_t* scheduler, void* unused)
{
  version_refresh_ctx ctx;
  ctx.scheduler = scheduler;
  ctx.count = 0;
 
  log_printf("NOTE: scheduler_version_refresh: killing all running agents "
      "for version update\n");
 
  g_tree_foreach(scheduler->agents,
      (GTraverseFunc)version_refresh_kill_agent, &ctx);
 
  log_printf("NOTE: scheduler_version_refresh: sent SIGKILL to %d agent(s); "
      "they will respawn automatically\n", ctx.count);
}
 
gint string_is_num(gchar* str)
{
  int len = strlen(str);
  int i;
 
  for(i = 0; i < len; i++)
    if(!isdigit(str[i]))
      return FALSE;
  return TRUE;
}
 
gint string_compare(gconstpointer a, gconstpointer b, gpointer user_data)
{
  return strcmp((char*)a, (char*)b);
}
 
gint int_compare(gconstpointer a, gconstpointer b, gpointer user_data)
{
  return *(int*)a - *(int*)b;
}