/*
    This file is part of darktable,
    Copyright (C) 2009-2025 darktable developers.

    darktable 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 3 of the License, or
    (at your option) any later version.

    darktable 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 darktable.  If not, see <http://www.gnu.org/licenses/>.
*/

#include "control/jobs.h"
#include "control/control.h"

#define DT_CONTROL_FG_PRIORITY 4
#define DT_CONTROL_MAX_JOBS 30

/* the queue can have scheduled jobs but all
    the workers are sleeping, so this kicks the workers
    on timed interval.
*/
typedef struct worker_thread_parameters_t
{
  dt_control_t *self;
  int32_t threadid;
} worker_thread_parameters_t;

typedef struct _dt_job_t
{
  dt_job_execute_callback execute;
  void *params;
  size_t params_size;
  dt_job_destroy_callback params_destroy;
  int32_t result;

  dt_pthread_mutex_t state_mutex;
  dt_pthread_mutex_t wait_mutex;

  dt_job_state_t state;
  unsigned char priority;
  dt_job_queue_t queue;

  dt_job_state_change_callback state_changed_cb;

  dt_progress_t *progress;

  char description[DT_CONTROL_DESCRIPTION_LEN];
  dt_view_type_flags_t view_creator;
  gboolean is_synchronous;
} _dt_job_t;

/** check if two jobs are to be considered equal. a simple memcmp won't work since the mutexes probably won't
   match
    we don't want to compare result, priority or state since these will change during the course of
   processing.
    NOTE: maybe allow to pass a comparator for params.
 */
static inline gboolean _control_job_equal(_dt_job_t *j1, _dt_job_t *j2)
{
  if(!j1 || !j2) return FALSE;
  if(j1->params_size != 0
      && j1->params_size == j2->params_size)
    return (j1->execute == j2->execute
            && j1->state_changed_cb == j2->state_changed_cb
            && j1->queue == j2->queue
            && (memcmp(j1->params, j2->params, j1->params_size) == 0));
  return (j1->execute == j2->execute
          && j1->state_changed_cb == j2->state_changed_cb
          && j1->queue == j2->queue
          && (g_strcmp0(j1->description, j2->description) == 0));
}

static void _control_job_set_state(_dt_job_t *job,
                                    dt_job_state_t state)
{
  if(!job) return;
  dt_pthread_mutex_lock(&job->state_mutex);
  if(state >= DT_JOB_STATE_FINISHED  && job->state != DT_JOB_STATE_RUNNING && job->progress)
  {
    dt_control_progress_destroy(job->progress);
    job->progress = NULL;
  }
  job->state = state;
  /* pass state change to callback */
  if(job->state_changed_cb) job->state_changed_cb(job, state);
  dt_pthread_mutex_unlock(&job->state_mutex);
}

dt_job_state_t dt_control_job_get_state(_dt_job_t *job)
{
  if(!job) return DT_JOB_STATE_DISPOSED;
  dt_pthread_mutex_lock(&job->state_mutex);
  dt_job_state_t state = job->state;
  dt_pthread_mutex_unlock(&job->state_mutex);
  return state;
}

void dt_control_job_set_params(_dt_job_t *job, void *params, dt_job_destroy_callback callback)
{
  if(!job || dt_control_job_get_state(job) != DT_JOB_STATE_INITIALIZED)
    return;
  job->params = params;
  job->params_size = 0;
  job->params_destroy = callback;
}

void dt_control_job_set_params_with_size(dt_job_t *job,
                                        void *params,
                                        size_t params_size,
                                        dt_job_destroy_callback callback)
{
  if(!job || dt_control_job_get_state(job) != DT_JOB_STATE_INITIALIZED)
    return;
  job->params = params;
  job->params_size = params_size;
  job->params_destroy = callback;
}

void *dt_control_job_get_params(const _dt_job_t *job)
{
  if(!job) return NULL;
  return job->params;
}

dt_job_t *dt_control_job_create(dt_job_execute_callback execute,
                                const char *msg, ...)
{
  if(!dt_control_running()) return NULL;
  _dt_job_t *job = calloc(1, sizeof(_dt_job_t));
  if(!job) return NULL;

  va_list ap;
  va_start(ap, msg);
  vsnprintf(job->description, DT_CONTROL_DESCRIPTION_LEN, msg, ap);
  va_end(ap);

  job->execute = execute;
  job->state = DT_JOB_STATE_INITIALIZED;
  job->view_creator = dt_view_get_current();

  dt_pthread_mutex_init(&job->state_mutex);
  dt_pthread_mutex_init(&job->wait_mutex);
  return job;
}

