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