job.c

Go to the documentation of this file.

/*
 SPDX-FileCopyrightText: © 2010, 2011, 2012 Hewlett-Packard Development Company, L.P.
 SPDX-FileCopyrightText: © 2015 Siemens AG
 
 SPDX-License-Identifier: GPL-2.0-only
*/
/* local includes */
#include <libfossrepo.h>
#include <agent.h>
#include <database.h>
#include <job.h>
#include <scheduler.h>
 
/* std library includes */
#include <stdlib.h>
 
/* unix library includes */
#include <sys/types.h>
#include <unistd.h>
#include <sys/time.h>
#include <sys/resource.h>
 
/* other library includes */
#include <glib.h>
#include <gio/gio.h>
 
#define TEST_NULV(j) if(!j) { errno = EINVAL; ERROR("job passed is NULL, cannot proceed"); return; }
#define TEST_NULL(j, ret) if(!j) { errno = EINVAL; ERROR("job passed is NULL, cannot proceed"); return ret; }
#define MAX_SQL 512;JOB_STATUS_TYPES
 
/* ************************************************************************** */
/* **** Locals ************************************************************** */
/* ************************************************************************** */
 
#define SELECT_STRING(passed) MK_STRING_LIT(JOB_##passed),
const char* job_status_strings[] = { JOB_STATUS_TYPES(SELECT_STRING) };
#undef SELECT_STRING
 
static int is_active(int* job_id, job_t* job, int* counter)
{
  if((job->running_agents != NULL || job->failed_agents != NULL) || job->id < 0)
    ++(*counter);
  return 0;
}
 
static int job_sstatus(int* job_id, job_t* job, GOutputStream* ostr)
{
  gchar* status_str = g_strdup_printf(
      "job:%d status:%s type:%s, priority:%d running:%d finished:%d failed:%d\n",
      job->id,
      job_status_strings[job->status],
      job->agent_type,
      job->priority,
      g_list_length(job->running_agents),
      g_list_length(job->finished_agents),
      g_list_length(job->failed_agents));
 
  V_JOB("JOB_STATUS: %s", status_str);
  g_output_stream_write(ostr, status_str, strlen(status_str), NULL, NULL);
 
  if(*job_id == 0)
    g_list_foreach(job->running_agents, (GFunc)agent_print_status, ostr);
 
  g_free(status_str);
  return 0;
}
 
static void job_transition(scheduler_t* scheduler, job_t* job, job_status new_status)
{
  /* book keeping */
  TEST_NULV(job);
  V_JOB("JOB[%d]: job status changed: %s => %s\n",
      job->id, job_status_strings[job->status], job_status_strings[new_status]);
 
  /* change the job status */
  job->status = new_status;
 
  /* only update database for real jobs */
  if(job->id >= 0)
    database_update_job(scheduler, job, new_status);
}
 
gint job_compare(gconstpointer a, gconstpointer b, gpointer user_data)
{
  return ((job_t*)a)->priority - ((job_t*)b)->priority;
}
 
/* ************************************************************************** */
/* **** Constructor Destructor ********************************************** */
/* ************************************************************************** */
 
job_t* job_init(GTree* job_list, GSequence* job_queue,
    char* type, char* host, int id, int parent_id, int user_id, int group_id, int priority, char *jq_cmd_args)
{
  job_t* job = g_new0(job_t, 1);
 
  job->agent_type      = g_strdup(type);
  job->required_host   = g_strdup(host);
  job->running_agents  = NULL;
  job->finished_agents = NULL;
  job->failed_agents   = NULL;
  job->log             = NULL;
  job->status          = JB_CHECKEDOUT;
  job->data            = NULL;
  job->db_result       = NULL;
  job->lock            = NULL;
  job->idx             = 0;
  job->message         = NULL;
  job->priority        = priority;
  job->verbose         = 0;
  job->parent_id       = parent_id;
  job->id              = id;
  job->user_id         = user_id;
  job->group_id        = group_id;
  job->jq_cmd_args     = g_strdup(jq_cmd_args);
  job->checkedout_at   = time(NULL);
 
  g_tree_insert(job_list, &job->id, job);
  if(id >= 0) g_sequence_insert_sorted(job_queue, job, job_compare, NULL);
  return job;
}
 