dt_view_type_flags_t dt_control_job_get_view_creator(const dt_job_t *job)
{
  return job && job->view_creator;
}

gboolean dt_control_job_is_synchronous(const dt_job_t *job)
{
  return job && job->is_synchronous;
}

void dt_control_job_dispose(_dt_job_t *job)
{
  if(!job) return;
  if(job->progress)
    dt_control_progress_destroy(job->progress);
  job->progress = NULL;
  _control_job_set_state(job, DT_JOB_STATE_DISPOSED);
  if(job->params_destroy)
    job->params_destroy(job->params);
  dt_pthread_mutex_destroy(&job->state_mutex);
  dt_pthread_mutex_destroy(&job->wait_mutex);
  free(job);
}

static inline const char *_queuename(const dt_job_queue_t id)
{
  switch(id)
  {
    case DT_JOB_QUEUE_USER_FG:      return "DT_JOB_QUEUE_USER_FG";
    case DT_JOB_QUEUE_SYSTEM_FG:    return "DT_JOB_QUEUE_SYSTEM_FG";
    case DT_JOB_QUEUE_USER_BG:      return "DT_JOB_QUEUE_USER_BG";
    case DT_JOB_QUEUE_USER_EXPORT:  return "DT_JOB_QUEUE_USER_EXPORT";
    case DT_JOB_QUEUE_SYSTEM_BG:    return "DT_JOB_QUEUE_SYSTEM_BG";
    case DT_JOB_QUEUE_SYNCHRONOUS:  return "DT_JOB_QUEUE_SYNCHRONOUS";
    default:                        return "???";
  }
}
// We don't want to log dt_get_wtime() as we already show the stamp
static void _control_job_print(_dt_job_t *job, const char *info, const char *err, int32_t res)
{
  if(!job) return;
  dt_print(DT_DEBUG_CONTROL, "[%s]\t%02d %s %s | queue: %s | priority: %d",
    info, res, err, job->description, _queuename(job->queue), job->priority);
}


static __thread int32_t threadid = -1;
// As threadid is `per thread` we don't have to use atomics
static inline int32_t _control_get_threadid()
{
  return threadid > -1 ? threadid : darktable.control->num_threads;
}

static inline int32_t _control_get_threadid_res()
{
  return threadid > -1 ? threadid : DT_CTL_WORKER_RESERVED;
}

void dt_control_job_cancel(_dt_job_t *job)
{
  _control_job_set_state(job, DT_JOB_STATE_CANCELLED);
}

static gboolean _control_run_job_res(dt_control_t *control, int32_t res)
{
  if(((unsigned int)res) >= DT_CTL_WORKER_RESERVED)
    return TRUE;

  _dt_job_t *job = NULL;
  dt_pthread_mutex_lock(&control->res_mutex);
  if(control->new_res[res])
  {
    job = control->job_res[res];
    control->job_res[res] = NULL; // this job belongs to us now, the queue may not touch it any longer
  }
  control->new_res[res] = 0;
  dt_pthread_mutex_unlock(&control->res_mutex);
  if(!job) return TRUE;

  /* change state to running */
  dt_pthread_mutex_lock(&job->wait_mutex);
  if(dt_control_job_get_state(job) == DT_JOB_STATE_QUEUED)
  {
    _control_job_print(job, "run_job+", "", res);

    _control_job_set_state(job, DT_JOB_STATE_RUNNING);

    /* execute job */
    job->result = job->execute(job);

    _control_job_set_state(job, DT_JOB_STATE_FINISHED);
    _control_job_print(job, "run_job-", "", res);
  }
  dt_pthread_mutex_unlock(&job->wait_mutex);
  dt_control_job_dispose(job);
  return FALSE;
}

