worker.c

   1 #include "worker.h"
   2 #include "locking.h"
   3 #include "list.h"
   4 #include "xmalloc.h"
   5 #include "debug.h"
   6
   7 #include <stdlib.h>
   8 #include <pthread.h>
   9
  10 struct worker_job {
  11         struct list_head node;
  12         /* >0, 0 is 'any' */
  13         int type;
  14         void (*job_cb)(void *data);
  15         void (*free_cb)(void *data);
  16         void *data;
  17 };
  18
  19 static LIST_HEAD(worker_job_head);
  20 static pthread_mutex_t worker_mutex = CMUS_MUTEX_INITIALIZER;
  21 static pthread_cond_t worker_cond = PTHREAD_COND_INITIALIZER;
  22 static pthread_t worker_thread;
  23 static int running = 1;
  24 static int cancel_type = JOB_TYPE_NONE;
  25
  26 /*
  27  * - only worker thread modifies this
  28  * - cur_job->job_cb can read this without locking
  29  * - anyone else must lock worker before reading this
  30  */
  31 static struct worker_job *cur_job = NULL;
  32
  33 #define worker_lock() cmus_mutex_lock(&worker_mutex)
  34 #define worker_unlock() cmus_mutex_unlock(&worker_mutex)
  35
  36 static void *worker_loop(void *arg)
  37 {
  38         worker_lock();
  39         while (1) {
  40                 if (list_empty(&worker_job_head)) {
  41                         int rc;
  42
  43                         if (!running)
  44                                 break;
  45
  46                         rc = pthread_cond_wait(&worker_cond, &worker_mutex);
  47                         if (rc)
  48                                 d_print("pthread_cond_wait: %s\n", strerror(rc));
  49                 } else {
  50                         struct list_head *item = worker_job_head.next;
  51                         uint64_t st, et;
  52
  53                         list_del(item);
  54                         cur_job = container_of(item, struct worker_job, node);
  55                         worker_unlock();
  56
  57                         timer_get(&st);
  58                         cur_job->job_cb(cur_job->data);
  59                         timer_get(&et);
  60                         timer_print("worker job", et - st);
  61
  62                         worker_lock();
  63                         cur_job->free_cb(cur_job->data);
  64                         free(cur_job);
  65                         cur_job = NULL;
  66
  67                         // wakeup worker_remove_jobs() if needed
  68                         if (cancel_type != JOB_TYPE_NONE)
  69                                 pthread_cond_signal(&worker_cond);
  70                 }
  71         }
  72         worker_unlock();
  73         return NULL;
  74 }
  75
  76 void worker_init(void)
  77 {
  78         int rc = pthread_create(&worker_thread, NULL, worker_loop, NULL);
  79
  80         BUG_ON(rc);
  81 }
  82
  83 void worker_exit(void)
  84 {
  85         worker_lock();
  86         running = 0;
  87         pthread_cond_signal(&worker_cond);
  88         worker_unlock();
  89
  90         pthread_join(worker_thread, NULL);
  91 }
  92
  93 void worker_add_job(int type, void (*job_cb)(void *data),
  94                 void (*free_cb)(void *data), void *data)
  95 {
  96         struct worker_job *job;
  97
  98         job = xnew(struct worker_job, 1);
  99         job->type = type;
 100         job->job_cb = job_cb;
 101         job->free_cb = free_cb;
 102         job->data = data;
 103
 104         worker_lock();
 105         list_add_tail(&job->node, &worker_job_head);
 106         pthread_cond_signal(&worker_cond);
 107         worker_unlock();
 108 }
 109
 110 void worker_remove_jobs(int type)
 111 {
 112         struct list_head *item;
 113
 114         worker_lock();
 115         cancel_type = type;
 116
 117         /* remove jobs of the specified type from the queue */
 118         item = worker_job_head.next;
 119         while (item != &worker_job_head) {
 120                 struct worker_job *job = container_of(item, struct worker_job, node);
 121                 struct list_head *next = item->next;
 122
 123                 if (type == JOB_TYPE_ANY || type == job->type) {
 124                         list_del(&job->node);
 125                         job->free_cb(job->data);
 126                         free(job);
 127                 }
 128                 item = next;
 129         }
 130
 131         /* wait current job to finish or cancel if it's of the specified type */
 132         if (cur_job && (type == JOB_TYPE_ANY || type == cur_job->type))
 133                 pthread_cond_wait(&worker_cond, &worker_mutex);
 134
 135         cancel_type = JOB_TYPE_NONE;
 136         worker_unlock();
 137 }
 138
 139 int worker_has_job(int type)
 140 {
 141         struct worker_job *job;
 142         int has_job = 0;
 143
 144         worker_lock();
 145         list_for_each_entry(job, &worker_job_head, node) {
 146                 if (type == JOB_TYPE_ANY || type == job->type) {
 147                         has_job = 1;
 148                         break;
 149                 }
 150         }
 151         if (cur_job && (type == JOB_TYPE_ANY || type == cur_job->type))
 152                 has_job = 1;
 153         worker_unlock();
 154         return has_job;
 155 }
 156
 157 /*
 158  * this is only called from the worker thread
 159  * cur_job is guaranteed to be non-NULL
 160  */
 161 int worker_cancelling(void)
 162 {
 163         return cur_job->type == cancel_type || cancel_type == JOB_TYPE_ANY;
 164 }