void job_destroy(job_t* job)
{
  TEST_NULV(job);
 
  if(job->db_result != NULL)
  {
    SafePQclear(job->db_result);
 
    /* lock may be NULL if job_set_data(sql=1) was never called. */
    if(job->lock != NULL)
    {
      g_mutex_lock(job->lock);
      g_mutex_unlock(job->lock);
#if GLIB_CHECK_VERSION(2, 32, 0)
      g_mutex_clear(job->lock);
#else
      g_mutex_free(job->lock);
#endif
      g_free(job->lock);
      job->lock = NULL;
    }
  }
 
  if(job->log)
    log_destroy(job->log);
 
  g_list_free(job->running_agents);
  g_list_free(job->finished_agents);
  g_list_free(job->failed_agents);
  g_free(job->message);
  g_free(job->agent_type);
  g_free(job->required_host);
  g_free(job->data);
  if (job->jq_cmd_args) g_free(job->jq_cmd_args);
  g_free(job);
}
 
/* ************************************************************************** */
/* **** Events ************************************************************** */
/* ************************************************************************** */
 
void job_verbose_event(scheduler_t* scheduler, job_t* job)
{
  GList* iter;
 
  TEST_NULV(job);
  for(iter = job->running_agents; iter != NULL; iter = iter->next)
    aprintf(iter->data, "VERBOSE %d\n", job->verbose);
}
 
void job_status_event(scheduler_t* scheduler, arg_int* params)
{
  const char end[] = "end\n";
  GError* error = NULL;
 
  int tmp = 0;
  char buf[1024];
 
  if(!params->second)
  {
    memset(buf, '\0', sizeof(buf));
    sprintf(buf, "scheduler:%d revision:%s daemon:%d jobs:%d log:%s port:%d verbose:%d\n",
        scheduler->s_pid, fo_config_get(scheduler->sysconfig, "BUILD", "COMMIT_HASH", &error),
        scheduler->s_daemon, g_tree_nnodes(scheduler->job_list), main_log->log_name,
        scheduler->i_port, verbose);
 
    g_output_stream_write(params->first, buf, strlen(buf), NULL, NULL);
    g_tree_foreach(scheduler->job_list, (GTraverseFunc)job_sstatus, params->first);
  }
  else
  {
    job_t* stat = g_tree_lookup(scheduler->job_list, &params->second);
    if(stat)
    {
      job_sstatus(&tmp, g_tree_lookup(scheduler->job_list, &params->second), params->first);
    }
    else
    {
      sprintf(buf, "ERROR: invalid job id = %d\n", params->second);
      g_output_stream_write(params->first, buf, strlen(buf), NULL, NULL);
    }
  }
 
  g_output_stream_write(params->first, end, sizeof(end), NULL, NULL);
  g_free(params);
}
 
void job_pause_event(scheduler_t* scheduler, arg_int* params)
{
  job_t  tmp_job;
  job_t* job = params->first;
  GList* iter;
 
  // if the job doesn't exist, create a fake
  if(params->first == NULL)
  {
    tmp_job.id             = params->second;
    tmp_job.status         = JB_NOT_AVAILABLE;
    tmp_job.running_agents = NULL;
    tmp_job.message        = NULL;
 
    job = &tmp_job;
  }
 
  job_transition(scheduler, job, JB_PAUSED);
  for(iter = job->running_agents; iter != NULL; iter = iter->next)
    agent_pause(iter->data);
 
  g_free(params);
}
 
void job_restart_event(scheduler_t* scheduler, arg_int* params)
{
  job_t  tmp_job;
  job_t* job = params->first;
  GList* iter;
 
  // if the job doesn't exist, create a fake
  if(job == NULL)
  {
    tmp_job.id             = params->second;
    tmp_job.status         = JB_PAUSED;
    tmp_job.running_agents = NULL;
    tmp_job.message        = NULL;
 
    event_signal(database_update_event, NULL);
    job = &tmp_job;
  }
 
  if(job->status != JB_PAUSED)
  {
    ERROR("attempt to restart job %d failed, job status was %s",
        job->id, job_status_strings[job->status]);
    g_free(params);
    return;
  }
 
  for(iter = job->running_agents; iter != NULL; iter = iter->next)
  {
    if(job->db_result != NULL) agent_write(iter->data, "OK\n", 3);
    agent_unpause(iter->data);
  }
 
  job_transition(scheduler, job, JB_RESTART);
  g_free(params);
}
 