static _dt_job_t *_control_schedule_job(dt_control_t *control)
{
  /*
   * job scheduling works like this:
   * - when there is a single job in the queue head with a maximal priority -> pick it
   * - otherwise pick among the ones with the maximal priority in the following order:
   *   * user foreground
   *   * system foreground
   *   * user background
   *   * system background
   * - the jobs that didn't get picked this round get their priority incremented
   */

  dt_pthread_mutex_lock(&control->queue_mutex);

  // find the job
  _dt_job_t *job = NULL;
  int winner_queue = DT_JOB_QUEUE_MAX;
  int max_priority = -1;
  for(int i = 0; i < DT_JOB_QUEUE_MAX; i++)
  {
    if(control->queues[i] == NULL) continue;
    if(control->export_scheduled && i == DT_JOB_QUEUE_USER_EXPORT) continue;
    _dt_job_t *_job = (_dt_job_t *)control->queues[i]->data;
    if(_job->priority > max_priority)
    {
      max_priority = _job->priority;
      job = _job;
      winner_queue = i;
    }
  }

  if(!job)
  {
    dt_pthread_mutex_unlock(&control->queue_mutex);
    return NULL;
  }

  // the order of the queues in control->queues matches our priority, and we only update job when the priority
  // is strictly bigger
  // invariant -> job is the one we are looking for

  // remove the to be scheduled job from its queue
  GList **queue = &control->queues[winner_queue];
  *queue = g_list_delete_link(*queue, *queue);
  control->queue_length[winner_queue]--;
  if(winner_queue == DT_JOB_QUEUE_USER_EXPORT) control->export_scheduled = TRUE;

  // and place it in scheduled job array (for job deduping)
  control->job[_control_get_threadid()] = job;

  // increment the priorities of the others
  for(int i = 0; i < DT_JOB_QUEUE_MAX; i++)
  {
    if(i == winner_queue || control->queues[i] == NULL) continue;
    ((_dt_job_t *)control->queues[i]->data)->priority++;
  }

  dt_pthread_mutex_unlock(&control->queue_mutex);

  return job;
}

static void _control_job_execute(_dt_job_t *job)
{
  _control_job_print(job, "run_job+", "", DT_CTL_WORKER_RESERVED + _control_get_threadid());

  _control_job_set_state(job, DT_JOB_STATE_RUNNING);

  /* execute job */
  job->result = job->execute(job);

  _control_job_set_state(job, DT_JOB_STATE_FINISHED);
  _control_job_print(job, "run_job-", "", DT_CTL_WORKER_RESERVED + _control_get_threadid());
}

static gboolean _control_run_job(dt_control_t *control)
{
  _dt_job_t *job = _control_schedule_job(control);

  if(!job) return TRUE;

  /* change state to running */
  dt_pthread_mutex_lock(&job->wait_mutex);
  if(dt_control_job_get_state(job) == DT_JOB_STATE_QUEUED)
    _control_job_execute(job);

  dt_pthread_mutex_unlock(&job->wait_mutex);

  // remove the job from scheduled job array (for job deduping)
  dt_pthread_mutex_lock(&control->queue_mutex);
  control->job[_control_get_threadid()] = NULL;
  if(job->queue == DT_JOB_QUEUE_USER_EXPORT) control->export_scheduled = FALSE;
  dt_pthread_mutex_unlock(&control->queue_mutex);

  // and free it
  dt_control_job_dispose(job);
  dt_atomic_sub_int(&control->pending_jobs, 1);

  return FALSE;
}

gboolean dt_control_add_job_res(_dt_job_t *job, const int32_t res)
{
  dt_control_t *control = darktable.control;
  if(((unsigned int)res) >= DT_CTL_WORKER_RESERVED || !job)
  {
    dt_control_job_dispose(job);
    return TRUE;
  }

  // TODO: pthread cancel and restart in tough cases?
  dt_pthread_mutex_lock(&control->res_mutex);

  // if there is a job in the queue we have to discard that first
  if(control->job_res[res])
  {
    _control_job_set_state(control->job_res[res], DT_JOB_STATE_DISCARDED);
    dt_control_job_dispose(control->job_res[res]);
  }

  _control_job_print(job, "add_job_res", "", res);

  _control_job_set_state(job, DT_JOB_STATE_QUEUED);
  control->job_res[res] = job;
  control->new_res[res] = 1;

  dt_pthread_mutex_unlock(&control->res_mutex);

  dt_pthread_mutex_lock(&control->cond_mutex);
  pthread_cond_broadcast(&control->cond);
  dt_pthread_mutex_unlock(&control->cond_mutex);

  return FALSE;
}

