worker.c

   1 #include "worker.h"
   2 #include "locking.h"
   3 #include "list.h"
   4 #include "utils.h"
   5 #include "xmalloc.h"
   6 #include "debug.h"
   7
   8 #include <stdlib.h>
   9 #include <pthread.h>
  10
  11 struct worker_job {
  12         struct list_head node;
  13         /* >0, 0 is 'any' */
  14         int type;
  15         void (*job_cb)(void *data);
  16         void (*free_cb)(void *data);
  17         void *data;
  18 };
  19
  20 static LIST_HEAD(worker_job_head);
  21 static pthread_mutex_t worker_mutex = CMUS_MUTEX_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         while (1) {
  39                 worker_lock();
  40                 if (list_empty(&worker_job_head)) {
  41                         if (!running) {
  42                                 worker_unlock();
  43                                 return NULL;
  44                         }
  45                         worker_unlock();
  46                         ms_sleep(100);
  47                 } else {
  48                         struct list_head *item = worker_job_head.next;
  49                         uint64_t st, et;
  50
  51                         list_del(item);
  52                         cur_job = container_of(item, struct worker_job, node);
  53                         worker_unlock();
  54
  55                         timer_get(&st);
  56                         cur_job->job_cb(cur_job->data);
  57                         timer_get(&et);
  58                         timer_print("worker job", et - st);
  59
  60                         worker_lock();
  61                         cur_job->free_cb(cur_job->data);
  62                         free(cur_job);
  63                         cur_job = NULL;
  64                         worker_unlock();
  65                 }
  66         }
  67 }
  68
  69 void worker_init(void)
  70 {
  71         int rc = pthread_create(&worker_thread, NULL, worker_loop, NULL);
  72
  73         BUG_ON(rc);
  74 }
  75
  76 void worker_exit(void)
  77 {
  78         worker_lock();
  79         running = 0;
  80         worker_unlock();
  81
  82         pthread_join(worker_thread, NULL);
  83 }
  84
  85 void worker_add_job(int type, void (*job_cb)(void *data),
  86                 void (*free_cb)(void *data), void *data)
  87 {
  88         struct worker_job *job;
  89
  90         job = xnew(struct worker_job, 1);
  91         job->type = type;
  92         job->job_cb = job_cb;
  93         job->free_cb = free_cb;
  94         job->data = data;
  95
  96         worker_lock();
  97         list_add_tail(&job->node, &worker_job_head);
  98         worker_unlock();
  99 }
 100
 101 void worker_remove_jobs(int type)
 102 {
 103         struct list_head *item;
 104
 105         worker_lock();
 106         cancel_type = type;
 107
 108         /* remove jobs of the specified type from the queue */
 109         item = worker_job_head.next;
 110         while (item != &worker_job_head) {
 111                 struct worker_job *job = container_of(item, struct worker_job, node);
 112                 struct list_head *next = item->next;
 113
 114                 if (type == JOB_TYPE_ANY || type == job->type) {
 115                         list_del(&job->node);
 116                         job->free_cb(job->data);
 117                         free(job);
 118                 }
 119                 item = next;
 120         }
 121
 122         /* wait current job to finish or cancel if it's of the specified type */
 123         while (cur_job && (type == JOB_TYPE_ANY || type == cur_job->type)) {
 124                 worker_unlock();
 125                 ms_sleep(50);
 126                 worker_lock();
 127         }
 128
 129         cancel_type = JOB_TYPE_NONE;
 130         worker_unlock();
 131 }
 132
 133 /*
 134  * this is only called from the worker thread
 135  * cur_job is guaranteed to be non-NULL
 136  */
 137 int worker_cancelling(void)
 138 {
 139         return cur_job->type == cancel_type || cancel_type == JOB_TYPE_ANY;
 140 }