void job_priority_event(scheduler_t* scheduler, arg_int* params)
{
  GList* iter;
  job_t* job = (job_t*)params->first;
 
  if(!job)
  {
    ERROR("job_priority_event: NULL job passed, ignoring");
    g_free(params);
    return;
  }
 
  database_job_priority(scheduler, job, params->second);
  job->priority = params->second;
 
  /* re-sort in the queue. g_sequence_search() is unsafe once the key changed,
   * so find the job by pointer and let g_sequence_sort_changed() move it */
  {
    GSequenceIter* it = g_sequence_get_begin_iter(scheduler->job_queue);
    while(!g_sequence_iter_is_end(it))
    {
      if(g_sequence_get(it) == job)
      {
        g_sequence_sort_changed(it, job_compare, NULL);
        break;
      }
      it = g_sequence_iter_next(it);
    }
  }
 
  for(iter = job->running_agents; iter; iter = iter->next)
    setpriority(PRIO_PROCESS, ((agent_t*)iter->data)->pid, params->second);
 
  g_free(params);
}
 
void job_fail_event(scheduler_t* scheduler, job_t* job)
{
  GList* iter;
 
  if(job->status != JB_FAILED)
  {
    job_transition(scheduler, job, JB_FAILED);
  }
 
  for(iter = job->running_agents; iter != NULL; iter = iter->next)
  {
    V_JOB("JOB[%d]: job failed, killing agent\n", job->id);
    agent_kill(iter->data);
  }
 
  /* Also kill the finished (AG_PAUSED) agents, which still hold a process slot
   * and pipes until the watchdog would fire. They were already accounted at the
   * PAUSED transition, so kill() directly (return_code=0) rather than agent_kill()
   * to avoid a double-decrement. */
  for(iter = job->finished_agents; iter != NULL; iter = iter->next)
  {
    agent_t* a = (agent_t*)iter->data;
    V_JOB("JOB[%d]: job failed, killing finished agent pid %d\n", job->id, a->pid);
    a->return_code = 0;
    kill(-a->pid, SIGKILL);  /* process group, so child processes die too */
  }
}
 
/* ************************************************************************** */
/* **** Functions *********************************************************** */
/* ************************************************************************** */
 
void job_add_agent(job_t* job, void* agent)
{
  TEST_NULV(job);
  TEST_NULV(agent);
  job->running_agents = g_list_append(job->running_agents, agent);
}
 
void job_remove_agent(job_t* job, GTree* job_list, void* agent)
{
  GList* curr;
  TEST_NULV(job);
 
  if(job->finished_agents && agent)
    job->finished_agents = g_list_remove(job->finished_agents, agent);
 
  if(job->finished_agents == NULL && job->running_agents == NULL &&
     (job->status == JB_COMPLETE || job->status == JB_FAILED))
  {
    V_JOB("JOB[%d]: job removed from system\n", job->id);
 
    for(curr = job->failed_agents; curr != NULL; curr = curr->next)
      ((agent_t*)curr->data)->owner = NULL;
 
    g_tree_remove(job_list, &job->id);
  }
}
 
void job_finish_agent(job_t* job, void* agent)
{
  TEST_NULV(job);
  TEST_NULV(agent);
 
  job->running_agents  = g_list_remove(job->running_agents,  agent);
  job->finished_agents = g_list_append(job->finished_agents, agent);
}
 
void job_fail_agent(job_t* job, void* agent)
{
  TEST_NULV(job);
  TEST_NULV(agent);
  job->running_agents  = g_list_remove(job->running_agents,  agent);
  job->failed_agents   = g_list_append(job->failed_agents,   agent);
}
 
void job_set_data(scheduler_t* scheduler, job_t* job, char* data, int sql)
{
  g_free(job->data);
  job->data = g_strdup(data);
  job->idx = 0;
 
  if(sql)
  {
    // TODO
    //j->db_result = PQexec(db_conn, j->data);
    //j->lock = g_mutex_new();
  }
}
 