gboolean dt_control_add_job(dt_job_queue_t queue_id, _dt_job_t *job)
{
  dt_control_t *control = darktable.control;
  if((((unsigned int)queue_id) >= DT_JOB_QUEUE_MAX && queue_id != DT_JOB_QUEUE_SYNCHRONOUS) || !job)
  {
    dt_control_job_dispose(job);
    return TRUE;
  }

  if(!dt_control_running() || queue_id == DT_JOB_QUEUE_SYNCHRONOUS)
  {
    // whatever we are adding here won't be scheduled as the system isn't running. execute it synchronous instead.
    dt_pthread_mutex_lock(&job->wait_mutex); // is that even needed?
    job->is_synchronous = TRUE;
    _control_job_execute(job);
    dt_pthread_mutex_unlock(&job->wait_mutex);

    dt_control_job_dispose(job);
    return FALSE;
  }

  job->queue = queue_id;

  _dt_job_t *job_for_disposal = NULL;

  dt_pthread_mutex_lock(&control->queue_mutex);

  GList **queue = &control->queues[queue_id];
  size_t length = control->queue_length[queue_id];

  _control_job_print(job, "add_job", "", (int32_t)length);

  dt_atomic_add_int(&control->pending_jobs, 1);
  if(queue_id == DT_JOB_QUEUE_SYSTEM_FG)
  {
    // this is a stack with limited size and bubble up and all that stuff
    job->priority = DT_CONTROL_FG_PRIORITY;

    // check if we have already scheduled the job
    for(int k = 0; k < control->num_threads; k++)
    {
      _dt_job_t *other_job = control->job[k];
      if(_control_job_equal(job, other_job))
      {
        _control_job_print(other_job, "add_job", "found job already in scheduled:", -1);

        dt_pthread_mutex_unlock(&control->queue_mutex);

        _control_job_set_state(job, DT_JOB_STATE_DISCARDED);
        dt_control_job_dispose(job);
        dt_atomic_sub_int(&control->pending_jobs, 1);

        return 0; // there can't be any further copy
      }
    }

    // if the job is already in the queue -> move it to the top
    for(GList *iter = *queue; iter; iter = g_list_next(iter))
    {
      _dt_job_t *other_job = iter->data;
      if(_control_job_equal(job, other_job))
      {
        _control_job_print(other_job, "add_job", "found job already in queue", -1);

        *queue = g_list_delete_link(*queue, iter);
        length--;
        dt_atomic_sub_int(&control->pending_jobs, 1);

        job_for_disposal = job;

        job = other_job;
        break; // there can't be any further copy in the list
      }
    }

    // now we can add the new job to the list
    *queue = g_list_prepend(*queue, job);
    length++;

    // and take care of the maximal queue size
    if(length > DT_CONTROL_MAX_JOBS)
    {
      GList *last = g_list_last(*queue);
      _control_job_set_state(last->data, DT_JOB_STATE_DISCARDED);
      dt_control_job_dispose(last->data);
      *queue = g_list_delete_link(*queue, last);
      length--;
      dt_atomic_sub_int(&control->pending_jobs, 1);
    }

    control->queue_length[queue_id] = length;
  }
  else
  {
    // the rest are FIFOs
    if(queue_id == DT_JOB_QUEUE_USER_BG ||
       queue_id == DT_JOB_QUEUE_USER_EXPORT ||
       queue_id == DT_JOB_QUEUE_SYSTEM_BG)
      job->priority = 0;
    else
      job->priority = DT_CONTROL_FG_PRIORITY;
    *queue = g_list_append(*queue, job);
    control->queue_length[queue_id]++;
  }
  _control_job_set_state(job, DT_JOB_STATE_QUEUED);
  dt_pthread_mutex_unlock(&control->queue_mutex);

  // notify workers
  dt_pthread_mutex_lock(&control->cond_mutex);
  pthread_cond_broadcast(&control->cond);
  dt_pthread_mutex_unlock(&control->cond_mutex);

  // dispose of dropped job, if any
  _control_job_set_state(job_for_disposal, DT_JOB_STATE_DISCARDED);
  dt_control_job_dispose(job_for_disposal);

  return FALSE;
}

static void *_control_work_res(void *ptr)
{
  worker_thread_parameters_t *params = (worker_thread_parameters_t *)ptr;
  dt_control_t *s = params->self;
  threadid = params->threadid;
  char name[16] = {0};
  snprintf(name, sizeof(name), "worker res %d", threadid);
  dt_pthread_setname(name);
  free(params);
  const int32_t threadid_res = _control_get_threadid_res();
  while(dt_control_running())
  {
    // dt_print(DT_DEBUG_CONTROL, "[control_work] %d", threadid_res);
    if(_control_run_job_res(s, threadid_res))
    {
      // wait for a new job.
      int old;
      pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old);
      dt_pthread_mutex_lock(&s->cond_mutex);
      dt_pthread_cond_wait(&s->cond, &s->cond_mutex);
      dt_pthread_mutex_unlock(&s->cond_mutex);
      int tmp;
      pthread_setcancelstate(old, &tmp);
    }
  }
  return NULL;
}

static void *_control_worker_kicker(void *ptr)
{
  dt_control_t *control = (dt_control_t *)ptr;
  dt_pthread_setname("kicker");
  while(dt_control_running())
  {
    sleep(2);
    dt_pthread_mutex_lock(&control->cond_mutex);
    pthread_cond_broadcast(&control->cond);
    dt_pthread_mutex_unlock(&control->cond_mutex);
  }
  return NULL;
}

static void *_control_work(void *ptr)
{
  worker_thread_parameters_t *params = (worker_thread_parameters_t *)ptr;
  dt_control_t *control = params->self;
  threadid = params->threadid;
  char name[16] = {0};
  snprintf(name, sizeof(name), "worker %d", threadid);
  dt_pthread_setname(name);
  free(params);
  while(dt_control_running())
  {
    if(_control_run_job(control))
    {
      // wait for a new job.
      dt_pthread_mutex_lock(&control->cond_mutex);
      dt_pthread_cond_wait(&control->cond, &control->cond_mutex);
      dt_pthread_mutex_unlock(&control->cond_mutex);
    }
  }
  return NULL;
}

// convenience functions to have a progress bar for the job.
// this allows to show the gui indicator of the job even before it got scheduled
void dt_control_job_add_progress(dt_job_t *job, const char *message, const gboolean cancellable)
{
  if(!job) return;
  job->progress = dt_control_progress_create(TRUE, message);
  if(cancellable)
    dt_control_progress_attach_job(job->progress, job);
}

void dt_control_job_set_progress_message(dt_job_t *job, const char *message, ...)
{
  if(!job || !job->progress) return;
  va_list ap;
  va_start(ap, message);
  char *msg = g_strdup_vprintf(message, ap);
  dt_control_progress_set_message(job->progress, msg);
  g_free(msg);
  va_end(ap);
}

void dt_control_job_set_progress(dt_job_t *job, const double value)
{
  if(!job || !job->progress) return;
  dt_control_progress_set_progress(job->progress, value);
}

double dt_control_job_get_progress(const dt_job_t *job)
{
  if(!job || !job->progress) return -1.0;
  return dt_control_progress_get_progress(job->progress);
}


// moved out of control.c to be able to make some helper functions static
void dt_control_jobs_init()
{
  dt_control_t *control = darktable.control;
  // start threads
  control->num_threads = dt_worker_threads();
  control->thread = (pthread_t *)calloc(control->num_threads, sizeof(pthread_t));
  // allocate enough jobs to match _control_get_threadid()
  control->job = (dt_job_t **)calloc(control->num_threads+1, sizeof(dt_job_t *));

  dt_atomic_set_int(&control->running, DT_CONTROL_STATE_RUNNING);

  int err = 0; // collected errors while creating all threads

  for(int k = 0; k < control->num_threads; k++)
  {
    worker_thread_parameters_t *params = calloc(1, sizeof(worker_thread_parameters_t));
    params->self = control;
    params->threadid = k;
    err |= dt_pthread_create(&control->thread[k], _control_work, params);
  }

  /* create queue kicker thread */
  err |= dt_pthread_create(&control->kick_on_workers_thread, _control_worker_kicker, control);

  for(int k = 0; k < DT_CTL_WORKER_RESERVED; k++)
  {
    control->job_res[k] = NULL;
    control->new_res[k] = 0;
    worker_thread_parameters_t *params = calloc(1, sizeof(worker_thread_parameters_t));
    params->self = control;
    params->threadid = k;
    err |= dt_pthread_create(&control->thread_res[k], _control_work_res, params);
  }

  if(err != 0)
  {
    /* FIXME can we handle this better? */
    dt_print(DT_DEBUG_ALWAYS, "[dt_control_jobs_init] ERROR STARTING THREADS, PROBLEMS AHEAD");
  }
}

void dt_control_jobs_cleanup()
{
  dt_control_t *control = darktable.control;
  free(control->job);
  control->job = NULL;
  free(control->thread);
  control->thread = NULL;
}

int dt_control_jobs_pending()
{
  dt_control_t *control = darktable.control;
  if(!control) return 0;

  int pending = dt_atomic_get_int(&control->pending_jobs);
  if(darktable.backthumbs.state == DT_JOB_STATE_RUNNING)
    pending--;
  return pending;
}
// clang-format off
// modelines: These editor modelines have been set for all relevant files by tools/update_modelines.py
// vim: shiftwidth=2 expandtab tabstop=2 cindent
// kate: tab-indents: off; indent-width 2; replace-tabs on; indent-mode cstyle; remove-trailing-spaces modified;
// clang-format on