void job_update(scheduler_t* scheduler, job_t* job)
{
  GList* iter;
  int finished = 1;
 
  TEST_NULV(job)
 
  for(iter = job->running_agents; iter != NULL; iter = iter->next)
    if(((agent_t*)iter->data)->status != AG_PAUSED)
      finished = 0;
 
  if(job->status != JB_PAUSED && job->status != JB_COMPLETE && finished)
  {
    if(job->failed_agents == NULL)
    {
      /* don't mark an already-failed job complete: a finished agent killed by
       * job_fail_event() can reach here after the job is already JB_FAILED */
      if(job->status != JB_FAILED)
      {
        job_transition(scheduler, job, JB_COMPLETE);
        for(iter = job->finished_agents; iter != NULL; iter = iter->next)
        {
          aprintf(iter->data, "CLOSE\n");
        }
      }
    }
    /* this indicates a failed agent */
    else
    {
      /* free the failed-agent list before failing the job, and skip if it is
       * already failed */
      if(job->status != JB_FAILED)
      {
        g_list_free(job->failed_agents);
        job->failed_agents = NULL;
        g_free(job->message);
        job->message = NULL;
        job_fail_event(scheduler, job);
      }
    }
  }
}
 
int job_is_open(scheduler_t* scheduler, job_t* job)
{
  /* local */
  int retval = 0;
 
  TEST_NULL(job, -1);
 
  /* check to make sure that the job status is correct */
  if(job->status == JB_CHECKEDOUT)
    job_transition(scheduler, job, JB_STARTED);
 
  /* check to see if we even need to worry about sql stuff */
  if(job->db_result == NULL)
    return (job->idx == 0 && job->data != NULL);
 
  g_mutex_lock(job->lock);
  if(job->idx < PQntuples(job->db_result))
  {
    retval = 1;
  }
  else
  {
    SafePQclear(job->db_result);
    job->db_result = database_exec(scheduler, job->data);
    job->idx = 0;
 
    /* database_exec() can return NULL (lost connection, failed query). Guard so
     * PQntuples() is never called on NULL, which would crash the scheduler. */
    if(job->db_result == NULL)
    {
      g_mutex_unlock(job->lock);
      ERROR("JOB[%d]: database_exec returned NULL, marking job as failed", job->id);
      return -1;
    }
 
    retval = PQntuples(job->db_result) != 0;
  }
 
  g_mutex_unlock(job->lock);
  return retval;
}
 
char* job_next(job_t* job)
{
  char* retval = NULL;
 
  TEST_NULL(job, NULL);
  if(job->db_result == NULL)
  {
    job->idx = 1;
    return job->data;
  }
 
  g_mutex_lock(job->lock);
 
  if(job->idx < PQntuples(job->db_result))
    retval = PQgetvalue(job->db_result, job->idx++, 0);
 
  g_mutex_unlock(job->lock);
  return retval;
}
 
log_t* job_log(job_t* job)
{
  FILE*  file;
  gchar* file_name;
  gchar* file_path;
 
  if(job->id < 0)
    return main_log;
 
  if(job->log)
    return job->log;
 
  file_name = g_strdup_printf("%06d", job->id);
  file_path = fo_RepMkPath("logs", file_name);
 
  if((file = fo_RepFwrite("logs", file_name)) == NULL)
  {
    ERROR("JOB[%d]: job unable to create log file: %s\n", job->id, file_path);
    g_free(file_name);
    free(file_path);
    return NULL;
  }
 
  V_JOB("JOB[%d]: job created log file:\n    %s\n", job->id, file_path);
  database_job_log(job->id, file_path);
  job->log = log_new_FILE(file, file_name, job->agent_type, 0);
 
  g_free(file_name);
  free(file_path);
  return job->log;
}
 
/* ************************************************************************** */
/* **** Job list Functions ************************************************** */
/* ************************************************************************** */
 
job_t* next_job(GSequence* job_queue)
{
  job_t* retval = NULL;
  GSequenceIter* beg = g_sequence_get_begin_iter(job_queue);
 
  if(g_sequence_get_length(job_queue) != 0)
  {
    retval = g_sequence_get(beg);
    g_sequence_remove(beg);
  }
 
  return retval;
}
 
job_t* peek_job(GSequence* job_queue)
{
  GSequenceIter* beg;
 
  if(g_sequence_get_length(job_queue) == 0)
  {
    return NULL;
  }
 
  beg = g_sequence_get_begin_iter(job_queue);
  return g_sequence_get(beg);
}
 
uint32_t active_jobs(GTree* job_list)
{
  int count = 0;
  g_tree_foreach(job_list, (GTraverseFunc)is_active, &count);
  return count;
}